diff --git a/contracts/agents-api/index.md b/contracts/agents-api/index.md index d312e0b0e..cfe8cd77c 100644 --- a/contracts/agents-api/index.md +++ b/contracts/agents-api/index.md @@ -93,6 +93,7 @@ Each item is Core's deliberate or native behavior where the official service beh - An empty text part beside other text is accepted and stored. - Empty input returns 400 `invalid_request` with a generic message and a null param; the official response is `invalid_request_error` with param `input`. - Session usage is available as soon as every root Turn has settled; official reads lag by seconds. +- U+0000 in Harness output, such as command output, reads as U+FFFD; the official service keeps it. **Files, Skills, Environment files and Artifacts** ([Files and Skills](./source-files.md), [Environment files and Artifacts](./environment-files.md)) diff --git a/contracts/agents-api/sessions-events.md b/contracts/agents-api/sessions-events.md index ee6ba5692..03a3a12b6 100644 --- a/contracts/agents-api/sessions-events.md +++ b/contracts/agents-api/sessions-events.md @@ -141,6 +141,8 @@ Turn and Item lists take `after`, `limit` and `order` ([list rules](./wire-seman A failed tool does not fail its Turn. Tool output is readable by the Session's Project and can contain the tool's own diagnostic text. +PostgreSQL cannot store U+0000, so Harness output that contains it, such as a command's output, is stored, streamed and listed with U+FFFD in its place, and the Turn continues. + ## Usage Turn and Session `usage` uses the pinned `TokenUsage` fields: input, cached input, output, reasoning output and total tokens. Null means unknown, never zero. diff --git a/contracts/agents-api/zh/index.md b/contracts/agents-api/zh/index.md index 838d1ef7e..009710471 100644 --- a/contracts/agents-api/zh/index.md +++ b/contracts/agents-api/zh/index.md @@ -1,7 +1,7 @@ --- title: "OpenAI 兼容性" source: contracts/agents-api/index.md -source_hash: 5766fdb6de7cdf07f02de16e2fdaf3fea0ce4d3b357d5648f15e9d709483b819 +source_hash: 535fce4a958c4176e98e0b59a6995fd9d72c874cb4a6493b797545652c37321b --- Core 实现下方固定版本的 OpenAI Agents API。本台账列出各项资源的覆盖情况、相对于官方服务的所有差异以及已知缺口。 @@ -96,6 +96,7 @@ Core 自身字段位于 `x_agents_core` 中（[Core extensions](../../../docs/zh - 与其他文本并存的空文本部分可被接受并存储。 - 空输入会返回 400 `invalid_request`，附带通用消息和值为 null 的 param；官方响应为 `invalid_request_error`，param 为 `input`。 - 一旦所有根级 Turn 均已进入终态，Session 用量即可用；官方读取会滞后数秒。 +- Harness 输出（例如命令输出）中的 U+0000 读取时为 U+FFFD；官方服务会保留该字符。 **Files、Skills、Environment files 和 Artifacts**（[Files and Skills](source-files.md)、[Environment files and Artifacts](environment-files.md)） diff --git a/contracts/agents-api/zh/sessions-events.md b/contracts/agents-api/zh/sessions-events.md index 8a90c6dfd..5a401ccf5 100644 --- a/contracts/agents-api/zh/sessions-events.md +++ b/contracts/agents-api/zh/sessions-events.md @@ -1,7 +1,7 @@ --- title: "Session、事件和历史" source: contracts/agents-api/sessions-events.md -source_hash: a87569d0232231fec4eeefeae610626125162474b3362ca490d2938ca0b8a024 +source_hash: 581c5a36cd2df2fa35d96da7c0e4d8fcbc7e01f4d97d35a2351b39386b15b380 --- 本契约涵盖 Session 内部发生的事情：发送输入、实时事件流，以及读取 Turn、Item 和使用量的持久化历史。Session 资源本身（创建配置、重试标识、更新、列出和删除）见 [API 行为](wire-semantics.md)。消息和函数结果内容见[消息内容](message-content.md)。[Agents API 指南](../../../docs/zh/api/public-agent-api.md)展示了使用 SDK 和 HTTP 的调用方式。 @@ -143,6 +143,8 @@ Turn 和 Item 列表接受 `after`、`limit` 和 `order`（[列表规则](wire-s 工具失败不会导致其 Turn 失败。Session 的项目可以读取工具输出，其中可能包含该工具自己的诊断文本。 +PostgreSQL 无法存储 U+0000，因此 Harness 输出（例如命令输出）中的 U+0000 在存储、事件流和列表中都会替换为 U+FFFD，Turn 照常继续。 + ## 使用量 {#usage} Turn 和 Session 的 `usage` 使用固定的 `TokenUsage` 字段：输入、缓存输入、输出、推理输出和 token 总数。`null` 表示未知，绝不表示零。 diff --git a/services/core/internal/execution/journal.go b/services/core/internal/execution/journal.go index cf3a15438..b9fc7901c 100644 --- a/services/core/internal/execution/journal.go +++ b/services/core/internal/execution/journal.go @@ -6,6 +6,7 @@ import ( "time" "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" + "github.com/MiniMax-AI/OpenAgentCore/internal/obs/log" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions" ) @@ -94,6 +95,8 @@ func (j *journal) flush(ctx context.Context) error { // An uncertain commit must retry the same batch even after more frames arrive. j.pendingCount = count if err := j.writer.AppendTurnEvents(ctx, j.tenant, j.session, j.turn, j.next, j.batch[:count]); err != nil { + // Driver errors can include submitted values, so the raw error is never logged. + log.Ctx(ctx).Error("oac-core turn journal write failed", "session_id", j.session, "turn_id", j.turn, "first_ordinal", j.next, "events", count) return err } j.pendingCount = 0 diff --git a/services/core/internal/persistence/postgres/sessionpg/execution_journal_test.go b/services/core/internal/persistence/postgres/sessionpg/execution_journal_test.go index a7a1ffe7a..4c2e082b8 100644 --- a/services/core/internal/persistence/postgres/sessionpg/execution_journal_test.go +++ b/services/core/internal/persistence/postgres/sessionpg/execution_journal_test.go @@ -114,6 +114,32 @@ func TestTurnJournalRecordsReplaysAndRollsBack(t *testing.T) { } } +func TestTurnJournalStoresNULInObservedTextAsReplacementCharacter(t *testing.T) { + pool, _, operations := leasedJournal(t) + tenant, session, turn := newTurn(t, pool, sessions.TurnInProgress) + batch := []sessions.ExecutionEvent{ + observation("tool_call", `{"id":"c","stage":"before","observation":{"status":"in_progress","kind":"command","command":"printf"}}`), + observation("command_output", `{"id":"c","delta":"before\u0000after"}`), + observation("tool_call", `{"id":"c","stage":"after","observation":{"status":"completed","kind":"command","command":"printf","output":"before\u0000after"}}`), + } + if err := operations.AppendTurnEvents(t.Context(), text(tenant), text(session), text(turn), 1, batch); err != nil { + t.Fatal(err) + } + if _, err := operations.CompleteExecution(t.Context(), text(tenant), text(session), text(turn), sessions.TurnFailed, json.RawMessage(`{"error":"native\u0000"}`), "", 0); err != nil { + t.Fatal(err) + } + var delta, output, failure string + if err := pool.QueryRow(t.Context(), `SELECT + (SELECT payload->>'delta' FROM turn_events WHERE turn_id = $1 AND ordinal = 2), + (SELECT payload#>>'{observation,output}' FROM turn_events WHERE turn_id = $1 AND ordinal = 3), + outcome->>'error' FROM turns WHERE id = $1`, turn).Scan(&delta, &output, &failure); err != nil { + t.Fatal(err) + } + if delta != "before\uFFFDafter" || output != "before\uFFFDafter" || failure != "native\uFFFD" { + t.Fatalf("stored %q, %q and %q", delta, output, failure) + } +} + func TestSubagentProjectionStoresTheSessionRows(t *testing.T) { pool, _, operations := leasedJournal(t) tenant, session, turn, device := uuid.New(), uuid.New(), uuid.New(), uuid.New() diff --git a/services/core/internal/sessions/execution_turns.go b/services/core/internal/sessions/execution_turns.go index a6fbf912c..875297b3a 100644 --- a/services/core/internal/sessions/execution_turns.go +++ b/services/core/internal/sessions/execution_turns.go @@ -221,7 +221,7 @@ func (o *ExecutionOperations) CompleteExecution(ctx context.Context, tenant, ses if !TerminalStatus(status) || len(outcome) > maxOutcomeBytes || len(nativeID) > 512 || appliedThrough < 0 { return Turn{}, ErrInvalidInput } - outcome, err := jsonobject.Normalize(outcome) + outcome, err := journalPayload(outcome) if err != nil { return Turn{}, fmt.Errorf("%w: %w", ErrInvalidInput, err) } diff --git a/services/core/internal/sessions/journal.go b/services/core/internal/sessions/journal.go index a2af177fd..9ffce502d 100644 --- a/services/core/internal/sessions/journal.go +++ b/services/core/internal/sessions/journal.go @@ -1,8 +1,11 @@ package sessions import ( + "bytes" "context" + "encoding/json" "fmt" + "strings" "github.com/google/uuid" @@ -22,7 +25,7 @@ const ( ) // JournalBatch is a validated batch of execution observations for a Turn's -// journal, each payload normalized. NewJournalBatch makes one. +// journal, each payload in its journalPayload form. NewJournalBatch makes one. type JournalBatch struct { turn string first int32 @@ -44,7 +47,7 @@ func NewJournalBatch(turn string, first int32, events []ExecutionEvent) (Journal if len(event.Payload) > journalPayloadBytes || !ValidEngine(event.Kind) { return JournalBatch{}, ErrInvalidInput } - payload, err := jsonobject.Normalize(event.Payload) + payload, err := journalPayload(event.Payload) if err != nil { return JournalBatch{}, fmt.Errorf("%w: %w", ErrInvalidInput, err) } @@ -57,6 +60,43 @@ func NewJournalBatch(turn string, first int32, events []ExecutionEvent) (Journal return batch, nil } +// journalPayload normalizes an execution observation's payload into the form +// the journal stores. Observed text is the Harness's own output, such as a +// command's, so U+0000, which jsonb cannot store, becomes U+FFFD in every +// string, as decoding already does for invalid UTF-8, instead of failing the +// Turn. +func journalPayload(raw json.RawMessage) (json.RawMessage, error) { + payload, err := jsonobject.Normalize(raw) + if err != nil || !bytes.Contains(payload, []byte(`\u0000`)) { + return payload, err + } + decoder := json.NewDecoder(bytes.NewReader(payload)) + decoder.UseNumber() + var value any + if err := decoder.Decode(&value); err != nil { + return nil, err + } + return json.Marshal(withoutNUL(value)) +} + +func withoutNUL(value any) any { + switch v := value.(type) { + case string: + return strings.ReplaceAll(v, "\x00", "\uFFFD") + case []any: + for i, item := range v { + v[i] = withoutNUL(item) + } + case map[string]any: + replaced := make(map[string]any, len(v)) + for key, item := range v { + replaced[strings.ReplaceAll(key, "\x00", "\uFFFD")] = withoutNUL(item) + } + return replaced + } + return value +} + // validID reports whether value can name a stored resource: a nonzero UUID. func validID(value string) bool { id, err := uuid.Parse(value)