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/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/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 new file mode 100644 index 000000000..25a43f000 --- /dev/null +++ b/contracts/agents-api/native-error-classification.md @@ -0,0 +1,50 @@ +# 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`. 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. +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/contracts/agents-api/session-diagnostics.md b/contracts/agents-api/session-diagnostics.md index e8b29a5c9..cfd8e1c6d 100644 --- a/contracts/agents-api/session-diagnostics.md +++ b/contracts/agents-api/session-diagnostics.md @@ -69,8 +69,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/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/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 01b99fd88..d8ef7b405 100644 --- a/packages/agents-client/src/session-diagnostics.test.ts +++ b/packages/agents-client/src/session-diagnostics.test.ts @@ -59,6 +59,24 @@ it.each([ 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(); +}); + const provisioning = { source: "environment", code: "environment_provisioning_failed", params: { step: "setup", index: null, exit_code: null }, failed_at: null }; it.each([ ["Session status", () => projectSessionDiagnostics({ ...session, status: ["failed"], failure: null }, id)], diff --git a/packages/agents-client/src/session-diagnostics.ts b/packages/agents-client/src/session-diagnostics.ts index 3b3845c9f..c8af4468e 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(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/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..cd7631f77 --- /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 = 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); + } + } 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..cf87b7d99 --- /dev/null +++ b/packages/claude-sdk-adapter/tests/native_failure.test.mjs @@ -0,0 +1,59 @@ +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),[]); +}); + + +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); + } +}); 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 16d71a245..ea327a0b3 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) + } + } +} 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) + } + } +}