From 74a6ee81764b7b0bd61147e0691c438ab09b1644 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Mon, 28 Sep 2026 05:44:52 +0000 Subject: [PATCH 1/3] feat(runtime): retain structured native failure classifications --- .../claudesdk/error_classification_test.go | 90 ++++++++++++++++++ .../internal/agent/claudesdk/session.go | 70 +++++++++----- .../internal/agent/claudesdk/session_test.go | 4 + .../agent/codex/error_classification.go | 60 ++++++++++++ .../agent/codex/error_classification_test.go | 95 +++++++++++++++++++ .../internal/agent/codex/session.go | 9 +- .../agents-api/native-error-classification.md | 48 ++++++++++ docs/design-principles.md | 6 ++ internal/agentdaemon/proto/engine_failure.go | 37 ++++++++ .../agentdaemon/proto/engine_failure_test.go | 42 ++++++++ internal/agentdaemon/proto/inbound.go | 4 +- packages/claude-sdk-adapter/src/adapter.ts | 20 +++- packages/claude-sdk-adapter/src/inputs.ts | 7 ++ .../claude-sdk-adapter/src/native_failure.ts | 33 +++++++ .../tests/classified_execution.test.mjs | 43 +++++++++ .../tests/native_failure.test.mjs | 50 ++++++++++ .../agents-api/internal/execution/delivery.go | 6 ++ .../internal/execution/dispatcher.go | 10 +- .../internal/execution/engine_failure_test.go | 75 +++++++++++++++ 19 files changed, 674 insertions(+), 35 deletions(-) create mode 100644 apps/parsar-daemon/internal/agent/claudesdk/error_classification_test.go create mode 100644 apps/parsar-daemon/internal/agent/codex/error_classification.go create mode 100644 apps/parsar-daemon/internal/agent/codex/error_classification_test.go create mode 100644 contracts/agents-api/native-error-classification.md create mode 100644 internal/agentdaemon/proto/engine_failure.go create mode 100644 internal/agentdaemon/proto/engine_failure_test.go create mode 100644 packages/claude-sdk-adapter/src/native_failure.ts create mode 100644 packages/claude-sdk-adapter/tests/classified_execution.test.mjs create mode 100644 packages/claude-sdk-adapter/tests/native_failure.test.mjs create mode 100644 services/agents-api/internal/execution/engine_failure_test.go diff --git a/apps/parsar-daemon/internal/agent/claudesdk/error_classification_test.go b/apps/parsar-daemon/internal/agent/claudesdk/error_classification_test.go new file mode 100644 index 000000000..04f76f1b4 --- /dev/null +++ b/apps/parsar-daemon/internal/agent/claudesdk/error_classification_test.go @@ -0,0 +1,90 @@ +//go:build unix + +package claudesdk + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/proto" +) + +func TestClassifiedBridgeFailurePreservesTerminalEvidence(t *testing.T) { + for _, mode := range []string{"valid", "wrong-session", "wrong-result", "success-usage", "cancelled", "unknown", "malformed-code", "after-terminal", "scanner-error", "process-error"} { + t.Run(mode, func(t *testing.T) { + root := t.TempDir() + t.Setenv("OAC_RUNTIME_HOME", root) + config := Config{Node: os.Args[0], Entrypoint: filepath.Join(root, "worker"), StateDir: filepath.Join(root, "state"), Env: []string{"GO_CLAUDE_SDK_HELPER=1", "SDK_HELPER_MODE=classified-" + mode, "GORACE=atexit_sleep_ms=0"}} + req := proto.PromptRequestPayload{RunID: "run", Input: proto.TextInput("hello"), AgentSessionID: "native-session", AgentOptions: map[string]any{"model": "fake-model", "system_prompt": "instructions"}} + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + out := make(chan proto.Envelope, 16) + s, err := NewFactory(config)(ctx, req, out) + if err != nil { + t.Fatal(err) + } + defer s.Cancel(context.Background()) + var failure proto.ErrorPayload + var done proto.DonePayload + usage, errors, dones := 0, 0, 0 + for event := range out { + switch event.Type { + case proto.TypeUsage: + usage++ + case proto.TypeError: + errors++ + _ = event.DecodePayload(&failure) + case proto.TypeDone: + dones++ + _ = event.DecodePayload(&done) + } + } + want := "" + if mode == "valid" { + want = "authentication_error" + } + if errors != 1 || dones != 1 || usage != 1 || failure.Code != want || done.Metadata[proto.DoneMetaAgentSessionID] != "native-session" || done.Usage.Raw["claude_sdk_result"] == nil { + t.Fatalf("lost evidence: %d/%d/%d %+v %+v", errors, dones, usage, failure, done) + } + }) + } +} + +func runClassifiedFailureHelper(request startRequest, mode string, encode func(bridgeEvent)) { + mode = strings.TrimPrefix(mode, "classified-") + encode(bridgeEvent{Type: "input_ready", SessionID: request.Resume}) + raw := strings.ReplaceAll(strings.ReplaceAll(usageFixture, `"subtype":"success"`, `"subtype":"error_during_execution"`), `"is_error":false`, `"is_error":true`) + if mode == "success-usage" { + raw = usageFixture + } + encode(bridgeEvent{Type: "usage", SessionID: request.Resume, ResultID: "result", Usage: json.RawMessage(raw)}) + e := bridgeEvent{Type: "error", Code: "execution_failed", SessionID: request.Resume, ResultID: "result", EngineErrorCode: json.RawMessage(`"authentication_error"`)} + switch mode { + case "wrong-session": + e.SessionID = "foreign" + case "wrong-result": + e.ResultID = "foreign" + case "cancelled": + e.Code = "cancelled" + case "unknown": + e.EngineErrorCode = json.RawMessage(`"future"`) + case "malformed-code": + e.EngineErrorCode = json.RawMessage(`{"secret":"value"}`) + } + encode(e) + if mode == "process-error" { + os.Exit(7) + } + if mode == "after-terminal" { + fmt.Fprintln(os.Stdout, `{"type":"delta","delta":"late"}`) + } + if mode == "scanner-error" { + fmt.Fprintln(os.Stdout, strings.Repeat("x", 2*1024*1024)) + } +} diff --git a/apps/parsar-daemon/internal/agent/claudesdk/session.go b/apps/parsar-daemon/internal/agent/claudesdk/session.go index 267811f80..fe88ea0cd 100644 --- a/apps/parsar-daemon/internal/agent/claudesdk/session.go +++ b/apps/parsar-daemon/internal/agent/claudesdk/session.go @@ -78,23 +78,24 @@ func NewFactory(config Config) agent.Factory { } type bridgeEvent struct { - Fact json.RawMessage `json:"fact"` - InputID string `json:"input_id"` - ResultID string `json:"result_id"` - Usage json.RawMessage `json:"usage,omitempty"` - Type string `json:"type"` - Delta string `json:"delta"` - SessionID string `json:"session_id"` - Text string `json:"text"` - Code string `json:"code"` - ItemID string `json:"item_id"` - Message *proto.OutputMessagePayload `json:"message"` - Call *proto.FunctionCallPayload `json:"call"` - CallID string `json:"call_id"` - DeliveryID string `json:"delivery_id"` - ID string `json:"id"` - Stage string `json:"stage"` - Observation *proto.ToolObservation `json:"observation"` + EngineErrorCode json.RawMessage `json:"engine_error_code"` + Fact json.RawMessage `json:"fact"` + InputID string `json:"input_id"` + ResultID string `json:"result_id"` + Usage json.RawMessage `json:"usage,omitempty"` + Type string `json:"type"` + Delta string `json:"delta"` + SessionID string `json:"session_id"` + Text string `json:"text"` + Code string `json:"code"` + ItemID string `json:"item_id"` + Message *proto.OutputMessagePayload `json:"message"` + Call *proto.FunctionCallPayload `json:"call"` + CallID string `json:"call_id"` + DeliveryID string `json:"delivery_id"` + ID string `json:"id"` + Stage string `json:"stage"` + Observation *proto.ToolObservation `json:"observation"` } func (s *session) run(ctx context.Context, runID string, start startRequest, out chan<- proto.Envelope, prepared *prepared) { @@ -141,7 +142,7 @@ func (s *session) run(ctx context.Context, runID string, start startRequest, out if prepared != nil { binding, err := prepared.awaitStart(scanner, failure) if binding == nil { - prepared.failure = s.drain(scanner, stderrDone, err) + prepared.failure, _ = s.drain(scanner, stderrDone, err) close(s.settled) return } @@ -159,8 +160,11 @@ func (s *session) run(ctx context.Context, runID string, start startRequest, out } var content strings.Builder var result *bridgeEvent + var classifiedFailure error + var engineCode string var usage proto.Usage var usageSession string + var failedResultID string usageIDs := map[string]bool{} var sequence uint64 terminal := false @@ -236,6 +240,14 @@ func (s *session) run(ctx context.Context, runID string, start startRequest, out usage = nextUsage usageIDs[event.ResultID] = true usageSession = event.SessionID + failedResultID = "" + var nativeResult struct { + Subtype string `json:"subtype"` + IsError bool `json:"is_error"` + } + if json.Unmarshal(event.Usage, &nativeResult) == nil && (nativeResult.Subtype == "error_during_execution" || nativeResult.Subtype == "success" && nativeResult.IsError) { + failedResultID = event.ResultID + } emit(proto.TypeUsage, proto.UsagePayload{Usage: usage}) case "result": if !s.matchesInputSession(event.SessionID) || event.SessionID == "" || start.Resume != "" && event.SessionID != start.Resume || usageSession != "" && event.SessionID != usageSession || !s.functionsComplete() || !s.steeringComplete() || !mcp.complete() || !commands.complete() { @@ -247,6 +259,13 @@ func (s *session) run(ctx context.Context, runID string, start startRequest, out terminal = true case "error": failure = bridgeFailure(event.Code) + if event.Code == "execution_failed" && event.SessionID != "" && s.matchesInputSession(event.SessionID) && + event.SessionID == usageSession && event.ResultID != "" && event.ResultID == failedResultID { + var code string + _ = json.Unmarshal(event.EngineErrorCode, &code) + engineCode, _ = proto.NormalizeEngineFailure(code, nil) + classifiedFailure = failure + } terminal = true default: failure = fmt.Errorf("claudesdk: unknown SDK bridge event") @@ -256,7 +275,8 @@ func (s *session) run(ctx context.Context, runID string, start startRequest, out break } } - failure = s.drain(scanner, stderrDone, failure) + var bridgeClean bool + failure, bridgeClean = s.drain(scanner, stderrDone, failure) if result == nil && failure == nil { failure = fmt.Errorf("claudesdk: SDK result is missing") } @@ -277,7 +297,10 @@ func (s *session) run(ctx context.Context, runID string, start startRequest, out // Router completion cleanup calls Cancel while consuming Done. Settle first. close(s.settled) if failure != nil { - emit(proto.TypeError, proto.ErrorPayload{Error: failure.Error()}) + if failure != classifiedFailure || !bridgeClean { + engineCode = "" + } + emit(proto.TypeError, proto.ErrorPayload{Error: failure.Error(), Code: engineCode}) } emit(proto.TypeDone, s.outcome) } @@ -294,7 +317,7 @@ func launch(ctx context.Context, config Config, start startRequest, env []string return &session{process: process, functions: functionState{calls: map[string]*pendingFunction{}}, settled: make(chan struct{})}, nil } -func (s *session) drain(scanner *bridgeOutput, stderrDone <-chan struct{}, failure error) error { +func (s *session) drain(scanner *bridgeOutput, stderrDone <-chan struct{}, failure error) (error, bool) { for scanner.Scan() { } if scanner.Err() != nil { @@ -304,10 +327,11 @@ func (s *session) drain(scanner *bridgeOutput, stderrDone <-chan struct{}, failu <-stderrDone s.stopWorkspaceReads() s.stopWorkspaceDirectories() - if err := s.process.Wait(); err != nil && failure == nil { + waitErr := s.process.Wait() + if waitErr != nil && failure == nil { failure = fmt.Errorf("claudesdk: SDK process failed") } - return failure + return failure, scanner.Err() == nil && waitErr == nil } func bridgeFailure(code string) error { diff --git a/apps/parsar-daemon/internal/agent/claudesdk/session_test.go b/apps/parsar-daemon/internal/agent/claudesdk/session_test.go index 2a736f7f2..37d64ded3 100644 --- a/apps/parsar-daemon/internal/agent/claudesdk/session_test.go +++ b/apps/parsar-daemon/internal/agent/claudesdk/session_test.go @@ -135,6 +135,10 @@ func runSDKHelper() { runSteeringHelper(request, mode, scanner, encode) return } + if strings.HasPrefix(mode, "classified-") { + runClassifiedFailureHelper(request, mode, encode) + return + } if strings.HasPrefix(mode, "usage-") { runUsageHelper(request, mode, encode) return diff --git a/apps/parsar-daemon/internal/agent/codex/error_classification.go b/apps/parsar-daemon/internal/agent/codex/error_classification.go new file mode 100644 index 000000000..482811a97 --- /dev/null +++ b/apps/parsar-daemon/internal/agent/codex/error_classification.go @@ -0,0 +1,60 @@ +package codex + +import ( + "encoding/json" + + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/proto" +) + +// Only the exact root terminal TurnError is authoritative. Error notifications, +// including exhausted/recovered retries, never become a sticky classification. +func classifyTurnError(failure *TurnError) proto.ErrorPayload { + if failure == nil { + return proto.ErrorPayload{} + } + var variant string + var code string + if json.Unmarshal(failure.CodexErrorInfo, &variant) == nil { + switch variant { + case "unauthorized": + code = "authentication_error" + case "usageLimitExceeded": + code = "usage_limit_exceeded" + case "rateLimitExceeded": + code = "rate_limit_exceeded" + case "contextWindowExceeded": + code = "context_length_exceeded" + case "serverOverloaded": + code = "server_overloaded" + case "internalServerError": + code = "server_error" + case "badRequest": + code = "invalid_request" + case "cyberPolicy": + code = "cyber_policy" + } + return proto.ErrorPayload{Code: code} + } + var object map[string]json.RawMessage + if json.Unmarshal(failure.CodexErrorInfo, &object) != nil || len(object) != 1 { + return proto.ErrorPayload{} + } + for variant, raw := range object { + switch variant { + case "httpConnectionFailed", "responseStreamConnectionFailed", "responseStreamDisconnected", "responseTooManyFailedAttempts": + var fields map[string]json.RawMessage + if json.Unmarshal(raw, &fields) != nil || fields == nil { + return proto.ErrorPayload{} + } + var status *int + _ = json.Unmarshal(fields["httpStatusCode"], &status) + code = "connection_failed" + if variant == "responseTooManyFailedAttempts" && status != nil && *status == 429 { + code = "rate_limit_exceeded" + } + code, status = proto.NormalizeEngineFailure(code, status) + return proto.ErrorPayload{Code: code, HTTPStatus: status} + } + } + return proto.ErrorPayload{} +} diff --git a/apps/parsar-daemon/internal/agent/codex/error_classification_test.go b/apps/parsar-daemon/internal/agent/codex/error_classification_test.go new file mode 100644 index 000000000..5dd41790c --- /dev/null +++ b/apps/parsar-daemon/internal/agent/codex/error_classification_test.go @@ -0,0 +1,95 @@ +package codex + +import ( + "context" + "encoding/json" + "testing" + + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/proto" +) + +func TestPinnedTerminalErrorClassification(t *testing.T) { + for value, want := range map[string]string{ + "unauthorized": "authentication_error", "usageLimitExceeded": "usage_limit_exceeded", "rateLimitExceeded": "rate_limit_exceeded", + "contextWindowExceeded": "context_length_exceeded", "serverOverloaded": "server_overloaded", "internalServerError": "server_error", + "badRequest": "invalid_request", "cyberPolicy": "cyber_policy", "sessionBudgetExceeded": "", "misalignmentPolicyViolation": "", + "threadRollbackFailed": "", "sandboxError": "", "other": "", "unknownFuture": "", + } { + raw, _ := json.Marshal(value) + got := classifyTurnError(&TurnError{Message: "unauthorized 429 timeout", CodexErrorInfo: raw}) + if got.Code != want || got.HTTPStatus != nil { + t.Fatalf("%s: %+v", value, got) + } + } + for _, tc := range []struct { + raw, code string + status int + }{ + {`{"httpConnectionFailed":{"httpStatusCode":401}}`, "connection_failed", 401}, + {`{"responseStreamConnectionFailed":{"httpStatusCode":503}}`, "connection_failed", 503}, + {`{"responseStreamDisconnected":{"httpStatusCode":null}}`, "connection_failed", 0}, + {`{"responseTooManyFailedAttempts":{"httpStatusCode":429}}`, "rate_limit_exceeded", 0}, + {`{"responseTooManyFailedAttempts":{"httpStatusCode":502}}`, "connection_failed", 502}, + {`{"httpConnectionFailed":{"httpStatusCode":"401"}}`, "connection_failed", 0}, + {`{"httpConnectionFailed":{"httpStatusCode":65535}}`, "connection_failed", 0}, + {`{"httpConnectionFailed":null}`, "", 0}, {`{"activeTurnNotSteerable":{"turnKind":"review"}}`, "", 0}, + {`{"httpConnectionFailed":{},"other":{}}`, "", 0}, {`null`, "", 0}, + } { + got := classifyTurnError(&TurnError{CodexErrorInfo: json.RawMessage(tc.raw)}) + status := 0 + if got.HTTPStatus != nil { + status = *got.HTTPStatus + } + if got.Code != tc.code || status != tc.status { + t.Fatalf("%s: %+v", tc.raw, got) + } + } +} + +func TestClassificationOnlyFromRootTerminalError(t *testing.T) { + for _, mode := range []string{"failed", "recovered", "notification-only"} { + t.Run(mode, func(t *testing.T) { + out := make(chan proto.Envelope, 16) + s := &Session{runID: "run", out: out, cancelCtx: context.Background(), cfg: defaultSessionConfig(), rpc: NewJSONRPCClient(JSONRPCConfig{})} + s.registerHandlers() + s.setThreadID("root") + scopeNotification(t, s, "turn/started", `{"threadId":"root","turn":{"id":"turn"}}`) + scopeNotification(t, s, "error", `{"threadId":"root","turnId":"turn","error":{"message":"retry","codexErrorInfo":"unauthorized"},"willRetry":true}`) + scopeNotification(t, s, "turn/completed", `{"threadId":"child","turn":{"id":"turn","status":"failed","error":{"message":"child","codexErrorInfo":"usageLimitExceeded"}}}`) + status, errJSON := "failed", `{"message":"terminal","codexErrorInfo":"serverOverloaded"}` + if mode == "recovered" { + status = "completed" + } + if mode == "notification-only" { + errJSON = "null" + } + scopeNotification(t, s, "turn/completed", `{"threadId":"root","turn":{"id":"turn","status":"`+status+`","usage":{"inputTokens":7,"outputTokens":2},"error":`+errJSON+`}}`) + errors, done := 0, false + for env := range out { + if env.Type == proto.TypeError { + errors++ + var p proto.ErrorPayload + _ = env.DecodePayload(&p) + want := "" + if mode == "failed" { + want = "server_overloaded" + } + if p.Code != want { + t.Fatal(p) + } + } + if env.Type == proto.TypeDone { + done = true + var p proto.DonePayload + _ = env.DecodePayload(&p) + if p.Metadata[proto.DoneMetaAgentSessionID] != "root" || p.Usage.InputTokens != 7 { + t.Fatal(p) + } + } + } + if !done || errors != map[bool]int{true: 0, false: 1}[mode == "recovered"] { + t.Fatal("terminal lost", done, errors) + } + }) + } +} diff --git a/apps/parsar-daemon/internal/agent/codex/session.go b/apps/parsar-daemon/internal/agent/codex/session.go index f9f624f1d..5c212c4e7 100644 --- a/apps/parsar-daemon/internal/agent/codex/session.go +++ b/apps/parsar-daemon/internal/agent/codex/session.go @@ -242,7 +242,7 @@ func (s *Session) onTurnCompleted(raw json.RawMessage) { if errText != "" { body = appendOnNewline(body, errText) } - s.emitTerminal(body, true) + s.emitTerminalFailure(body, true, classifyTurnError(p.Turn.Error)) s.finishAfterTerminal() return } @@ -392,6 +392,10 @@ func (s *Session) emitUsage(u TurnUsage) { } func (s *Session) emitTerminal(message string, asError bool) { + s.emitTerminalFailure(message, asError, proto.ErrorPayload{}) +} + +func (s *Session) emitTerminalFailure(message string, asError bool, failure proto.ErrorPayload) { if !s.terminal.CompareAndSwap(false, true) { return } @@ -414,7 +418,8 @@ func (s *Session) emitTerminal(message string, asError bool) { } var events []proto.Envelope if asError { - env, err := proto.NewEnvelope(proto.TypeError, s.runID, proto.ErrorPayload{Error: message}) + failure.Error = message + env, err := proto.NewEnvelope(proto.TypeError, s.runID, failure) if err == nil { events = append(events, env) } diff --git a/contracts/agents-api/native-error-classification.md b/contracts/agents-api/native-error-classification.md new file mode 100644 index 000000000..45aa7f2f0 --- /dev/null +++ b/contracts/agents-api/native-error-classification.md @@ -0,0 +1,48 @@ +# Native failure classification + +Runtime adapters may add `code` and `http_status` to a prompt-level `error` frame. +The fields are optional neutral metadata, not a new terminal event or a public +Agents API error contract. Existing error text, Usage, Done, native Session identity +and cancellation receipts retain their existing ownership and order. + +Core stores accepted values in the Turn outcome as `engine_error_code` and +`engine_http_status`. Classification is subordinate to terminal status and Core's +`error_code`: it cannot turn a completed or cancelled Turn into a failure, hide an +incomplete event stream, or override persistence and cancellation-receipt failures. +Normal delivery and terminal journal draining use the same extraction rule. + +Accepted codes are `authentication_error`, `rate_limit_exceeded`, +`usage_limit_exceeded`, `server_overloaded`, `server_error`, `invalid_request`, +`resource_not_found`, `request_timeout`, `context_length_exceeded`, `cyber_policy` +and `connection_failed`. Only `connection_failed` retains an integer HTTP status +in 100..599; all other status metadata is discarded. Missing, malformed and unknown +optional values degrade to unclassified metadata without discarding Usage or Done. +Old Runtime frames remain generic harness failures. Old readers ignore new fields. +The catalog is not a claim that every harness can produce every classification. + +Codex 0.153.4 uses only the exact root terminal Turn's `codexErrorInfo`, never +notification text or a retry notification. Its finite string mappings cover +unauthorized, usageLimitExceeded, rateLimitExceeded, contextWindowExceeded, +serverOverloaded, internalServerError, badRequest and cyberPolicy. The connection +object variants preserve valid upstream status. responseTooManyFailedAttempts maps +to rate_limit_exceeded only for 429; otherwise it is connection_failed. Session +budget, misalignment policy, rollback, sandbox and steering-state errors remain +unclassified. A successful terminal Turn does not inherit an earlier error. + +Claude SDK 0.3.269 records finite root assistant errors against current submitted +input UUIDs and the same native Session. Only a matching unsuccessful native result +commits that candidate. Replayed, synthetic, child, foreign and completed-input +messages cannot classify the root failure; recovered results clear candidates. +Authentication variants map to authentication_error, billing_error to +usage_limit_exceeded, rate_limit to rate_limit_exceeded, overloaded to +server_overloaded, invalid_request to invalid_request, model_not_found to +resource_not_found and server_error to server_error. Unknown/max-output-token, +max-turn/budget/structured-retry results and unstructured exceptions remain generic. +The bridge retains native usage and identity and rejects stale classification after +malformed output or an unsuccessful bridge process exit. A validated native failure +may itself have a nonzero native child exit; that is distinct from bridge failure. + +Neither adapter classifies error prose, exposes provider messages through safe +metadata, nor synthesizes tool timeouts. request_timeout has no current producer. +MiniMax remains unclassified. Real provider rejection evidence is distinct from +controlled protocol/SDK-stream tests and must be qualified against a matched Runtime. diff --git a/docs/design-principles.md b/docs/design-principles.md index be2c42ba6..bccbf88b9 100644 --- a/docs/design-principles.md +++ b/docs/design-principles.md @@ -81,3 +81,9 @@ ownership. No request bodies, secrets or file contents enter audit records. Do not add product users, RBAC, cross-Project shared assets, administrator execution, or old private-protocol compatibility to this management model. Existing public execution and Runtime ownership rules remain in [CONTRIBUTING.md](../CONTRIBUTING.md). + +Native failure classification is adapter-owned and uses finite structured native +values. Optional Runtime error metadata is normalized once and retained through +normal delivery and journal draining; it never replaces Core terminal authority, +cancellation receipts, Usage or native identity. See +[native failure classification](../contracts/agents-api/native-error-classification.md). diff --git a/internal/agentdaemon/proto/engine_failure.go b/internal/agentdaemon/proto/engine_failure.go new file mode 100644 index 000000000..03574f499 --- /dev/null +++ b/internal/agentdaemon/proto/engine_failure.go @@ -0,0 +1,37 @@ +package proto + +import "encoding/json" + +// NormalizeEngineFailure accepts only neutral, finite classifications. Optional +// metadata from newer or malformed peers never changes terminal/usage decoding. +func NormalizeEngineFailure(code string, status *int) (string, *int) { + switch code { + case "authentication_error", "rate_limit_exceeded", "usage_limit_exceeded", "server_overloaded", "server_error", "invalid_request", "resource_not_found", "request_timeout", "context_length_exceeded", "cyber_policy": + return code, nil + case "connection_failed": + if status != nil && *status >= 100 && *status <= 599 { + return code, status + } + return code, nil + default: + return "", nil + } +} + +func (p *ErrorPayload) UnmarshalJSON(raw []byte) error { + var wire struct { + Error string `json:"error"` + Code json.RawMessage `json:"code"` + HTTPStatus json.RawMessage `json:"http_status"` + } + if err := json.Unmarshal(raw, &wire); err != nil { + return err + } + var code string + var status *int + _ = json.Unmarshal(wire.Code, &code) + _ = json.Unmarshal(wire.HTTPStatus, &status) + p.Error = wire.Error + p.Code, p.HTTPStatus = NormalizeEngineFailure(code, status) + return nil +} diff --git a/internal/agentdaemon/proto/engine_failure_test.go b/internal/agentdaemon/proto/engine_failure_test.go new file mode 100644 index 000000000..29b6bf38b --- /dev/null +++ b/internal/agentdaemon/proto/engine_failure_test.go @@ -0,0 +1,42 @@ +package proto + +import ( + "encoding/json" + "testing" +) + +func TestErrorClassificationCompatibility(t *testing.T) { + for _, tc := range []struct { + raw, code string + status int + }{ + {`{"error":"old"}`, "", 0}, + {`{"error":"old","code":"authentication_error","http_status":401}`, "authentication_error", 0}, + {`{"error":"old","code":"connection_failed","http_status":503}`, "connection_failed", 503}, + {`{"error":"old","code":"connection_failed","http_status":"secret"}`, "connection_failed", 0}, + {`{"error":"old","code":"connection_failed","http_status":99}`, "connection_failed", 0}, + {`{"error":"old","code":"connection_failed","http_status":600}`, "connection_failed", 0}, + {`{"error":"old","code":"future","http_status":503}`, "", 0}, + {`{"error":"old","code":{"secret":"value"},"http_status":503}`, "", 0}, + {`{"error":"old","code":null,"http_status":null}`, "", 0}, + } { + var p ErrorPayload + if err := json.Unmarshal([]byte(tc.raw), &p); err != nil { + t.Fatal(err) + } + status := 0 + if p.HTTPStatus != nil { + status = *p.HTTPStatus + } + if p.Error != "old" || p.Code != tc.code || status != tc.status { + t.Fatalf("%s: %+v", tc.raw, p) + } + raw, _ := json.Marshal(p) + var old struct { + Error string `json:"error"` + } + if json.Unmarshal(raw, &old) != nil || old.Error != "old" { + t.Fatal("old peer compatibility", string(raw)) + } + } +} diff --git a/internal/agentdaemon/proto/inbound.go b/internal/agentdaemon/proto/inbound.go index 1b2770ee0..1060f6515 100644 --- a/internal/agentdaemon/proto/inbound.go +++ b/internal/agentdaemon/proto/inbound.go @@ -229,7 +229,9 @@ type UsagePayload struct { // ErrorPayload reports a prompt-level failure. type ErrorPayload struct { - Error string `json:"error"` + Error string `json:"error"` + Code string `json:"code,omitempty"` + HTTPStatus *int `json:"http_status,omitempty"` } // DonePayload mirrors connector.PromptOutput shape. Redeclared (not diff --git a/packages/claude-sdk-adapter/src/adapter.ts b/packages/claude-sdk-adapter/src/adapter.ts index e4b86b070..6be20e8e9 100644 --- a/packages/claude-sdk-adapter/src/adapter.ts +++ b/packages/claude-sdk-adapter/src/adapter.ts @@ -1,3 +1,4 @@ +import { NativeFailure } from "./native_failure.js"; import { StructuredOutput } from "./structured_output.js"; import { toolSearchEnvironment } from "./tool_search.js"; import { Subagents } from "./subagents.js"; @@ -32,9 +33,12 @@ export type Event = | { type: "usage"; session_id: string; result_id: string; usage: NativeUsage } | { type: "delta"; delta: string } | { type: "result"; session_id: string; text: string } - | { type: "error"; code: "invalid_request" | "history_unavailable" | "execution_failed" | "cancelled" }; + | { type: "error"; code: "invalid_request" | "history_unavailable" | "execution_failed" | "cancelled"; engine_error_code?: string; session_id?: string; result_id?: string }; export async function execute(request: Start | Prepare, emit: (event: Event) => Promise, abort: AbortController, functions = new FunctionBridge(emit), inputs = new Inputs(immediateInput(request)), reads = new WorkspaceReads(emit, abort), directories = new WorkspaceDirectories(emit, abort)): Promise { + const nativeFailure = new NativeFailure(); + const nativeResultFailure = new Error("unsuccessful native result"); + let classifiedFailure: { engine_error_code: string; session_id: string; result_id: string } | undefined; const definitions = (request.functions ?? []).map(tool => ({ name: tool.name, description: tool.description, inputSchema: tool.parameters, deferLoading: tool.defer_loading })); const names = definitions.map(tool => `mcp__functions__${tool.name}`); const allowed = [...names, ...(request.tool_search ? ["ToolSearch"] : [])]; @@ -132,6 +136,7 @@ export async function execute(request: Start | Prepare, emit: (event: Event) => inputs.release(request.input); } else stream = query({ prompt: inputs, options }); for await (const message of stream) { + const errorCode = nativeFailure.observe(message, nativeID, inputs.pendingInputIDs(message)); subagents?.consume(message); structured?.consume(message, nativeID); await functions.consume(message, nativeID); @@ -157,7 +162,10 @@ export async function execute(request: Start | Prepare, emit: (event: Event) => resultIDs.add(message.uuid); await emit({ type: "usage", session_id: nativeID, result_id: message.uuid, usage: resultUsage(message) }); for (const event of inputs.consume(message)) await emit(event); - if (message.subtype !== "success" || message.is_error) throw new Error("unsuccessful native result"); + if (message.subtype !== "success" || message.is_error) { + if (errorCode) classifiedFailure = { engine_error_code: errorCode, session_id: nativeID, result_id: message.uuid }; + throw nativeResultFailure; + } if (structured) await emit(structured.complete(message)); result = { type: "result", session_id: nativeID, text: message.result }; } @@ -167,18 +175,20 @@ export async function execute(request: Start | Prepare, emit: (event: Event) => functions.assertComplete(); mcp?.assertComplete(); commands?.assertComplete(); - } catch { + } catch (error) { + if (error !== nativeResultFailure) classifiedFailure = undefined; failed = true; } finally { profile?.close(); inputs.close(); functions.close(); - try { await directories.close(); } catch { failed = true; } + try { await directories.close(); } catch { failed = true; classifiedFailure = undefined; } await reads.close(); stream?.close(); warm?.close(); abort.signal.removeEventListener("abort", closeInputs); const exits = await Promise.all(children); + if (!exits.length) classifiedFailure = undefined; if (!exits.length || exits.some(code => code !== 0)) failed = true; if (subagents && abort.signal.aborted && exits.length === 1) { try { for (const event of await subagents.facts(Date.now())) await emit(event); } @@ -189,6 +199,6 @@ export async function execute(request: Start | Prepare, emit: (event: Event) => } if (cancellationFactsFailed) await emit({ type: "error", code: "execution_failed" }); else if (abort.signal.aborted) await emit({ type: "error", code: "cancelled" }); - else if (failed || !result || !inputs.complete) await emit({ type: "error", code: "execution_failed" }); + else if (failed || !result || !inputs.complete) await emit({ type: "error", code: "execution_failed", ...classifiedFailure }); else await emit(result); } diff --git a/packages/claude-sdk-adapter/src/inputs.ts b/packages/claude-sdk-adapter/src/inputs.ts index 67b37d768..1e77ffcd1 100644 --- a/packages/claude-sdk-adapter/src/inputs.ts +++ b/packages/claude-sdk-adapter/src/inputs.ts @@ -92,6 +92,13 @@ export class Inputs implements AsyncIterable { return receipts; } + // Only current, admitted input identities can support terminal error attribution. + pendingInputIDs(message: SDKMessage): string[] { + if (message.type !== "assistant" && message.type !== "result") return []; + const ids = message.user_message_uuids ?? (message.user_message_uuid ? [message.user_message_uuid] : []); + return ids.every(id => { const input = this.submitted.get(id); return input !== undefined && !input.completed; }) ? ids : []; + } + get complete(): boolean { return [...this.submitted.values()].every(input => input.completed); } get hasInput(): boolean { return this.submitted.size > 0; } diff --git a/packages/claude-sdk-adapter/src/native_failure.ts b/packages/claude-sdk-adapter/src/native_failure.ts new file mode 100644 index 000000000..436a7aa32 --- /dev/null +++ b/packages/claude-sdk-adapter/src/native_failure.ts @@ -0,0 +1,33 @@ +import type { SDKAssistantMessageError, SDKMessage } from "@anthropic-ai/claude-agent-sdk"; + +const codes: Record = { + authentication_failed: "authentication_error", oauth_org_not_allowed: "authentication_error", + account_on_hold: "authentication_error", verification_required: "authentication_error", + cloud_credential_error: "authentication_error", billing_error: "usage_limit_exceeded", + rate_limit: "rate_limit_exceeded", overloaded: "server_overloaded", + invalid_request: "invalid_request", model_not_found: "resource_not_found", server_error: "server_error", + unknown: undefined, max_output_tokens: undefined, +}; + +// Candidates are scoped to uncompleted, Core-submitted input UUIDs, not a query's +// last error text. A matching native error result alone commits a classification. +export class NativeFailure { + private readonly candidates = new Map(); + + observe(message: SDKMessage, sessionID: string, pending: readonly string[]): string | undefined { + if (!sessionID || message.session_id !== sessionID || !pending.length || + ("isReplay" in message && message.isReplay) || ("isSynthetic" in message && message.isSynthetic)) return; + if (message.type === "assistant" && message.parent_tool_use_id === null) { + const code = message.error === undefined || !Object.hasOwn(codes, message.error) ? undefined : codes[message.error]; + for (const id of pending) { + this.candidates.delete(id); + if (code) this.candidates.set(id, code); + } + } else if (message.type === "result") { + const candidate = this.candidates.get(pending[0]!); + const same = pending.every(id => this.candidates.get(id) === candidate); + for (const id of pending) this.candidates.delete(id); + if (same && (message.subtype === "error_during_execution" || message.subtype === "success" && message.is_error)) return candidate; + } + } +} diff --git a/packages/claude-sdk-adapter/tests/classified_execution.test.mjs b/packages/claude-sdk-adapter/tests/classified_execution.test.mjs new file mode 100644 index 000000000..9a40bcf2f --- /dev/null +++ b/packages/claude-sdk-adapter/tests/classified_execution.test.mjs @@ -0,0 +1,43 @@ +import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; +import test from "node:test"; + +// Actual execute()/child lifetime with a controlled SDK stream, not model evidence. +const fixture = ` +import assert from "node:assert/strict"; +import { registerHooks } from "node:module"; +const sdk='export async function getSessionInfo(){} export function query(o){return globalThis.queryFixture(o);} export function startup(){throw new Error("unexpected");}'; +registerHooks({resolve(s,c,n){if(s==="@anthropic-ai/claude-agent-sdk")return {url:"data:text/javascript,"+encodeURIComponent(sdk),shortCircuit:true};return n(s,c);}}); +const {execute}=await import(${JSON.stringify(new URL("../dist/adapter.js",import.meta.url).href)}); +const mode=process.argv[1], events=[], abort=new AbortController(); +globalThis.queryFixture=({prompt,options})=>{ + const child=options.spawnClaudeCodeProcess({command:process.execPath,args:["-e","process.stdin.resume();process.stdin.on('end',()=>process.exit(1));"],env:process.env}); + return {close(){child.stdin.end();},async *[Symbol.asyncIterator](){ + const input=(await prompt[Symbol.asyncIterator]().next()).value; + yield {type:"system",subtype:"init",session_id:"native",tools:[],mcp_servers:[]}; + yield {type:"assistant",session_id:"native",parent_tool_use_id:mode==="child"?"child":null, + uuid:"assistant",user_message_uuid:mode==="foreign-input"?"foreign":input.uuid,error:"authentication_failed",message:{content:[]}}; + if(mode==="missing-result")throw new Error("unstructured 401 failure"); + if(mode==="recovered")yield {type:"assistant",session_id:"native",parent_tool_use_id:null,uuid:"recovery",user_message_uuid:input.uuid,message:{content:[]}}; + if(mode==="cancel")abort.abort(); + yield {type:"result",uuid:"result",session_id:"native",user_message_uuid:input.uuid, + subtype:mode==="recovered"?"success":mode==="budget"?"error_max_budget_usd":"error_during_execution", + is_error:mode!=="recovered",result:"answer",usage:{input_tokens:2,output_tokens:1},modelUsage:{},total_cost_usd:0.01}; + }}; +}; +await execute({type:"start",input:[{content:[{type:"input_text",text:"hello"}]}],model:"fixture",system_prompt:"",cwd:process.cwd()},async e=>events.push(e),abort); +const last=events.at(-1); +// A nonzero native exit is expected on a structured provider rejection; the bridge +// still retains its validated result and usage. Recovery + nonzero exit is generic. +assert.equal(last.type,"error"); +assert.equal(last.code,mode==="cancel"?"cancelled":"execution_failed"); +assert.equal(last.engine_error_code,mode==="classified"?"authentication_error":undefined); +if(mode==="classified"){assert.equal(last.session_id,"native");assert.equal(last.result_id,"result");} +assert.equal(events.filter(e=>e.type==="usage").length,mode==="missing-result"?0:1); +`; +for (const mode of ["classified","child","foreign-input","missing-result","recovered","budget","cancel"]) { + test(`settled classified SDK failure: ${mode}`,{timeout:15000},()=>{ + const child=spawnSync(process.execPath,["--input-type=module","-e",fixture,mode],{encoding:"utf8",timeout:10000}); + assert.equal(child.status,0,child.stderr||child.error?.message); + }); +} diff --git a/packages/claude-sdk-adapter/tests/native_failure.test.mjs b/packages/claude-sdk-adapter/tests/native_failure.test.mjs new file mode 100644 index 000000000..5496fc976 --- /dev/null +++ b/packages/claude-sdk-adapter/tests/native_failure.test.mjs @@ -0,0 +1,50 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { NativeFailure } from "../dist/native_failure.js"; +import { Inputs } from "../dist/inputs.js"; + +const assistant = (error, extra = {}) => ({ type: "assistant", session_id: "native", parent_tool_use_id: null, error, ...extra }); +const result = (extra = {}) => ({ type: "result", session_id: "native", subtype: "error_during_execution", is_error: true, ...extra }); + +test("pinned SDK finite native mapping", () => { + for (const [native, code] of Object.entries({ authentication_failed:"authentication_error", oauth_org_not_allowed:"authentication_error", + account_on_hold:"authentication_error", verification_required:"authentication_error", cloud_credential_error:"authentication_error", + billing_error:"usage_limit_exceeded", rate_limit:"rate_limit_exceeded", overloaded:"server_overloaded", invalid_request:"invalid_request", + model_not_found:"resource_not_found", server_error:"server_error", unknown:undefined, max_output_tokens:undefined, + future:undefined, toString:undefined, __proto__:undefined })) { + const f = new NativeFailure(); + assert.equal(f.observe(assistant(native), "native", ["input"]), undefined); + assert.equal(f.observe(result(), "native", ["input"]), code); + } +}); + +test("classification requires same root, input and unsuccessful terminal result", () => { + for (const transform of [m=>({...m,parent_tool_use_id:"child"}),m=>({...m,session_id:"other"}),m=>({...m,isReplay:true}),m=>({...m,isSynthetic:true})]) { + const f = new NativeFailure(); f.observe(transform(assistant("rate_limit")),"native",["input"]); + assert.equal(f.observe(result(),"native",["input"]),undefined); + } + for (const terminal of [result({subtype:"success",is_error:false}),result({subtype:"error_max_turns"}),result({subtype:"error_max_budget_usd"}),result({subtype:"error_max_structured_output_retries"})]) { + const f = new NativeFailure(); f.observe(assistant("rate_limit"),"native",["input"]); + assert.equal(f.observe(terminal,"native",["input"]),undefined); + assert.equal(f.observe(result(),"native",["input"]),undefined); + } + const f = new NativeFailure(); + f.observe(assistant("rate_limit"),"native",["old"]); + assert.equal(f.observe(result(),"native",["new"]),undefined); + f.observe(assistant("rate_limit"),"native",["input"]); + f.observe(assistant(undefined),"native",["input"]); + assert.equal(f.observe(result(),"native",["input"]),undefined); + f.observe({type:"system",subtype:"api_retry",session_id:"native",error:"rate_limit"},"native",["retry"]); + assert.equal(f.observe(result(),"native",["retry"]),undefined); +}); + +test("only current admitted input identities reach classification", async () => { + const inputs = new Inputs([{content:[{type:"input_text",text:"hello"}]}]); + const native = (await inputs[Symbol.asyncIterator]().next()).value; + inputs.start("native"); + const good = result({user_message_uuid:native.uuid}); + assert.deepEqual(inputs.pendingInputIDs(good),[native.uuid]); + assert.deepEqual(inputs.pendingInputIDs(result({user_message_uuids:[native.uuid,"foreign"]})),[]); + inputs.consume(good); + assert.deepEqual(inputs.pendingInputIDs(good),[]); +}); diff --git a/services/agents-api/internal/execution/delivery.go b/services/agents-api/internal/execution/delivery.go index 52b6d59dd..8c31ec0c4 100644 --- a/services/agents-api/internal/execution/delivery.go +++ b/services/agents-api/internal/execution/delivery.go @@ -296,6 +296,12 @@ func (d *Dispatcher) deliver(ctx context.Context, tenantID, sessionID string, pe func (r *Result) mergeObservation(env proto.Envelope) error { switch env.Type { + case proto.TypeError: + var failure proto.ErrorPayload + if err := env.DecodePayload(&failure); err != nil { + return err + } + r.EngineErrorCode, r.EngineHTTPStatus = proto.NormalizeEngineFailure(failure.Code, failure.HTTPStatus) case proto.TypeUsage: return env.DecodePayload(&r.Done.Usage) case proto.TypeDone: diff --git a/services/agents-api/internal/execution/dispatcher.go b/services/agents-api/internal/execution/dispatcher.go index ba92fc92c..10aed47a7 100644 --- a/services/agents-api/internal/execution/dispatcher.go +++ b/services/agents-api/internal/execution/dispatcher.go @@ -46,10 +46,12 @@ type Dispatcher struct { } type Result struct { - Done proto.DonePayload `json:"done"` - ErrorCode string `json:"error_code,omitempty"` - Error string `json:"error,omitempty"` - AppliedThrough int64 `json:"applied_through"` + EngineErrorCode string `json:"engine_error_code,omitempty"` + EngineHTTPStatus *int `json:"engine_http_status,omitempty"` + Done proto.DonePayload `json:"done"` + ErrorCode string `json:"error_code,omitempty"` + Error string `json:"error,omitempty"` + AppliedThrough int64 `json:"applied_through"` } // Run claims once before subscribing or sending. Uncertain deliveries are not replayed. diff --git a/services/agents-api/internal/execution/engine_failure_test.go b/services/agents-api/internal/execution/engine_failure_test.go new file mode 100644 index 000000000..d0241e433 --- /dev/null +++ b/services/agents-api/internal/execution/engine_failure_test.go @@ -0,0 +1,75 @@ +package execution + +import ( + "context" + "encoding/json" + "testing" + + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/proto" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" +) + +func TestEngineClassificationSurvivesDrainWithoutChangingSettlement(t *testing.T) { + for _, prior := range []string{"", "event_persistence_failed", "cancel_unconfirmed", "event_stream_incomplete"} { + result := Result{ErrorCode: prior} + writer := &recoveringWriter{} + j := journal{store: writer, next: 1} + events := make(chan proto.Envelope, 3) + failure, _ := proto.NewEnvelope(proto.TypeError, "run", proto.ErrorPayload{Error: "raw native text", Code: "authentication_error"}) + usage, _ := proto.NewEnvelope(proto.TypeUsage, "run", proto.Usage{InputTokens: 7}) + done, _ := proto.NewEnvelope(proto.TypeDone, "run", map[string]any{"metadata": map[string]any{proto.DoneMetaAgentSessionID: "native"}}) + events <- failure + events <- usage + events <- done + close(events) + if err := j.drain(events, &result); err != nil { + t.Fatal(err) + } + if result.ErrorCode != prior || result.EngineErrorCode != "authentication_error" || result.Done.Usage.InputTokens != 7 || result.Done.Metadata[proto.DoneMetaAgentSessionID] != "native" { + t.Fatal(result) + } + reply := cancellationResult{ack: proto.InteractionDecisionAckPayload{Applied: true, Outcome: &proto.DonePayload{Usage: proto.Usage{InputTokens: 9}, Metadata: map[string]any{proto.DoneMetaAgentSessionID: "native"}}}} + if err := recordCancellation(context.Background(), &j, reply, &result); err != nil { + t.Fatal(err) + } + if result.ErrorCode != prior || result.Done.Usage.InputTokens != 9 { + t.Fatal(result) + } + raw, _ := json.Marshal(result) + var persisted Result + if json.Unmarshal(raw, &persisted) != nil || persisted.EngineErrorCode != result.EngineErrorCode { + t.Fatal(string(raw)) + } + unknown := proto.Envelope{Type: proto.TypeError, Payload: json.RawMessage(`{"error":"new","code":{"arbitrary":"value"},"http_status":"secret"}`)} + if err := result.mergeObservation(unknown); err != nil || result.EngineErrorCode != "" || result.ErrorCode != prior || result.Done.Usage.InputTokens != 9 { + t.Fatal(err, result) + } + } +} + +func TestClassifiedErrorCannotOverrideCancellationReceipt(t *testing.T) { + for _, drainFirst := range []bool{true, false} { + result := Result{ErrorCode: "engine_failed"} + j := journal{store: &recoveringWriter{}, next: 1} + failure, _ := proto.NewEnvelope(proto.TypeError, "run", proto.ErrorPayload{Code: "rate_limit_exceeded"}) + events := make(chan proto.Envelope, 1) + events <- failure + close(events) + ack := make(chan cancellationResult, 1) + ack <- cancellationResult{ack: proto.InteractionDecisionAckPayload{Applied: true, Outcome: &proto.DonePayload{Usage: proto.Usage{InputTokens: 9}, Metadata: map[string]any{proto.DoneMetaAgentSessionID: "native"}}}} + if drainFirst { + if err := j.drain(events, &result); err != nil { + t.Fatal(err) + } + } + status := finishDelivery(context.Background(), &j, &result, ack, false, nil) + if !drainFirst { + if err := j.drain(events, &result); err != nil { + t.Fatal(err) + } + } + if status != store.TurnCancelled || result.Done.Usage.InputTokens != 9 || result.Done.Metadata[proto.DoneMetaAgentSessionID] != "native" || result.EngineErrorCode != "rate_limit_exceeded" { + t.Fatalf("receipt lost authority: %s %+v", status, result) + } + } +} From 2264a10777121ef77daedc5cf5d10580eab85320 Mon Sep 17 00:00:00 2001 From: SaladDay Date: Mon, 28 Sep 2026 09:03:26 +0000 Subject: [PATCH 2/3] feat(core): expose safe native failure diagnostics --- apps/docs/content/docs/concepts.mdx | 6 ++ apps/docs/content/guide-sources.json | 4 +- apps/docs/openapi/core-api.yaml | 52 +++++++++++ apps/docs/openapi/sources.json | 4 +- contracts/agents-api/core-errors.md | 20 ++++ contracts/agents-api/core.openapi.yaml | 52 +++++++++++ .../agents-api/native-error-classification.md | 4 +- contracts/agents-api/session-diagnostics.md | 7 +- packages/agents-client/src/index.ts | 2 +- .../src/session-diagnostics.test.ts | 18 ++++ .../agents-client/src/session-diagnostics.ts | 12 ++- .../native_classification_integration_test.go | 92 +++++++++++++++++++ .../internal/api/session_diagnostics.go | 2 +- .../internal/api/turn_diagnostic_failure.go | 29 ++++-- .../api/turn_diagnostic_failure_test.go | 53 +++++++++++ 15 files changed, 338 insertions(+), 19 deletions(-) create mode 100644 services/agents-api/internal/api/native_classification_integration_test.go diff --git a/apps/docs/content/docs/concepts.mdx b/apps/docs/content/docs/concepts.mdx index 9f8505337..38b3434ee 100644 --- a/apps/docs/content/docs/concepts.mdx +++ b/apps/docs/content/docs/concepts.mdx @@ -85,4 +85,10 @@ Do not add product users, RBAC, cross-Project shared assets, administrator execu private-protocol compatibility to this management model. Existing public execution and Runtime ownership rules remain in [CONTRIBUTING.md](https://github.com/MiniMax-AI/parsar-core/blob/f36ac3592bfe9f77a2172680aa52b14c6f525062/CONTRIBUTING.md). +Native failure classification is adapter-owned and uses finite structured native +values. Optional Runtime error metadata is normalized once and retained through +normal delivery and journal draining; it never replaces Core terminal authority, +cancellation receipts, Usage or native identity. See +[native failure classification](https://github.com/MiniMax-AI/parsar-core/blob/f36ac3592bfe9f77a2172680aa52b14c6f525062/contracts/agents-api/native-error-classification.md). + [Repository source](https://github.com/MiniMax-AI/parsar-core/blob/f36ac3592bfe9f77a2172680aa52b14c6f525062/docs/design-principles.md) diff --git a/apps/docs/content/guide-sources.json b/apps/docs/content/guide-sources.json index 135587dfd..dd7c7ae9e 100644 --- a/apps/docs/content/guide-sources.json +++ b/apps/docs/content/guide-sources.json @@ -2,7 +2,7 @@ "revision": "f36ac3592bfe9f77a2172680aa52b14c6f525062", "sources": { "docs/getting-started/README.md": "e4ca3bcbfdcbf3832c1d9e0525f98083ca11a4e42148f9f45090bbe1bc0f2b5c", - "docs/design-principles.md": "5e1bff3cdcc77cd02ad692f67977cb4e0a5654989982d722bac0e3c79dbf82cb", + "docs/design-principles.md": "71285ff0b9a7796481b9fe42f7c4b71d0377c767157980ca6386792d9ca8951b", "docs/web/architecture.md": "80bdb8053c06c2b2030b193aae17d5434545f9c72af50e9d3df3ade522fe25ed", "docs/getting-started/install.md": "15cfe81d4005a7337ac82bf6c35a8b441e9f9b0f2a2299021ad6c539f3f336a3", "docs/configuration.md": "e911757a03fbce4b803a284c8bb40bdb7000ca2f70e27938706dcdcd74bc7f9a", @@ -24,7 +24,7 @@ "outputs": { "content/docs/index.mdx": "75339c35292bb1f3c6b4313d717b82686bee8caa265bb575fcecbd0ba33bb780", "content/docs/index.zh.mdx": "5bdd41893946f1d229eb4cf1599a97f2e38cb9e09722d2a69f62053376ac88a9", - "content/docs/concepts.mdx": "304e18737cdf807562501981e5b8312466a7c7ce29db3bf4823075850dec10c4", + "content/docs/concepts.mdx": "2c2516963cccafed4f66083201a9f84e8d12c4d7f4406ce06179bc26616ea019", "content/docs/concepts.zh.mdx": "c60644f395dbac5ee607cdb053249e05ec8352d28de97ef1c38d6c1e4443bbb1", "content/docs/execution-model.mdx": "8dd05250255fdb39a4aee6ffa51a16379fbd19d2761f510e1ad9cc3a193f750c", "content/docs/execution-model.zh.mdx": "9976a9203c6a83d49c2c753e3a8528a369078e87aeb88850654da948ac8c17e1", diff --git a/apps/docs/openapi/core-api.yaml b/apps/docs/openapi/core-api.yaml index b6d4983be..60db44d0c 100644 --- a/apps/docs/openapi/core-api.yaml +++ b/apps/docs/openapi/core-api.yaml @@ -4357,6 +4357,32 @@ components: api.DiagnosticFailure: properties: code: + enum: + - harness_error + - model_provider_required + - runtime_unavailable + - runtime_disconnected + - runtime_preparation_failed + - execution_interrupted + - delivery_unconfirmed + - input_rejected + - executor_protocol_error + - core_storage_failed + - internal_error + - environment_connection_timeout + - environment_unavailable + - environment_provisioning_failed + - authentication_error + - rate_limit_exceeded + - usage_limit_exceeded + - server_overloaded + - server_error + - invalid_request + - resource_not_found + - request_timeout + - context_length_exceeded + - cyber_policy + - connection_failed type: string failed_at: anyOf: @@ -4667,6 +4693,32 @@ components: api.SessionDiagnosticFailure: properties: code: + enum: + - harness_error + - model_provider_required + - runtime_unavailable + - runtime_disconnected + - runtime_preparation_failed + - execution_interrupted + - delivery_unconfirmed + - input_rejected + - executor_protocol_error + - core_storage_failed + - internal_error + - environment_connection_timeout + - environment_unavailable + - environment_provisioning_failed + - authentication_error + - rate_limit_exceeded + - usage_limit_exceeded + - server_overloaded + - server_error + - invalid_request + - resource_not_found + - request_timeout + - context_length_exceeded + - cyber_policy + - connection_failed type: string failed_at: anyOf: diff --git a/apps/docs/openapi/sources.json b/apps/docs/openapi/sources.json index 6e84dee2a..49e202ccd 100644 --- a/apps/docs/openapi/sources.json +++ b/apps/docs/openapi/sources.json @@ -1,7 +1,7 @@ { "sources": { "contracts/agents-api/openapi.yaml": "a8896e8b67d78ed0608c7cc5636077ec33f4b4895a04bd6cfb8b038afe941e01", - "contracts/agents-api/core.openapi.yaml": "ff8526ba681f7561e49cd8e1818ddb7c6e57ce044021b92d2d69d5b087356900", + "contracts/agents-api/core.openapi.yaml": "8fe158844c170eea78d4b24e30e7caa2adbedd58ffb4a863bdd9c98ecbb50618", "contracts/agents-api/runtime.openapi.yaml": "23852205d3b45892f2de9ecf3d52518c0dd0808962c0c512f219b565aa7ccef2" }, "outputs": { @@ -40,7 +40,7 @@ "content/docs/api-reference/core/vaults.mdx": "12d0aac96cd4a3472ef4eaa287f7945f78862017956e79bfc3ed0ef8e772e78a", "content/docs/api-reference/core/credentials.mdx": "27d586b9a8fd738b9947802f32c80458195370252886bc26abedccf2125ee981", "content/docs/api-reference/core/sandbox-manager.mdx": "6be7c52e057365949f0247c18c710561d680847a955f78f1bbe57825d4d5f549", - "openapi/core-api.yaml": "18e100de2375835350d172e8a472db6966e36f1dd3aaae5e5d8132ba3039fa9c", + "openapi/core-api.yaml": "4d72411e4fc9945aba575a91b9c95e7231262e2749f87796646ef17768022238", "content/docs/api-reference/core/meta.json": "588b163ed2de2100b0f8376688274f80698a6bf6ac037eef56d17c93fa8a5062", "content/docs/api-reference/core/index.mdx": "ddf0a8d87a6c21761beac703a32f2a40486a1c74abb341bd726f40c44a866405", "content/docs/api-reference/machine/sandbox-node.mdx": "5b367a08d43cea82686345d2ba1c75cc90627a7b2970fe42a0df9d77fb3e1e80", diff --git a/contracts/agents-api/core-errors.md b/contracts/agents-api/core-errors.md index 04f9a2932..d895cc245 100644 --- a/contracts/agents-api/core-errors.md +++ b/contracts/agents-api/core-errors.md @@ -121,6 +121,17 @@ inside a successful HTTP 200 snapshot, not the operation error envelope. Public | Code | Stored cause or safe meaning | | --- | --- | | `harness_error` | `engine_failed` without a native classification | +| `authentication_error` | Native provider authentication rejected | +| `rate_limit_exceeded` | Native rate limit classification | +| `usage_limit_exceeded` | Native billing or usage limit classification | +| `server_overloaded` | Native overload classification | +| `server_error` | Native server failure classification | +| `invalid_request` | Native request rejection | +| `resource_not_found` | Native resource/model not found | +| `request_timeout` | Reserved neutral timeout category; no current adapter producer | +| `context_length_exceeded` | Native context limit classification | +| `cyber_policy` | Native cyber policy rejection | +| `connection_failed` | Native connection failure; params contain `http_status`, an integer in 100–599 or null | | `model_provider_required` | Missing frozen model provider | | `runtime_unavailable` | `execution_device_unavailable`, `execution_unavailable` | | `runtime_disconnected` | `device_disconnected`, `event_stream_incomplete` | @@ -139,3 +150,12 @@ inside a successful HTTP 200 snapshot, not the operation error envelope. Public reader is not configured; it has no details. A database failure remains an error, never a healthy or empty diagnostic snapshot. Historical provisioning reasons and private native messages are not parsed for categories or parameters. + +Native categories apply only to a failed Turn with top-level +`error_code: engine_failed`. Only the finite `engine_error_code` allowlist is +accepted; unknown, +malformed and absent metadata retains `harness_error`. Only `connection_failed` +uses `engine_http_status`. Nested metadata and provider prose never classify a +failure. Core persistence, incomplete-stream and cancellation failures retain +priority, and cancelled/completed Turns have no failure. See +[native classification](native-error-classification.md) for adapter coverage. diff --git a/contracts/agents-api/core.openapi.yaml b/contracts/agents-api/core.openapi.yaml index 40879eceb..e6607611a 100644 --- a/contracts/agents-api/core.openapi.yaml +++ b/contracts/agents-api/core.openapi.yaml @@ -253,6 +253,32 @@ definitions: api.DiagnosticFailure: properties: code: + enum: + - harness_error + - model_provider_required + - runtime_unavailable + - runtime_disconnected + - runtime_preparation_failed + - execution_interrupted + - delivery_unconfirmed + - input_rejected + - executor_protocol_error + - core_storage_failed + - internal_error + - environment_connection_timeout + - environment_unavailable + - environment_provisioning_failed + - authentication_error + - rate_limit_exceeded + - usage_limit_exceeded + - server_overloaded + - server_error + - invalid_request + - resource_not_found + - request_timeout + - context_length_exceeded + - cyber_policy + - connection_failed type: string failed_at: type: string @@ -546,6 +572,32 @@ definitions: api.SessionDiagnosticFailure: properties: code: + enum: + - harness_error + - model_provider_required + - runtime_unavailable + - runtime_disconnected + - runtime_preparation_failed + - execution_interrupted + - delivery_unconfirmed + - input_rejected + - executor_protocol_error + - core_storage_failed + - internal_error + - environment_connection_timeout + - environment_unavailable + - environment_provisioning_failed + - authentication_error + - rate_limit_exceeded + - usage_limit_exceeded + - server_overloaded + - server_error + - invalid_request + - resource_not_found + - request_timeout + - context_length_exceeded + - cyber_policy + - connection_failed type: string failed_at: type: string diff --git a/contracts/agents-api/native-error-classification.md b/contracts/agents-api/native-error-classification.md index 45aa7f2f0..25a43f000 100644 --- a/contracts/agents-api/native-error-classification.md +++ b/contracts/agents-api/native-error-classification.md @@ -14,7 +14,9 @@ Normal delivery and terminal journal draining use the same extraction rule. Accepted codes are `authentication_error`, `rate_limit_exceeded`, `usage_limit_exceeded`, `server_overloaded`, `server_error`, `invalid_request`, `resource_not_found`, `request_timeout`, `context_length_exceeded`, `cyber_policy` -and `connection_failed`. Only `connection_failed` retains an integer HTTP status +and `connection_failed`. Core diagnostics expose these categories only for failed +`engine_failed` Turns. Their params are empty except `connection_failed`, whose +`http_status` is a valid integer or null. Only `connection_failed` retains an integer HTTP status in 100..599; all other status metadata is discarded. Missing, malformed and unknown optional values degrade to unclassified metadata without discarding Usage or Done. Old Runtime frames remain generic harness failures. Old readers ignore new fields. diff --git a/contracts/agents-api/session-diagnostics.md b/contracts/agents-api/session-diagnostics.md index 4454eaf80..f0d4fd790 100644 --- a/contracts/agents-api/session-diagnostics.md +++ b/contracts/agents-api/session-diagnostics.md @@ -67,8 +67,11 @@ nullable integer `observed_duration_ms`. The public successful Turn completion time may originate at the native executor; that public timestamp remains unchanged. Public Session, Turn, Item, SSE and -`duration_ms` contracts are unchanged. Subagent diagnostics and native failure -classification are separate work. +`duration_ms` contracts are unchanged. Subagent diagnostics are separate work. +Native failure classification uses only +the finite top-level outcome metadata of a failed `engine_failed` Turn; see +[native classification](native-error-classification.md). Connection failure params +contain `http_status` (100–599 or null); other native categories have empty params. The typed client exposes `retrieveSessionDiagnostics` and `retrieveTurnDiagnostics` on `AdminClient`, retaining request cancellation and diff --git a/packages/agents-client/src/index.ts b/packages/agents-client/src/index.ts index 4831faa2e..af03182c7 100644 --- a/packages/agents-client/src/index.ts +++ b/packages/agents-client/src/index.ts @@ -13,4 +13,4 @@ export { AdminClient } from "./admin-client"; // Skill and SkillVersion come from ./types; the admin projections use the same shapes. export type { AdminClientOptions, AdminProject, CreateAdminProjectInput, RenameAdminProjectInput, AdminAPIKey, AdminIssuedAPIKey, IssueAdminAPIKeyInput, AdminPage, AdminDeleted, SessionArtifact, AdminContent, AdminResourceType, AdminKeyProvenance, AdminResourceOwner, AdminWriteOperation, AdminWriteOperationOptions, AdminWriteOperationPage, AdminSummaryOptions, AdminSummaryEntry, AdminSummary, AdminRuntimeObservation, AdminAuditOptions, AdminAuditResultID, AdminAuditEntry, AdminAuditPage, ExecutorCredential, ExecutorConnection, ExecutorCredentialList, IssueExecutorCredentialInput, IssuedExecutorCredential, CoreProjectReader, CoreInstallation, CoreInstallationConfiguration, CoreInstallationSetting, CoreAddressBindings } from "./admin-types"; -export type { DiagnosticFailureCode, ProvisioningFailureParams, DiagnosticFailure, SessionDiagnosticFailure, SessionDiagnostics, ItemDiagnosticTiming, TurnDiagnostics } from "./session-diagnostics"; +export type { DiagnosticFailureCode, NativeFailureCode, ConnectionFailureParams, ProvisioningFailureParams, DiagnosticFailure, SessionDiagnosticFailure, SessionDiagnostics, ItemDiagnosticTiming, TurnDiagnostics } from "./session-diagnostics"; diff --git a/packages/agents-client/src/session-diagnostics.test.ts b/packages/agents-client/src/session-diagnostics.test.ts index c2c089a4e..f8b69ed60 100644 --- a/packages/agents-client/src/session-diagnostics.test.ts +++ b/packages/agents-client/src/session-diagnostics.test.ts @@ -58,3 +58,21 @@ it.each([ const value = { ...session, failure: { source: "environment", code: "environment_provisioning_failed", params, failed_at: time } }; expect(() => projectSessionDiagnostics(value, id)).toThrow(); }); + +it.each(["authentication_error", "rate_limit_exceeded", "usage_limit_exceeded", "server_overloaded", "server_error", "invalid_request", "resource_not_found", "request_timeout", "context_length_exceeded", "cyber_policy", "connection_failed"])("projects finite native category %s", (code) => { + const params = code === "connection_failed" ? { http_status: 503 } : {}; + const failed = { ...failure, code, params }; + expect(projectTurnDiagnostics({ ...snapshot, failure: failed }, id, turn).failure).toEqual(failed); + expect(projectSessionDiagnostics({ ...session, failure: { ...failed, source: "turn", turn_id: turn } }, id).failure?.code).toBe(code); + expect(() => projectSessionDiagnostics({ ...session, failure: { ...failed, source: "environment_input" } }, id)).toThrow(); +}); +it.each([null, 100, 429, 599])("preserves safe connection status %s", (http_status) => { + const failed = { ...failure, code: "connection_failed", params: { http_status } }; + expect(projectTurnDiagnostics({ ...snapshot, failure: failed }, id, turn).failure).toEqual(failed); +}); +it.each([{}, { http_status: 99 }, { http_status: 600 }, { http_status: 503.5 }, { http_status: "503" }, { http_status: 503, message: "secret-canary" }])("rejects unsafe connection parameters", (params) => { + expect(() => projectTurnDiagnostics({ ...snapshot, failure: { ...failure, code: "connection_failed", params } }, id, turn)).toThrow(); +}); +it("rejects HTTP status parameters on nonconnection categories", () => { + expect(() => projectTurnDiagnostics({ ...snapshot, failure: { ...failure, code: "authentication_error", params: { http_status: 401 } } }, id, turn)).toThrow(); +}); diff --git a/packages/agents-client/src/session-diagnostics.ts b/packages/agents-client/src/session-diagnostics.ts index 3bb20bab6..638ea8e0c 100644 --- a/packages/agents-client/src/session-diagnostics.ts +++ b/packages/agents-client/src/session-diagnostics.ts @@ -1,16 +1,18 @@ import { canonicalUuid, exactFields, isRecord, sameResourceId } from "./response-projection"; import { invalidAdminResponse } from "./admin-projection"; -export type DiagnosticFailureCode = "harness_error" | "model_provider_required" | "runtime_unavailable" | "runtime_disconnected" | "runtime_preparation_failed" | "execution_interrupted" | "delivery_unconfirmed" | "input_rejected" | "executor_protocol_error" | "core_storage_failed" | "internal_error" | "environment_connection_timeout" | "environment_unavailable" | "environment_provisioning_failed"; +export type NativeFailureCode = "authentication_error" | "rate_limit_exceeded" | "usage_limit_exceeded" | "server_overloaded" | "server_error" | "invalid_request" | "resource_not_found" | "request_timeout" | "context_length_exceeded" | "cyber_policy" | "connection_failed"; +export type DiagnosticFailureCode = NativeFailureCode | "harness_error" | "model_provider_required" | "runtime_unavailable" | "runtime_disconnected" | "runtime_preparation_failed" | "execution_interrupted" | "delivery_unconfirmed" | "input_rejected" | "executor_protocol_error" | "core_storage_failed" | "internal_error" | "environment_connection_timeout" | "environment_unavailable" | "environment_provisioning_failed"; export type ProvisioningFailureParams = { step: "setup" | "python" | "npm" | "system" | "file" | "skill" | null; index: number | null; exit_code: number | null }; -export type DiagnosticFailure = { code: DiagnosticFailureCode; params: Record | ProvisioningFailureParams; failed_at: string | null }; +export type ConnectionFailureParams = { http_status: number | null }; +export type DiagnosticFailure = { code: DiagnosticFailureCode; params: Record | ProvisioningFailureParams | ConnectionFailureParams; failed_at: string | null }; export type SessionDiagnosticFailure = DiagnosticFailure & { source: "turn" | "environment" | "environment_input"; turn_id?: string }; export type SessionDiagnostics = { object: "core.session_diagnostics"; session_id: string; status: "idle" | "in_progress" | "requires_action" | "failed"; failure: SessionDiagnosticFailure | null }; /** Core receipt intervals, distinct from the public tool-reported duration_ms. */ export type ItemDiagnosticTiming = { item_id: string; started_at: string; completed_at: string | null; observed_duration_ms: number | null }; export type TurnDiagnostics = { object: "core.turn_diagnostics"; session_id: string; turn_id: string; status: string; failure: DiagnosticFailure | null; items: ItemDiagnosticTiming[]; items_truncated: boolean }; -const turnCodes = new Set(["harness_error", "model_provider_required", "runtime_unavailable", "runtime_disconnected", "runtime_preparation_failed", "execution_interrupted", "delivery_unconfirmed", "input_rejected", "executor_protocol_error", "core_storage_failed", "internal_error"]); +const turnCodes = new Set(["authentication_error", "rate_limit_exceeded", "usage_limit_exceeded", "server_overloaded", "server_error", "invalid_request", "resource_not_found", "request_timeout", "context_length_exceeded", "cyber_policy", "connection_failed", "harness_error", "model_provider_required", "runtime_unavailable", "runtime_disconnected", "runtime_preparation_failed", "execution_interrupted", "delivery_unconfirmed", "input_rejected", "executor_protocol_error", "core_storage_failed", "internal_error"]); const inputCodes = new Set(["environment_connection_timeout", "environment_unavailable", "model_provider_required", "internal_error"]); const turnStatuses = new Set(["queued", "in_progress", "waiting", "completed", "failed", "cancelled"]); const fields = (...values: string[]) => new Set(values); @@ -27,6 +29,10 @@ function failure(value: unknown, codes: Set, session = false): Diagnosti (p.index !== null && (!Number.isSafeInteger(p.index) || (p.index as number) < 0 || p.step !== "setup")) || (p.exit_code !== null && (!Number.isSafeInteger(p.exit_code) || (p.exit_code as number) < 1 || (p.exit_code as number) > 255 || !["setup", "python", "npm", "system"].includes(String(p.step))))) return invalidAdminResponse(); params = { step: p.step as ProvisioningFailureParams["step"], index: p.index as number | null, exit_code: p.exit_code as number | null }; + } else if (value.code === "connection_failed") { + const p = value.params; + if (!exactFields(p, fields("http_status")) || (p.http_status !== null && (!Number.isSafeInteger(p.http_status) || (p.http_status as number) < 100 || (p.http_status as number) > 599))) return invalidAdminResponse(); + params = { http_status: p.http_status as number | null }; } else if (Object.keys(value.params).length !== 0) return invalidAdminResponse(); return { code: value.code as DiagnosticFailureCode, params, failed_at: value.failed_at as string | null }; } diff --git a/services/agents-api/internal/api/native_classification_integration_test.go b/services/agents-api/internal/api/native_classification_integration_test.go new file mode 100644 index 000000000..a6ec85e04 --- /dev/null +++ b/services/agents-api/internal/api/native_classification_integration_test.go @@ -0,0 +1,92 @@ +package api + +import ( + "bytes" + "encoding/json" + "strings" + "testing" + + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/proto" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/execution" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/identity" + "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" + "github.com/google/uuid" +) + +func TestNativeClassificationPostgresRoundTripAndPublicPrivacy(t *testing.T) { + s, pool := diagnosticDatabase(t) + h, _, tenant := adminTestHandler(t, func(h *Handler) { h.store = s }) + for _, code := range []string{"authentication_error", "connection_failed", "secret-canary"} { + t.Run(code, func(t *testing.T) { + session, err := s.CreateSession(t.Context(), tenant, store.CreateSessionInput{Creator: identity.Subject{Kind: "service_account", ID: "native-classification"}, Engine: "codex", IdempotencyKey: uuid.NewString(), Configuration: json.RawMessage(`{"agent":{"id":"agent_root","model":"test"},"environment":{"type":"none"}}`)}) + if err != nil { + t.Fatal(err) + } + receipt, err := s.SubmitMessage(t.Context(), tenant, session.ID, "input", json.RawMessage(`{"text":"test"}`)) + if err != nil { + t.Fatal(err) + } + if _, err = s.TransitionTurn(t.Context(), tenant, session.ID, receipt.TurnID, store.TurnTransition{ExpectedStatus: store.TurnQueued, Status: store.TurnInProgress}); err != nil { + t.Fatal(err) + } + status := 503 + result := execution.Result{ErrorCode: "engine_failed", Error: "Bearer secret-canary https://private.example/key", EngineErrorCode: code, EngineHTTPStatus: &status, Done: proto.DonePayload{Usage: proto.Usage{InputTokens: 7, OutputTokens: 3}, Metadata: map[string]any{proto.DoneMetaAgentSessionID: "native-secret-canary"}}} + outcome, err := json.Marshal(result) + if err != nil { + t.Fatal(err) + } + if _, err = s.TransitionTurn(t.Context(), tenant, session.ID, receipt.TurnID, store.TurnTransition{ExpectedStatus: store.TurnInProgress, Status: store.TurnFailed, Outcome: outcome}); err != nil { + t.Fatal(err) + } + snap, err := s.GetTurnDiagnosticsSnapshot(t.Context(), tenant, session.ID, receipt.TurnID) + if err != nil { + t.Fatal(err) + } + var saved execution.Result + if err = json.Unmarshal(snap.Turn.Outcome, &saved); err != nil || saved.EngineErrorCode != code || saved.EngineHTTPStatus == nil || *saved.EngineHTTPStatus != 503 || saved.Done.Usage.InputTokens != 7 || saved.Done.Usage.OutputTokens != 3 || saved.Done.Metadata[proto.DoneMetaAgentSessionID] != "native-secret-canary" { + t.Fatal("outcome round trip lost facts", err) + } + want := code + if code == "secret-canary" { + want = "harness_error" + } + base := adminSessionsPath + session.ID + for _, path := range []string{base + "/diagnostics", base + "/turns/" + receipt.TurnID + "/diagnostics"} { + w := diagnosticRequest(h, path, "Bearer admin") + if w.Code != 200 || !strings.Contains(w.Body.String(), `"code":"`+want+`"`) || strings.Contains(w.Body.String(), "canary") || strings.Contains(w.Body.String(), "private.example") { + t.Fatal(w.Code, w.Body) + } + if code == "connection_failed" && !strings.Contains(w.Body.String(), `"http_status":503`) { + t.Fatal(w.Body) + } + } + paths := []string{"/v1/agents/sessions/" + session.ID, "/v1/agents/sessions/" + session.ID + "/turns/" + receipt.TurnID, "/v1/agents/sessions/" + session.ID + "/items"} + before := make([][]byte, len(paths)) + for i, path := range paths { + w := diagnosticRequest(h, path, "Bearer caller") + if w.Code != 200 || strings.Contains(w.Body.String(), "canary") || strings.Contains(w.Body.String(), "engine_error") || strings.Contains(w.Body.String(), "http_status") { + t.Fatal(w.Code, w.Body) + } + before[i] = append([]byte(nil), w.Body.Bytes()...) + } + events, err := s.ListSessionEvents(t.Context(), tenant, session.ID, 0) + if err != nil { + t.Fatal(err) + } + eventBytes, _ := json.Marshal(events) + if strings.Contains(string(eventBytes), "canary") || strings.Contains(string(eventBytes), "engine_error") || strings.Contains(string(eventBytes), "http_status") { + t.Fatal("private metadata entered SSE") + } + // Remove only additive private fields. All public response bytes must match. + if _, err = pool.Exec(t.Context(), `UPDATE turns SET outcome=outcome-'engine_error_code'-'engine_http_status' WHERE id=$1`, receipt.TurnID); err != nil { + t.Fatal(err) + } + for i, path := range paths { + w := diagnosticRequest(h, path, "Bearer caller") + if w.Code != 200 || !bytes.Equal(before[i], w.Body.Bytes()) { + t.Fatal("private classification changed public response", path, w.Body) + } + } + }) + } +} diff --git a/services/agents-api/internal/api/session_diagnostics.go b/services/agents-api/internal/api/session_diagnostics.go index f5a6231a1..47033ad00 100644 --- a/services/agents-api/internal/api/session_diagnostics.go +++ b/services/agents-api/internal/api/session_diagnostics.go @@ -10,7 +10,7 @@ import ( ) type DiagnosticFailure struct { - Code string `json:"code"` + Code string `json:"code" enums:"harness_error,model_provider_required,runtime_unavailable,runtime_disconnected,runtime_preparation_failed,execution_interrupted,delivery_unconfirmed,input_rejected,executor_protocol_error,core_storage_failed,internal_error,environment_connection_timeout,environment_unavailable,environment_provisioning_failed,authentication_error,rate_limit_exceeded,usage_limit_exceeded,server_overloaded,server_error,invalid_request,resource_not_found,request_timeout,context_length_exceeded,cyber_policy,connection_failed"` Params CoreErrorDetails `json:"params" swaggertype:"object"` FailedAt *time.Time `json:"failed_at" extensions:"x-nullable"` } diff --git a/services/agents-api/internal/api/turn_diagnostic_failure.go b/services/agents-api/internal/api/turn_diagnostic_failure.go index 31992df92..05623019d 100644 --- a/services/agents-api/internal/api/turn_diagnostic_failure.go +++ b/services/agents-api/internal/api/turn_diagnostic_failure.go @@ -3,6 +3,7 @@ package api import ( "encoding/json" + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/proto" "github.com/MiniMax-AI-Dev/parsar/services/agents-api/internal/store" ) @@ -10,16 +11,30 @@ func turnDiagnosticFailure(turn store.Turn) *DiagnosticFailure { if turn.Status != store.TurnFailed { return nil } - var outcome struct { - Code string `json:"error_code"` - } - if json.Unmarshal(turn.Outcome, &outcome) != nil { - outcome.Code = "" + var outcome map[string]json.RawMessage + var coreCode string + if json.Unmarshal(turn.Outcome, &outcome) == nil { + _ = json.Unmarshal(outcome["error_code"], &coreCode) } code := "internal_error" - switch outcome.Code { + params := CoreErrorDetails{} + switch coreCode { case "engine_failed": code = "harness_error" + // Optional malformed metadata cannot hide the authoritative Core error. + var nativeCode string + var nativeStatus *int + _ = json.Unmarshal(outcome["engine_error_code"], &nativeCode) + _ = json.Unmarshal(outcome["engine_http_status"], &nativeStatus) + if classified, status := proto.NormalizeEngineFailure(nativeCode, nativeStatus); classified != "" { + code = classified + if code == "connection_failed" { + params["http_status"] = CoreErrorNull() + if status != nil { + params["http_status"] = CoreErrorNumber(float64(*status)) + } + } + } case "model_provider_required": code = "model_provider_required" case "execution_device_unavailable", "execution_unavailable": @@ -39,5 +54,5 @@ func turnDiagnosticFailure(turn store.Turn) *DiagnosticFailure { case "event_persistence_failed", "artifact_capture_failed": code = "core_storage_failed" } - return &DiagnosticFailure{Code: code, Params: CoreErrorDetails{}, FailedAt: diagnosticTime(turn.CompletedAt)} + return &DiagnosticFailure{Code: code, Params: params, FailedAt: diagnosticTime(turn.CompletedAt)} } diff --git a/services/agents-api/internal/api/turn_diagnostic_failure_test.go b/services/agents-api/internal/api/turn_diagnostic_failure_test.go index fe719d556..b8d015eb0 100644 --- a/services/agents-api/internal/api/turn_diagnostic_failure_test.go +++ b/services/agents-api/internal/api/turn_diagnostic_failure_test.go @@ -42,3 +42,56 @@ func TestDiagnosticFailureWhitelist(t *testing.T) { } } } + +func TestDiagnosticNativeClassification(t *testing.T) { + codes := []string{"authentication_error", "rate_limit_exceeded", "usage_limit_exceeded", "server_overloaded", "server_error", "invalid_request", "resource_not_found", "request_timeout", "context_length_exceeded", "cyber_policy", "connection_failed"} + catalog, err := os.ReadFile("../../../../contracts/agents-api/core-errors.md") + if err != nil { + t.Fatal(err) + } + for _, code := range codes { + t.Run(code, func(t *testing.T) { + if !strings.Contains(string(catalog), "`"+code+"`") { + t.Fatal("uncatalogued code", code) + } + for _, status := range []any{nil, 100, 429, 599, 99, 600, 503.5, "503", true, map[string]any{"token": "secret-canary"}} { + raw, _ := json.Marshal(map[string]any{"error_code": "engine_failed", "error": "secret-canary", "engine_error_code": code, "engine_http_status": status}) + got := turnDiagnosticFailure(store.Turn{Status: store.TurnFailed, Outcome: raw}) + encoded, _ := json.Marshal(got) + if got.Code != code || strings.Contains(string(encoded), "canary") { + t.Fatal(string(encoded)) + } + params, _ := json.Marshal(got.Params) + want := `{}` + if code == "connection_failed" { + want = `{"http_status":null}` + if n, ok := status.(int); ok && n >= 100 && n <= 599 { + b, _ := json.Marshal(map[string]int{"http_status": n}) + want = string(b) + } + } + if string(params) != want { + t.Fatalf("status=%v params=%s want=%s", status, params, want) + } + } + }) + } + for _, optional := range []string{``, `,"engine_error_code":null`, `,"engine_error_code":17`, `,"engine_error_code":{"code":"authentication_error"}`, `,"engine_error_code":"secret-canary"`, `,"done":{"engine_error_code":"authentication_error"}`, `,"Engine_Error_Code":"authentication_error"`} { + got := turnDiagnosticFailure(store.Turn{Status: store.TurnFailed, Outcome: json.RawMessage(`{"error_code":"engine_failed"` + optional + `}`)}) + if got.Code != "harness_error" || len(got.Params) != 0 { + t.Fatal(optional, got) + } + } + for core, want := range map[string]string{"event_persistence_failed": "core_storage_failed", "artifact_capture_failed": "core_storage_failed", "event_stream_incomplete": "runtime_disconnected", "cancel_unconfirmed": "delivery_unconfirmed", "invalid_executor_result": "executor_protocol_error", "secret-canary": "internal_error"} { + raw, _ := json.Marshal(map[string]any{"error_code": core, "engine_error_code": "connection_failed", "engine_http_status": 503}) + got := turnDiagnosticFailure(store.Turn{Status: store.TurnFailed, Outcome: raw}) + if got.Code != want || len(got.Params) != 0 { + t.Fatal(got) + } + } + for _, status := range []string{store.TurnQueued, store.TurnInProgress, store.TurnWaiting, store.TurnCompleted, store.TurnCancelled} { + if got := turnDiagnosticFailure(store.Turn{Status: status, Outcome: json.RawMessage(`{"error_code":"engine_failed","engine_error_code":"authentication_error"}`)}); got != nil { + t.Fatal(status, got) + } + } +} From 3e4bebbff4b682d0f46b328c7a82a5412cb50c96 Mon Sep 17 00:00:00 2001 From: SaladDay Date: Mon, 28 Sep 2026 09:05:26 +0000 Subject: [PATCH 3/3] fix(runtime): reject nonstring native failure classifications --- packages/claude-sdk-adapter/src/native_failure.ts | 2 +- .../claude-sdk-adapter/tests/native_failure.test.mjs | 9 +++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/packages/claude-sdk-adapter/src/native_failure.ts b/packages/claude-sdk-adapter/src/native_failure.ts index 436a7aa32..cd7631f77 100644 --- a/packages/claude-sdk-adapter/src/native_failure.ts +++ b/packages/claude-sdk-adapter/src/native_failure.ts @@ -18,7 +18,7 @@ export class NativeFailure { if (!sessionID || message.session_id !== sessionID || !pending.length || ("isReplay" in message && message.isReplay) || ("isSynthetic" in message && message.isSynthetic)) return; if (message.type === "assistant" && message.parent_tool_use_id === null) { - const code = message.error === undefined || !Object.hasOwn(codes, message.error) ? undefined : codes[message.error]; + const code = typeof message.error !== "string" || !Object.hasOwn(codes, message.error) ? undefined : codes[message.error]; for (const id of pending) { this.candidates.delete(id); if (code) this.candidates.set(id, code); diff --git a/packages/claude-sdk-adapter/tests/native_failure.test.mjs b/packages/claude-sdk-adapter/tests/native_failure.test.mjs index 5496fc976..cf87b7d99 100644 --- a/packages/claude-sdk-adapter/tests/native_failure.test.mjs +++ b/packages/claude-sdk-adapter/tests/native_failure.test.mjs @@ -48,3 +48,12 @@ test("only current admitted input identities reach classification", async () => inputs.consume(good); assert.deepEqual(inputs.pendingInputIDs(good),[]); }); + + +test("malformed native classifications cannot coerce to a finite code", () => { + for (const malformed of [["rate_limit"], { toString: () => "authentication_failed" }, null, 429, true]) { + const f = new NativeFailure(); + f.observe(assistant(malformed), "native", ["input"]); + assert.equal(f.observe(result(), "native", ["input"]), undefined); + } +});