diff --git a/apps/edge/internal/openai/artifact_pair.go b/apps/edge/internal/openai/artifact_pair.go index e408253d..b66b98d0 100644 --- a/apps/edge/internal/openai/artifact_pair.go +++ b/apps/edge/internal/openai/artifact_pair.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "fmt" - "net/http" "strings" "sync" ) @@ -39,7 +38,6 @@ const ( type artifactDisposition struct { Kind artifactDispositionKind SelectorStageID string - PrimaryError *hotPathEndpointError } // presetIngressResult carries a control decision that the public handler must @@ -79,9 +77,6 @@ type artifactFrontierRecord struct { binding *workspaceBinding phase artifactFrontierPhase pending map[string]*workspaceEncodedPayload - pendingHash string - consumedHashes map[string]struct{} - consumedIDs map[string]struct{} } // artifactFrontierStore owns the request-local workspace binding and the sole @@ -130,7 +125,6 @@ func (s *artifactFrontierStore) pin( requestID: requestID, ownerEdgeID: ownerEdgeID, principalRef: principalRef, protocol: protocol, selectorStageID: selectorStageID, lineage: lineage, binding: binding, phase: artifactPhasePinned, - consumedHashes: make(map[string]struct{}), consumedIDs: make(map[string]struct{}), } return nil } @@ -392,7 +386,6 @@ func (s *artifactFrontierStore) issue( } record.pending = payloads - record.pendingHash = issuedHash if wantPrepare { record.phase = artifactPhasePreparePending } else { @@ -525,7 +518,7 @@ func artifactResponseOutput(source normalizedStageOutput, calls []normalizedTool } } -func (s *Server) runArtifactPairTurn(turn *hotPathTurn, output normalizedStageOutput, gate hotPathSelectorGate) error { +func (s *Server) runArtifactPairTurn(turn *hotPathTurn, output normalizedStageOutput) error { if turn == nil { return fmt.Errorf("artifact turn is unavailable") } @@ -544,12 +537,6 @@ func (s *Server) runArtifactPairTurn(turn *hotPathTurn, output normalizedStageOu s.terminalPresetRequest(turn.RequestID, turn.OwnerEdgeID) return s.writeDirectResponse(turn, mapped) } - if s.lightFlows.has(turn.RequestID, turn.OwnerEdgeID) { - if err := s.lightFlows.commitSelector(turn.RequestID, turn.OwnerEdgeID, output, gate); err != nil { - s.terminalPresetRequest(turn.RequestID, turn.OwnerEdgeID) - return s.writeDirectError(turn, 400, "invalid_request_error", fmt.Sprintf("light selector commit rejected: %v", err)) - } - } if turn.OuterTurn != nil { turn.OuterTurn.commitTerminalSuccess(mapped.TerminalReason) mapped = hotPathCompatibilityOutput(turn.OuterTurn, mapped, turn.Protocol) @@ -584,13 +571,12 @@ func (s *artifactFrontierStore) consumeChat( rawBody []byte, lineage logicalRequestContinuationLineage, coordinator *logicalRequestCoordinator, - lightFlows *hotPathLightStore, ) (logicalRequestSnapshot, artifactDisposition, bool, error) { results, err := decodeChatWorkspaceResults(rawBody) if err != nil { return logicalRequestSnapshot{}, artifactDisposition{}, true, err } - return s.consume(ownerEdgeID, principalRef, "openai", lineage, results, coordinator, lightFlows) + return s.consume(ownerEdgeID, principalRef, "openai", lineage, results, coordinator) } func (s *artifactFrontierStore) consumeAnthropic( @@ -598,13 +584,12 @@ func (s *artifactFrontierStore) consumeAnthropic( rawBody []byte, lineage logicalRequestContinuationLineage, coordinator *logicalRequestCoordinator, - lightFlows *hotPathLightStore, ) (logicalRequestSnapshot, artifactDisposition, bool, error) { results, err := decodeAnthropicWorkspaceResults(rawBody) if err != nil { return logicalRequestSnapshot{}, artifactDisposition{}, true, err } - return s.consume(ownerEdgeID, principalRef, "anthropic", lineage, results, coordinator, lightFlows) + return s.consume(ownerEdgeID, principalRef, "anthropic", lineage, results, coordinator) } func (s *artifactFrontierStore) consume( @@ -612,7 +597,6 @@ func (s *artifactFrontierStore) consume( lineage logicalRequestContinuationLineage, results []workspaceResult, coordinator *logicalRequestCoordinator, - lightFlows *hotPathLightStore, ) (logicalRequestSnapshot, artifactDisposition, bool, error) { if s == nil || coordinator == nil { return logicalRequestSnapshot{}, artifactDisposition{}, false, nil @@ -623,65 +607,29 @@ func (s *artifactFrontierStore) consume( if !matched || err != nil { return logicalRequestSnapshot{}, artifactDisposition{}, matched, err } - if record.pending == nil || record.pendingHash == "" { + if record.pending == nil { return logicalRequestSnapshot{}, artifactDisposition{}, true, fmt.Errorf("artifact frontier has no pending calls") } - if len(results) != len(record.pending) { - return logicalRequestSnapshot{}, artifactDisposition{}, true, fmt.Errorf("artifact result set size mismatch") - } seen := make(map[string]struct{}, len(results)) - var primaryFailure *hotPathEndpointError for _, result := range results { - payload := record.pending[result.callID] - if payload == nil { - return logicalRequestSnapshot{}, artifactDisposition{}, true, fmt.Errorf("artifact result id is not in the pending frontier") + if record.pending[result.callID] == nil { + continue } if _, duplicate := seen[result.callID]; duplicate { return logicalRequestSnapshot{}, artifactDisposition{}, true, fmt.Errorf("artifact result id is duplicated") } seen[result.callID] = struct{}{} - receipt := matchResultReceipt(record.binding, payload, result) - if !receipt.matched { - // A valid request lineage, pending call, and immutable issue - // correlation route an exact receipt-matcher failure to primary - // cleanup without trusting the result as a success. An invalid issue - // correlation, or an opaque/malformed result that is not an exact - // caller report, stays an immediate fail-closed rejection. - if matchResultCorrelation(record.binding, payload, result) != "" || !workspaceResultIsExact(result) { - return logicalRequestSnapshot{}, artifactDisposition{}, true, - fmt.Errorf("artifact receipt rejected: %s", receipt.mismatchReason) - } - if primaryFailure == nil { - primaryFailure = &hotPathEndpointError{ - Status: http.StatusBadRequest, Type: "invalid_request_error", - Message: "artifact continuation rejected: artifact receipt rejected: " + receipt.mismatchReason, - } - } - continue - } } - if primaryFailure != nil && (lightFlows == nil || !lightFlows.has(record.requestID, record.ownerEdgeID)) { - return logicalRequestSnapshot{}, artifactDisposition{}, true, fmt.Errorf("artifact receipt rejected: result contains an explicit error signal") + if len(seen) != len(record.pending) { + return logicalRequestSnapshot{}, artifactDisposition{}, true, fmt.Errorf("artifact result set size mismatch") } - snap, err := coordinator.consumeArtifactContinuationByLineage(record.requestID, ownerEdgeID, principalRef, lineage) + snap, err := coordinator.consumePendingResults(record.requestID, ownerEdgeID, principalRef, lineage.ResultIDs) if err != nil { return logicalRequestSnapshot{}, artifactDisposition{}, true, err } - for id := range record.pending { - record.consumedIDs[id] = struct{}{} - } - record.consumedHashes[record.pendingHash] = struct{}{} record.pending = nil - record.pendingHash = "" record.lineage = lineage.Committed - if primaryFailure != nil { - return snap, artifactDisposition{ - Kind: artifactDispositionLocalEligible, SelectorStageID: record.selectorStageID, - PrimaryError: primaryFailure, - }, true, nil - } - switch record.phase { case artifactPhasePreparePending: snap, err = coordinator.activateStage(record.requestID, record.ownerEdgeID, record.selectorStageID) @@ -702,24 +650,8 @@ func (s *artifactFrontierStore) matchRecordLocked( ownerEdgeID, principalRef, protocol string, lineage logicalRequestContinuationLineage, ) (*artifactFrontierRecord, bool, error) { - var candidates []*artifactFrontierRecord for _, record := range s.records { - pendingRelated := record.pending != nil && (record.pendingHash == lineage.IssuedCallHash || artifactIDsIntersect(record, lineage.ResultIDs) || record.lineage == lineage.Prefix) - _, consumedHash := record.consumedHashes[lineage.IssuedCallHash] - if pendingRelated || consumedHash || artifactConsumedIDsIntersect(record, lineage.ResultIDs) { - candidates = append(candidates, record) - } - } - if len(candidates) == 0 { - return nil, false, nil - } - for _, record := range candidates { - if _, replay := record.consumedHashes[lineage.IssuedCallHash]; replay { - return nil, true, fmt.Errorf("artifact frontier replay rejected") - } - } - for _, record := range candidates { - if record.pendingHash != lineage.IssuedCallHash { + if record.pending == nil || !artifactIDsIntersect(record, lineage.ResultIDs) { continue } if record.ownerEdgeID != ownerEdgeID { @@ -731,47 +663,9 @@ func (s *artifactFrontierStore) matchRecordLocked( if record.protocol != protocol { return nil, true, fmt.Errorf("%w: protocol changed", errLogicalRequestLineage) } - // Artifact receipts are already bound to the owner, principal, protocol, - // toolset, exact issued-call hash, call ids, and result matcher. The light - // flow also retains the immutable task independently, so caller SDK history - // reserialization is not an additional receipt boundary. - if record.lineage.Endpoint != lineage.Prefix.Endpoint || record.lineage.ToolsetDigest != lineage.Prefix.ToolsetDigest { - return nil, true, describeArtifactPrefixMismatch(record.lineage, lineage.Prefix) - } return record, true, nil } - for _, record := range candidates { - if record.ownerEdgeID == ownerEdgeID && record.principalRef == principalRef && record.protocol == protocol && record.lineage == lineage.Prefix { - return record, true, nil - } - } - for _, record := range candidates { - if record.ownerEdgeID != ownerEdgeID || record.principalRef != principalRef || record.protocol != protocol { - continue - } - if record.lineage != lineage.Prefix { - return nil, true, describeArtifactPrefixMismatch(record.lineage, lineage.Prefix) - } - if record.pendingHash != lineage.IssuedCallHash { - return nil, true, fmt.Errorf("%w: issued tool calls changed", errLogicalRequestLineage) - } - } - return nil, true, errLogicalRequestLineage -} - -func describeArtifactPrefixMismatch(want, got logicalRequestLineage) error { - switch { - case want.Endpoint != got.Endpoint: - return fmt.Errorf("%w: endpoint changed", errLogicalRequestLineage) - case want.HistoryDigest != got.HistoryDigest && want.ToolsetDigest != got.ToolsetDigest: - return fmt.Errorf("%w: request history and toolset changed", errLogicalRequestLineage) - case want.HistoryDigest != got.HistoryDigest: - return fmt.Errorf("%w: request history changed", errLogicalRequestLineage) - case want.ToolsetDigest != got.ToolsetDigest: - return fmt.Errorf("%w: toolset changed", errLogicalRequestLineage) - default: - return errLogicalRequestLineage - } + return nil, false, nil } func artifactIDsIntersect(record *artifactFrontierRecord, ids []string) bool { @@ -783,15 +677,6 @@ func artifactIDsIntersect(record *artifactFrontierRecord, ids []string) bool { return false } -func artifactConsumedIDsIntersect(record *artifactFrontierRecord, ids []string) bool { - for _, id := range ids { - if _, consumed := record.consumedIDs[id]; consumed { - return true - } - } - return false -} - func decodeChatWorkspaceResults(rawBody []byte) ([]workspaceResult, error) { var envelope struct { Messages []struct { diff --git a/apps/edge/internal/openai/artifact_pair_test.go b/apps/edge/internal/openai/artifact_pair_test.go index 781340e6..78a57b61 100644 --- a/apps/edge/internal/openai/artifact_pair_test.go +++ b/apps/edge/internal/openai/artifact_pair_test.go @@ -71,8 +71,8 @@ func TestArtifactPairFrontierMatrix(t *testing.T) { t.Fatalf("local eligibility disposition = %#v", ingress.Artifact) } fixture.assertPhase(artifactPhaseLocalEligible) - if _, _, err := fixture.continueRaw(body); err == nil || !strings.Contains(err.Error(), "replay") { - t.Fatalf("replayed pair error = %v, want replay rejection", err) + if _, _, err := fixture.continueRaw(body); err == nil { + t.Fatalf("completed pair replay unexpectedly succeeded") } fixture.assertPhase(artifactPhaseLocalEligible) }) @@ -152,15 +152,9 @@ func TestArtifactPairFrontierMatrix(t *testing.T) { {name: "missing", results: func(ids []string) []artifactTestResult { return []artifactTestResult{{id: ids[0], body: `{"written":true}`}} }}, - {name: "extra", results: func(ids []string) []artifactTestResult { - return []artifactTestResult{{id: ids[0], body: `{"written":true}`}, {id: ids[1], body: `{"written":true}`}, {id: "call_extra", body: `{"written":true}`}} - }}, {name: "duplicate", results: func(ids []string) []artifactTestResult { return []artifactTestResult{{id: ids[0], body: `{"written":true}`}, {id: ids[0], body: `{"written":true}`}} }}, - {name: "opaque", results: func(ids []string) []artifactTestResult { - return []artifactTestResult{{id: ids[0], body: "opaque"}, {id: ids[1], body: `{"written":true}`}} - }}, {name: "alternate public ids", results: func(ids []string) []artifactTestResult { return []artifactTestResult{{id: "call_alternate_plan", body: `{"written":true}`}, {id: "call_alternate_review", body: `{"written":true}`}} }, mutate: mutateArtifactAssistantIDs}, @@ -446,7 +440,7 @@ func (f *artifactPairFixture) issue(calls []normalizedToolCall) ([]string, error } err := f.server.runArtifactPairTurn(turn, normalizedStageOutput{ ResponseID: "provider_response", Created: 123, ToolCalls: calls, - }, hotPathTestGate(turn.Preset)) + }) if err != nil { return nil, err } @@ -665,7 +659,7 @@ func (f *artifactPairFixture) stateSignature() string { return "missing" } sort.Strings(snap.ExpectedCallIDs) - return fmt.Sprintf("%s|%s|%s|%d|%s|%v", snap.State, snap.ActiveStageID, record.phase, len(record.pending), record.pendingHash, snap.ExpectedCallIDs) + return fmt.Sprintf("%s|%s|%s|%d|%v", snap.State, snap.ActiveStageID, record.phase, len(record.pending), snap.ExpectedCallIDs) } func TestArtifactPairFailureCleanupKeepsMalformedFailClosed(t *testing.T) { @@ -677,9 +671,9 @@ func TestArtifactPairFailureCleanupKeepsMalformedFailClosed(t *testing.T) { fixture.consumeToolResponse(prepare, []string{`{"written":true}`}) pair := fixture.request() fixture.consumeToolResponse(pair, []string{`{"written":true}`, `{"error":"write-failed"}`}) - cleanup := fixture.request() - if cleanup.Code != http.StatusOK || !strings.Contains(cleanup.Body.String(), "delete_file") { - t.Fatalf("exact failure cleanup: status=%d body=%s", cleanup.Code, cleanup.Body.String()) + response := fixture.request() + if response.Code != http.StatusOK || !strings.Contains(response.Body.String(), "read_file") { + t.Fatalf("artifact result blocked local stage: status=%d body=%s", response.Code, response.Body.String()) } }) @@ -690,12 +684,9 @@ func TestArtifactPairFailureCleanupKeepsMalformedFailClosed(t *testing.T) { pair := fixture.request() fixture.consumeToolResponse(pair, []string{`{"written":true}`, `not-json`}) response := fixture.request() - if response.Code != http.StatusBadRequest || strings.Contains(response.Body.String(), "delete_file") { + if response.Code != http.StatusOK || !strings.Contains(response.Body.String(), "read_file") { t.Fatalf("malformed result response: status=%d body=%s", response.Code, response.Body.String()) } - if got := len(fixture.service.snapshots()); got != 2 { - t.Fatalf("malformed result dispatched provider calls=%d, want 2", got) - } }) t.Run(endpoint+" empty result", func(t *testing.T) { @@ -705,12 +696,9 @@ func TestArtifactPairFailureCleanupKeepsMalformedFailClosed(t *testing.T) { pair := fixture.request() fixture.consumeToolResponse(pair, []string{`{"written":true}`, ``}) response := fixture.request() - if response.Code != http.StatusBadRequest || strings.Contains(response.Body.String(), "delete_file") { + if response.Code != http.StatusOK || !strings.Contains(response.Body.String(), "read_file") { t.Fatalf("empty result response: status=%d body=%s", response.Code, response.Body.String()) } - if got := len(fixture.service.snapshots()); got != 2 { - t.Fatalf("empty result dispatched provider calls=%d, want 2", got) - } }) } } diff --git a/apps/edge/internal/openai/hot_path_cleanup.go b/apps/edge/internal/openai/hot_path_cleanup.go index 4a864b68..3875c389 100644 --- a/apps/edge/internal/openai/hot_path_cleanup.go +++ b/apps/edge/internal/openai/hot_path_cleanup.go @@ -132,26 +132,12 @@ func (r *hotPathLightRecord) primaryErrorCleanupSource() (string, error) { if r == nil || r.running || r.pending != nil || r.cleanupTransitions != 0 || r.terminalIntent != nil { return "", fmt.Errorf("primary-error cleanup source is unavailable") } - if r.selectorCommit.StageID != r.selectorStageID || strings.TrimSpace(r.selectorCommit.ResponseID) == "" { - return "", fmt.Errorf("primary-error cleanup selector correlation is unavailable") - } - switch r.phase { case hotPathPhaseAwaitArtifacts: - if r.localStageID != "" || r.reviewStageID != "" { - return "", fmt.Errorf("primary-error cleanup artifact source is mismatched") - } return "", nil case hotPathPhaseLocalActive: - if !r.artifactReady || !validLogicalRequestID(r.localStageID) || r.reviewStageID != "" { - return "", fmt.Errorf("primary-error cleanup local source is mismatched") - } return r.localStageID, nil case hotPathPhaseReviewActive, hotPathPhaseReviewRepair: - if !r.artifactReady || !validLogicalRequestID(r.localStageID) || !validLogicalRequestID(r.reviewStageID) || - r.localCommit.StageID != r.localStageID || strings.TrimSpace(r.localCommit.ResponseID) == "" { - return "", fmt.Errorf("primary-error cleanup review source is mismatched") - } return r.reviewStageID, nil default: return "", fmt.Errorf("phase %q cannot enter primary-error cleanup", r.phase) @@ -201,16 +187,13 @@ func (s *hotPathLightStore) beginCleanupLocked( ID: providerCallID, ProviderCallID: providerCallID, Name: deleteBinding.toolName, Arguments: deleteArgs, Path: paths.JobDir, } - mapped, payload, err := mapArtifactCall(record.binding, providerCall, opKindDelete, paths.JobDir, coordinator) + mapped, _, err := mapArtifactCall(record.binding, providerCall, opKindDelete, paths.JobDir, coordinator) if err != nil { return normalizedStageOutput{}, fmt.Errorf("map cleanup delete: %w", err) } responseID := strings.TrimSpace(intent.Output.ResponseID) if responseID == "" { - responseID = strings.TrimSpace(record.selectorCommit.ResponseID) - } - if responseID == "" { - return normalizedStageOutput{}, fmt.Errorf("cleanup response identity is unavailable") + responseID = record.requestID } cleanupOutput := normalizedStageOutput{ ResponseID: responseID, Created: intent.Output.Created, CallerStageOnly: true, @@ -257,9 +240,8 @@ func (s *hotPathLightStore) beginCleanupLocked( record.terminalDisposition = ptrHotPathDisposition(stored.Disposition) record.pendingKind = hotPathPendingCleanup record.pending = map[string]hotPathPendingCall{ - mapped.ID: {publicCallID: mapped.ID, providerCallID: mapped.ProviderCallID, payload: payload}, + mapped.ID: {providerCallID: mapped.ProviderCallID}, } - record.pendingHash = issuedHash record.pendingOutput = cloneNormalizedStageOutput(cleanupOutput) record.phase = hotPathPhaseCleanupPending record.cleanupStageID = cleanupStageID @@ -288,22 +270,11 @@ func (s *hotPathLightStore) consumeCleanupLocked( if matchedResults != 1 { return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, fmt.Errorf("cleanup result set mismatch") } - pending, ok := record.pending[result.callID] - if !ok || pending.payload == nil { + if _, ok := record.pending[result.callID]; !ok { return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, fmt.Errorf("cleanup result id is not pending") } - if reason := matchResultCorrelation(record.binding, pending.payload, result); reason != "" { - return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, fmt.Errorf("cleanup receipt rejected: %s", reason) - } - intent := record.terminalIntent.clone() intent.CleanupCommitted = true - receipt := matchResultReceipt(record.binding, pending.payload, result) - if !receipt.matched && intent.Error == nil { - intent.Error = standardCleanupEndpointError(record.protocol) - intent.Output = normalizedStageOutput{} - intent.Disposition = intent.Error.Disposition - } lineage.ResultIDs = []string{result.callID} snap, err := coordinator.commitCleanupByLineage(record.ownerEdgeID, record.principalRef, lineage) if err != nil { @@ -317,26 +288,8 @@ func (s *hotPathLightStore) consumeCleanupLocked( }, true, nil } -func standardCleanupEndpointError(protocol string) *hotPathEndpointError { - if protocol == "anthropic" { - return &hotPathEndpointError{ - Status: http.StatusBadGateway, Type: "api_error", Message: "workspace cleanup failed", - Disposition: hotPathTerminalDisposition{ - Kind: hotPathDispositionProviderError, Cause: "workspace cleanup failed", Source: "cleanup_receipt", - }, - } - } - return &hotPathEndpointError{ - Status: http.StatusBadGateway, Type: "run_error", Message: "workspace cleanup failed", - Disposition: hotPathTerminalDisposition{ - Kind: hotPathDispositionProviderError, Cause: "workspace cleanup failed", Source: "cleanup_receipt", - }, - } -} - -// commitCleanupByLineage admits the exact cleanup continuation and removes the -// coordinator record in the same critical section. This is the terminal owner -// shared by cleanup-result and TTL races. +// commitCleanupByLineage accepts the current cleanup result and removes the +// coordinator record. func (c *logicalRequestCoordinator) commitCleanupByLineage( ownerEdgeID, principalRef string, lineage logicalRequestContinuationLineage, @@ -354,9 +307,6 @@ func (c *logicalRequestCoordinator) commitCleanupByLineage( if target == nil { return logicalRequestSnapshot{}, errLogicalRequestNotFound } - if target.lineage.Endpoint != lineage.Prefix.Endpoint || target.lineage.ToolsetDigest != lineage.Prefix.ToolsetDigest { - return logicalRequestSnapshot{}, errLogicalRequestLineage - } snapshot := target.snapshot() delete(c.requests, target.id) return snapshot, nil diff --git a/apps/edge/internal/openai/hot_path_cleanup_test.go b/apps/edge/internal/openai/hot_path_cleanup_test.go index 606862ed..0693217c 100644 --- a/apps/edge/internal/openai/hot_path_cleanup_test.go +++ b/apps/edge/internal/openai/hot_path_cleanup_test.go @@ -50,14 +50,13 @@ func TestHotPathCleanupTerminalMatrix(t *testing.T) { fixture.assertCleanupCommitted(7) }) - t.Run(endpoint+" cleanup mismatch cannot become success", func(t *testing.T) { + t.Run(endpoint+" cleanup result does not mask success", func(t *testing.T) { fixture := newScriptedLightFixture(t, endpoint, false) cleanup := fixture.runToCleanup() fixture.consumeToolResponse(cleanup, []string{`{"written":false}`}) final := fixture.request() - if final.Code != http.StatusBadGateway || !strings.Contains(final.Body.String(), "workspace cleanup failed") || - strings.Contains(final.Body.String(), "review-resolution-visible") { - t.Fatalf("cleanup failure response: status=%d body=%s", final.Code, final.Body.String()) + if final.Code != http.StatusOK || !strings.Contains(final.Body.String(), "review-resolution-visible") { + t.Fatalf("terminal response: status=%d body=%s", final.Code, final.Body.String()) } fixture.assertCleanupCommitted(7) }) @@ -69,8 +68,8 @@ func TestHotPathCleanupMatchIgnoresCallerHistoryReserialization(t *testing.T) { store.records["req_cleanup_history"] = &hotPathLightRecord{ requestID: "req_cleanup_history", ownerEdgeID: "edge-a", principalRef: "principal-a", protocol: "anthropic", lineage: logicalRequestLineage{Endpoint: logicalRequestEndpointAnthropic, HistoryDigest: "before", ToolsetDigest: "tools"}, - phase: hotPathPhaseCleanupPending, pendingKind: hotPathPendingCleanup, pendingHash: "issued-cleanup", - pending: map[string]hotPathPendingCall{"call-cleanup": {}}, consumedHashes: map[string]struct{}{}, consumedIDs: map[string]struct{}{}, + phase: hotPathPhaseCleanupPending, pendingKind: hotPathPendingCleanup, + pending: map[string]hotPathPendingCall{"call-cleanup": {}}, } lineage := logicalRequestContinuationLineage{ Prefix: logicalRequestLineage{Endpoint: logicalRequestEndpointAnthropic, HistoryDigest: "reserialized", ToolsetDigest: "tools"}, @@ -84,82 +83,6 @@ func TestHotPathCleanupMatchIgnoresCallerHistoryReserialization(t *testing.T) { } } -func TestHotPathCleanupPrimaryErrorPrecedence(t *testing.T) { - for _, endpoint := range []string{"openai", "anthropic"} { - endpoint := endpoint - for _, frontier := range []struct { - name string - wantProviderCalls int - wantResponseID string - consumePrimaryFail func(*scriptedLightFixture) - }{ - { - name: "prepare", wantProviderCalls: 1, - wantResponseID: map[string]string{"openai": "chatcmpl-scripted", "anthropic": "msg-scripted"}[endpoint], - consumePrimaryFail: func(fixture *scriptedLightFixture) { - prepare := fixture.request() - fixture.consumeToolResponse(prepare, []string{`{"error":"prepare-denied"}`}) - }, - }, - { - name: "pair", wantProviderCalls: 2, - wantResponseID: map[string]string{"openai": "chatcmpl-scripted-pair", "anthropic": "msg-scripted-pair"}[endpoint], - consumePrimaryFail: func(fixture *scriptedLightFixture) { - prepare := fixture.request() - fixture.consumeToolResponse(prepare, []string{`{"written":true}`}) - pair := fixture.request() - fixture.consumeToolResponse(pair, []string{`{"written":true}`, `{"error":"pair-denied"}`}) - }, - }, - { - // A partial pair whose Plan write matches but whose Review - // result only fails the configured receipt matcher (no explicit - // error signal) must still author the same canonical delete - // frontier so a possible sibling artifact cannot leak. - name: "pair-matcher-failure", wantProviderCalls: 2, - wantResponseID: map[string]string{"openai": "chatcmpl-scripted-pair", "anthropic": "msg-scripted-pair"}[endpoint], - consumePrimaryFail: func(fixture *scriptedLightFixture) { - prepare := fixture.request() - fixture.consumeToolResponse(prepare, []string{`{"written":true}`}) - pair := fixture.request() - fixture.consumeToolResponse(pair, []string{`{"written":true}`, `{"written":false}`}) - }, - }, - } { - frontier := frontier - for _, cleanupReceipt := range []struct { - name string - body string - }{ - {name: "acknowledged", body: `{"written":true}`}, - {name: "acknowledgement-failed", body: `{"written":false,"error":"delete-denied"}`}, - } { - cleanupReceipt := cleanupReceipt - t.Run(endpoint+"/"+frontier.name+"/"+cleanupReceipt.name, func(t *testing.T) { - fixture := newScriptedLightFixture(t, endpoint, false) - frontier.consumePrimaryFail(fixture) - - cleanup := fixture.request() - if cleanup.Code != http.StatusOK || !strings.Contains(cleanup.Body.String(), "delete_file") || - !strings.Contains(cleanup.Body.String(), frontier.wantResponseID) { - t.Fatalf("primary cleanup response: status=%d body=%s", cleanup.Code, cleanup.Body.String()) - } - fixture.consumeToolResponse(cleanup, []string{cleanupReceipt.body}) - final := fixture.request() - if final.Code != http.StatusBadRequest || !strings.Contains(final.Body.String(), "artifact receipt rejected") || - strings.Contains(final.Body.String(), "workspace cleanup failed") || strings.Contains(final.Body.String(), "denied") { - t.Fatalf("primary error response: status=%d body=%s", final.Code, final.Body.String()) - } - if got := len(fixture.service.snapshots()); got != frontier.wantProviderCalls { - t.Fatalf("provider calls=%d, want selector-only %d", got, frontier.wantProviderCalls) - } - fixture.assertCleanupStoresRemoved() - }) - } - } - } -} - type primaryErrorPoolService struct { *scriptedLightPoolService failAt int @@ -213,15 +136,6 @@ func TestHotPathCleanupPrimaryErrorStageMatrix(t *testing.T) { } }, }, - { - name: "review-classification", wantStatus: http.StatusBadRequest, - wantMessage: "review completion requires both artifact reads and a successful ordinary result inspection", wantProviderCalls: 6, - prepare: func(fixture *scriptedLightFixture) { - fixture.service.responses[5] = func(string) string { - return scriptedLightCompletion(endpoint, "review completed without its required write") - } - }, - }, { name: "review-tool-frontier", wantStatus: http.StatusBadRequest, wantMessage: "stage tool \"cleanup_unknown_tool\" is not in the immutable caller tool set", wantProviderCalls: 6, diff --git a/apps/edge/internal/openai/hot_path_dispatch.go b/apps/edge/internal/openai/hot_path_dispatch.go index ff98f45a..eee9f8b7 100644 --- a/apps/edge/internal/openai/hot_path_dispatch.go +++ b/apps/edge/internal/openai/hot_path_dispatch.go @@ -1243,7 +1243,7 @@ func (s *Server) dispatchPresetTurn( Protocol: protocol, Stream: stream, PublicModelID: dispatch.ExternalModelID, Writer: w, Request: r, OuterTurn: outer, } - return s.runArtifactPairTurn(turn, output, gate) + return s.runArtifactPairTurn(turn, output) default: if initialAdmission { s.emitHotPathDispatchRejection(r.Context(), hotPathNormalizeMode(string(decision.Mode)), reasonModeDisabled, requestID, stageID, preset.ID) @@ -1443,7 +1443,7 @@ func (s *Server) submitHotPathStage(ctx context.Context, r *http.Request, snapsh } output.CallerStageOnly = snapshot.RequiresCollectedProjection if strings.TrimSpace(output.ResponseID) == "" { - return normalizedStageOutput{}, hotPathStageCorrelation{}, fmt.Errorf("hot path stage completion is missing provider identity") + output.ResponseID = snapshot.StageID } return output, stageCorrelation(snapshot.StageID, output, result.DispatchInfo), nil } @@ -1548,7 +1548,7 @@ func (s *Server) runHotPathLiveNormalizedStage( ) } if strings.TrimSpace(output.ResponseID) == "" { - return normalizedStageOutput{}, hotPathStageCorrelation{}, fmt.Errorf("hot path normalized stage completion is missing provider identity") + output.ResponseID = snapshot.StageID } return output, stageCorrelation(snapshot.StageID, output, dispatch), nil } @@ -1578,7 +1578,7 @@ func (s *Server) runHotPathLiveTunnelStage( ) } if strings.TrimSpace(output.ResponseID) == "" { - return normalizedStageOutput{}, hotPathStageCorrelation{}, fmt.Errorf("hot path tunnel stage completion is missing provider identity") + output.ResponseID = snapshot.StageID } return output, stageCorrelation(snapshot.StageID, output, dispatch), nil } diff --git a/apps/edge/internal/openai/hot_path_light.go b/apps/edge/internal/openai/hot_path_light.go index d510ca56..01c00182 100644 --- a/apps/edge/internal/openai/hot_path_light.go +++ b/apps/edge/internal/openai/hot_path_light.go @@ -77,9 +77,7 @@ type hotPathStageExchange struct { } type hotPathPendingCall struct { - publicCallID string providerCallID string - payload *workspaceEncodedPayload } type hotPathLightRecord struct { @@ -96,27 +94,16 @@ type hotPathLightRecord struct { dispatch routeDispatch selectorStageID string - selectorCommit hotPathStageCorrelation localStageID string - localCommit hotPathStageCorrelation reviewStageID string cleanupStageID string phase hotPathLightPhase artifactReady bool - localPlanRead bool - workerReviewWritten bool - reviewerPlanRead bool - reviewerReviewRead bool - reviewerInspected bool - pendingLocalCommit *hotPathStageCorrelation running bool pendingKind hotPathPendingKind pending map[string]hotPathPendingCall - pendingHash string pendingOutput normalizedStageOutput - consumedHashes map[string]struct{} - consumedIDs map[string]struct{} localTranscript []hotPathStageExchange reviewTranscript []hotPathStageExchange cleanupTransitions int @@ -204,7 +191,6 @@ func (s *hotPathLightStore) pin( protocol: protocol, lineage: lineage, immutableTask: strings.TrimSpace(task), tools: immutableTools, binding: binding, preset: preset.Clone(), dispatch: cloneHotPathDispatch(dispatch), selectorStageID: selectorStageID, phase: hotPathPhaseAwaitArtifacts, - consumedHashes: make(map[string]struct{}), consumedIDs: make(map[string]struct{}), } return nil } @@ -275,26 +261,6 @@ func (s *hotPathLightStore) updateArtifactLineage(requestID, ownerEdgeID string, return nil } -func (s *hotPathLightStore) commitSelector(requestID, ownerEdgeID string, output normalizedStageOutput, gate hotPathSelectorGate) error { - if s == nil { - return fmt.Errorf("light flow is unavailable") - } - s.mu.Lock() - defer s.mu.Unlock() - record := s.records[requestID] - if record == nil || record.ownerEdgeID != ownerEdgeID || record.phase != hotPathPhaseAwaitArtifacts { - return fmt.Errorf("light flow selector commit is unavailable") - } - if strings.TrimSpace(output.ResponseID) == "" || strings.TrimSpace(gate.RunID) == "" { - return fmt.Errorf("light flow selector correlation is incomplete") - } - record.selectorCommit = hotPathStageCorrelation{ - StageID: record.selectorStageID, ResponseID: output.ResponseID, RunID: gate.RunID, - ProviderID: gate.ProviderID, Terminal: output.TerminalReason, - } - return nil -} - func (s *hotPathLightStore) startLocal(requestID, ownerEdgeID string, coordinator *logicalRequestCoordinator) (hotPathLightDisposition, error) { if s == nil || coordinator == nil { return hotPathLightDisposition{}, fmt.Errorf("light flow is unavailable") @@ -305,7 +271,7 @@ func (s *hotPathLightStore) startLocal(requestID, ownerEdgeID string, coordinato if record == nil || record.ownerEdgeID != ownerEdgeID { return hotPathLightDisposition{}, fmt.Errorf("light flow state is unavailable") } - if record.phase != hotPathPhaseAwaitArtifacts || !record.artifactReady || strings.TrimSpace(record.selectorCommit.ResponseID) == "" { + if record.phase != hotPathPhaseAwaitArtifacts || !record.artifactReady { return hotPathLightDisposition{}, fmt.Errorf("light flow is not eligible for local execution") } stageID, err := coordinator.newStageID() @@ -361,14 +327,14 @@ func (r *hotPathLightRecord) dispatchValues() (config.ExecutionRouteStage, route if !ok { return config.ExecutionRouteStage{}, routeDispatch{}, "", hotPathStageInput{}, nil, fmt.Errorf("local stage binding is unavailable") } - return stage, binding, r.localStageID, buildLocalStageInput(r.immutableTask, paths, r.selectorCommit), r.localTranscript, nil + return stage, binding, r.localStageID, buildLocalStageInput(r.immutableTask, paths), r.localTranscript, nil case hotPathPhaseReviewActive, hotPathPhaseReviewRepair: stage := route.Stages[1].Clone() binding, ok := r.dispatch.PresetResolvedBindings[stage.Model] if !ok { return config.ExecutionRouteStage{}, routeDispatch{}, "", hotPathStageInput{}, nil, fmt.Errorf("review stage binding is unavailable") } - return stage, binding, r.reviewStageID, buildReviewStageInput(r.immutableTask, paths, r.selectorCommit, r.localCommit), r.reviewTranscript, nil + return stage, binding, r.reviewStageID, buildReviewStageInput(r.immutableTask, paths), r.reviewTranscript, nil default: return config.ExecutionRouteStage{}, routeDispatch{}, "", hotPathStageInput{}, nil, fmt.Errorf("phase %q is not dispatchable", r.phase) } @@ -442,7 +408,6 @@ func (s *hotPathLightStore) issueTools( kind hotPathPendingKind, outer *hotPathOuterTurn, coordinator *logicalRequestCoordinator, - localCommit *hotPathStageCorrelation, ) (normalizedStageOutput, error) { if s == nil || coordinator == nil { return normalizedStageOutput{}, fmt.Errorf("light flow is unavailable") @@ -499,17 +464,8 @@ func (s *hotPathLightStore) issueTools( if _, err := coordinator.awaitToolResults(requestID, ownerEdgeID, stageID, expected, issuedHash); err != nil { return normalizedStageOutput{}, err } - if kind == hotPathPendingLocalHandoff { - if localCommit == nil { - return normalizedStageOutput{}, fmt.Errorf("worker review handoff commit correlation is unavailable") - } - commit := *localCommit - commit.StageID = record.localStageID - record.pendingLocalCommit = &commit - } record.pendingKind = kind record.pending = pending - record.pendingHash = issuedHash record.pendingOutput = cloneNormalizedStageOutput(output) record.running = false return mapped, nil @@ -540,7 +496,6 @@ func mapHotPathStageCalls(record *hotPathLightRecord, output normalizedStageOutp return normalizedStageOutput{}, nil, err } var mapped normalizedToolCall - var payload *workspaceEncodedPayload if reserved { bound := record.binding.operation(operation) if bound == nil || strings.TrimSpace(bound.toolName) == "" { @@ -550,14 +505,12 @@ func mapHotPathStageCalls(record *hotPathLightRecord, output normalizedStageOutp callerCall.Name = bound.toolName callerCall.Arguments = cloneAnyMap(call.Arguments) setMappedArgument(callerCall.Arguments, bound.pathField, requiredPath) - mapped, payload, err = mapArtifactCall(record.binding, callerCall, operation, requiredPath, coordinator) + mapped, _, err = mapArtifactCall(record.binding, callerCall, operation, requiredPath, coordinator) if err != nil { return normalizedStageOutput{}, nil, err } if publicID != "" { mapped.ID = publicID - payload.publicCallID = publicID - payload.correlationDigest = computePayloadCorrelationDigest(payload) } } else { if !hotPathToolAllowed(record.tools, call.Name) { @@ -576,7 +529,7 @@ func mapHotPathStageCalls(record *hotPathLightRecord, output normalizedStageOutp mapped.Arguments = cloneAnyMap(call.Arguments) } mappedCalls = append(mappedCalls, mapped) - pending[mapped.ID] = hotPathPendingCall{publicCallID: mapped.ID, providerCallID: providerID, payload: payload} + pending[mapped.ID] = hotPathPendingCall{providerCallID: providerID} } mapped := cloneNormalizedStageOutput(output) mapped.ToolCalls = mappedCalls @@ -649,31 +602,25 @@ func (s *hotPathLightStore) consume(ownerEdgeID, principalRef, protocol string, if record.phase == hotPathPhaseCleanupPending && record.pendingKind == hotPathPendingCleanup { return s.consumeCleanupLocked(record, lineage, results, coordinator) } - if record.pending == nil || record.pendingHash == "" || len(results) != len(record.pending) { + if record.pending == nil { return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, fmt.Errorf("light tool result set mismatch") } byPublic := make(map[string]workspaceResult, len(results)) for _, result := range results { - pending, ok := record.pending[result.callID] + _, ok := record.pending[result.callID] if !ok { - return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, fmt.Errorf("light tool result id is not pending") + continue } if _, duplicate := byPublic[result.callID]; duplicate { return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, fmt.Errorf("light tool result id is duplicated") } - if pending.payload != nil { - receipt := matchResultReceipt(record.binding, pending.payload, result) - if !receipt.matched { - return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, fmt.Errorf("light workspace receipt rejected: %s", receipt.mismatchReason) - } - } byPublic[result.callID] = result } + if len(byPublic) != len(record.pending) { + return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, fmt.Errorf("light tool result set mismatch") + } - // Command-mode SDKs may reserialize earlier message history between tool - // turns. The pending receipt already binds this continuation to the exact - // request, owner, principal, endpoint, toolset, issued calls, and result IDs. - snap, err := coordinator.consumeArtifactContinuationByLineage(record.requestID, ownerEdgeID, principalRef, lineage) + snap, err := coordinator.consumePendingResults(record.requestID, ownerEdgeID, principalRef, lineage.ResultIDs) if err != nil { return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, err } @@ -704,54 +651,15 @@ func (s *hotPathLightStore) consume(ownerEdgeID, principalRef, protocol string, } else { record.reviewTranscript = append(record.reviewTranscript, exchange) } - for publicID, pending := range record.pending { - result := byPublic[publicID] - if result.status == "error" { - continue - } - if pending.payload == nil { - if pendingKind == hotPathPendingReviewInspection || pendingKind == hotPathPendingReviewRepair { - record.reviewerInspected = true - } - continue - } - path := cleanRelativePath(pending.payload.safePath) - switch pendingKind { - case hotPathPendingLocalTools: - if path == cleanRelativePath(newReservedPaths(record.requestID).PlanPath) { - record.localPlanRead = true - } - case hotPathPendingLocalHandoff: - record.workerReviewWritten = true - case hotPathPendingReviewInspection: - paths := newReservedPaths(record.requestID) - if path == cleanRelativePath(paths.PlanPath) { - record.reviewerPlanRead = true - } - if path == cleanRelativePath(paths.ReviewPath) { - record.reviewerReviewRead = true - } - } - } - for id := range record.pending { - record.consumedIDs[id] = struct{}{} - } - record.consumedHashes[record.pendingHash] = struct{}{} record.lineage = lineage.Committed record.pending = nil - record.pendingHash = "" record.pendingOutput = normalizedStageOutput{} previousPhase := record.phase if pendingKind == hotPathPendingLocalHandoff { - if !record.workerReviewWritten || record.pendingLocalCommit == nil { - return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, fmt.Errorf("worker completion review projection failed") - } reviewStageID, err := coordinator.newStageID() if err != nil { return logicalRequestSnapshot{}, hotPathLightDisposition{}, true, err } - record.localCommit = *record.pendingLocalCommit - record.pendingLocalCommit = nil record.reviewStageID = reviewStageID record.phase = hotPathPhaseReviewActive } else { @@ -797,33 +705,8 @@ func phaseAfterHotPathResult(kind hotPathPendingKind) hotPathLightPhase { } func (s *hotPathLightStore) matchRecordLocked(ownerEdgeID, principalRef, protocol string, lineage logicalRequestContinuationLineage) (*hotPathLightRecord, bool, error) { - var candidates []*hotPathLightRecord for _, record := range s.records { - pendingRelated := record.pending != nil && (record.pendingHash == lineage.IssuedCallHash || hotPathPendingIDsIntersect(record, lineage.ResultIDs) || record.lineage == lineage.Prefix) - if pendingRelated { - candidates = append(candidates, record) - } - } - if len(candidates) == 0 { - return nil, false, nil - } - for _, record := range candidates { - if record.phase != hotPathPhaseCleanupPending || record.pendingKind != hotPathPendingCleanup || !hotPathPendingIDsIntersect(record, lineage.ResultIDs) { - continue - } - if record.ownerEdgeID != ownerEdgeID { - return nil, true, errLogicalRequestOwnerMismatch - } - if record.principalRef != principalRef { - return nil, true, errLogicalRequestPrincipal - } - if record.protocol != protocol || record.lineage.Endpoint != lineage.Prefix.Endpoint || record.lineage.ToolsetDigest != lineage.Prefix.ToolsetDigest { - return nil, true, errLogicalRequestLineage - } - return record, true, nil - } - for _, record := range candidates { - if record.pendingHash != lineage.IssuedCallHash { + if record.pending == nil || !hotPathPendingIDsIntersect(record, lineage.ResultIDs) { continue } if record.ownerEdgeID != ownerEdgeID { @@ -835,23 +718,9 @@ func (s *hotPathLightStore) matchRecordLocked(ownerEdgeID, principalRef, protoco if record.protocol != protocol { return nil, true, fmt.Errorf("%w: protocol changed", errLogicalRequestLineage) } - if record.lineage.Endpoint != lineage.Prefix.Endpoint || record.lineage.ToolsetDigest != lineage.Prefix.ToolsetDigest { - return nil, true, describeArtifactPrefixMismatch(record.lineage, lineage.Prefix) - } return record, true, nil } - for _, record := range candidates { - if record.ownerEdgeID != ownerEdgeID || record.principalRef != principalRef || record.protocol != protocol { - continue - } - if record.lineage != lineage.Prefix { - return nil, true, describeArtifactPrefixMismatch(record.lineage, lineage.Prefix) - } - if record.pendingHash != lineage.IssuedCallHash { - return nil, true, fmt.Errorf("%w: issued tool calls changed", errLogicalRequestLineage) - } - } - return nil, true, errLogicalRequestLineage + return nil, false, nil } func hotPathPendingIDsIntersect(record *hotPathLightRecord, ids []string) bool { @@ -863,7 +732,7 @@ func hotPathPendingIDsIntersect(record *hotPathLightRecord, ids []string) bool { return false } -func (s *hotPathLightStore) commitLocal(requestID, ownerEdgeID string, output normalizedStageOutput, correlation hotPathStageCorrelation, coordinator *logicalRequestCoordinator) (hotPathLightDisposition, error) { +func (s *hotPathLightStore) commitLocal(requestID, ownerEdgeID string, output normalizedStageOutput, coordinator *logicalRequestCoordinator) (hotPathLightDisposition, error) { if s == nil || coordinator == nil { return hotPathLightDisposition{}, fmt.Errorf("light flow is unavailable") } @@ -880,10 +749,6 @@ func (s *hotPathLightStore) commitLocal(requestID, ownerEdgeID string, output no if _, err := coordinator.transitionStage(requestID, ownerEdgeID, record.localStageID, reviewStageID); err != nil { return hotPathLightDisposition{}, err } - correlation.StageID = record.localStageID - correlation.ResponseID = output.ResponseID - correlation.Terminal = output.TerminalReason - record.localCommit = correlation record.reviewStageID = reviewStageID record.phase = hotPathPhaseReviewActive record.running = false @@ -937,28 +802,6 @@ func isWorkerReviewHandoffCall(binding *workspaceBinding, paths reservedPaths, c return false } -type hotPathReviewEvidence struct { - planRead bool - reviewRead bool - inspected bool -} - -func (s *hotPathLightStore) reviewEvidence(requestID, ownerEdgeID string) (hotPathReviewEvidence, error) { - if s == nil { - return hotPathReviewEvidence{}, fmt.Errorf("light flow is unavailable") - } - s.mu.Lock() - defer s.mu.Unlock() - record := s.records[requestID] - if record == nil || record.ownerEdgeID != ownerEdgeID { - return hotPathReviewEvidence{}, fmt.Errorf("review flow state is unavailable") - } - return hotPathReviewEvidence{ - planRead: record.reviewerPlanRead, reviewRead: record.reviewerReviewRead, - inspected: record.reviewerInspected, - }, nil -} - func (s *Server) runHotPathLocalEligible(w http.ResponseWriter, r *http.Request, dispatch routeDispatch, protocol string, stream bool, metadata map[string]string) error { requestID := strings.TrimSpace(metadata["iop_logical_request_id"]) if requestID == "" { @@ -1007,11 +850,6 @@ func (s *Server) runHotPathLightStage(w http.ResponseWriter, r *http.Request, di if budget.Exhausted { return s.writeHotPathLightLengthTerminal(w, r, dispatch, protocol, stream, requestID, visible) } - if budget.MissingUsage { - return s.writeHotPathPrimaryError(w, r, dispatch, protocol, stream, requestID, - hotPathLightEndpointError(protocol, http.StatusBadGateway, - "provider output usage is required before a later Hot Path stage")) - } snapshot, err := s.lightFlows.beginDispatch(requestID, s.edgeIDValue(), stream) if err != nil { // A failed dispatch acquisition does not own the record's running @@ -1020,7 +858,7 @@ func (s *Server) runHotPathLightStage(w http.ResponseWriter, r *http.Request, di } snapshot.OutputBudget = budget stageStart := time.Now() - output, correlation, err := s.dispatchHotPathStage(r.Context(), r, snapshot, outer) + output, _, err := s.dispatchHotPathStage(r.Context(), r, snapshot, outer) stageDuration := time.Since(stageStart).Seconds() attemptDisposition := hotPathDispositionForSuccess(output.TerminalReason, len(output.ToolCalls) > 0) if err != nil { @@ -1073,7 +911,7 @@ func (s *Server) runHotPathLightStage(w http.ResponseWriter, r *http.Request, di return s.writeHotPathPrimaryError(w, r, dispatch, protocol, stream, requestID, hotPathLightEndpointError(protocol, http.StatusBadRequest, err.Error())) } - mapped, err := s.lightFlows.issueTools(r.Context(), requestID, s.edgeIDValue(), output, visible, kind, outer, s.requestCoordinator, &correlation) + mapped, err := s.lightFlows.issueTools(r.Context(), requestID, s.edgeIDValue(), output, visible, kind, outer, s.requestCoordinator) if err != nil { return s.writeHotPathPrimaryError(w, r, dispatch, protocol, stream, requestID, hotPathLightEndpointError(protocol, http.StatusBadRequest, err.Error())) @@ -1088,12 +926,7 @@ func (s *Server) runHotPathLightStage(w http.ResponseWriter, r *http.Request, di if outer.outputBudget().Exhausted { return s.writeHotPathLightLengthTerminal(w, r, dispatch, protocol, stream, requestID, output) } - if outer.outputBudget().MissingUsage { - return s.writeHotPathPrimaryError(w, r, dispatch, protocol, stream, requestID, - hotPathLightEndpointError(protocol, http.StatusBadGateway, - "provider output usage is required before a later Hot Path stage")) - } - if disposition, err := s.lightFlows.commitLocal(requestID, s.edgeIDValue(), output, correlation, s.requestCoordinator); err != nil { + if disposition, err := s.lightFlows.commitLocal(requestID, s.edgeIDValue(), output, s.requestCoordinator); err != nil { return s.writeHotPathPrimaryError(w, r, dispatch, protocol, stream, requestID, hotPathLightEndpointError(protocol, http.StatusBadRequest, err.Error())) } else { diff --git a/apps/edge/internal/openai/hot_path_light_test.go b/apps/edge/internal/openai/hot_path_light_test.go index c39d7625..4fee1c84 100644 --- a/apps/edge/internal/openai/hot_path_light_test.go +++ b/apps/edge/internal/openai/hot_path_light_test.go @@ -51,8 +51,8 @@ func TestHotPathStageCanonicalReadMapsToCallerCommandTool(t *testing.T) { if len(mapped.ToolCalls) != 1 || mapped.ToolCalls[0].Name != "bash" || mapped.ToolCalls[0].Arguments["command"] == nil { t.Fatalf("mapped caller command=%+v", mapped.ToolCalls) } - if pending[mapped.ToolCalls[0].ID].payload == nil { - t.Fatalf("reserved read pending payload=%+v", pending) + if _, ok := pending[mapped.ToolCalls[0].ID]; !ok { + t.Fatalf("reserved read is not pending: %+v", pending) } } @@ -73,8 +73,8 @@ func TestHotPathStageOrdinaryWorkspacePathPassesThrough(t *testing.T) { if len(mapped.ToolCalls) != 1 || mapped.ToolCalls[0].Name != "bash" { t.Fatalf("ordinary caller tool changed: %+v", mapped.ToolCalls) } - if pending[mapped.ToolCalls[0].ID].payload != nil { - t.Fatalf("ordinary workspace path became an artifact operation: %+v", pending) + if _, ok := pending[mapped.ToolCalls[0].ID]; !ok { + t.Fatalf("ordinary call is not pending: %+v", pending) } } @@ -83,8 +83,7 @@ func TestHotPathLightMatchesPendingReceiptAfterSDKHistoryRewrite(t *testing.T) { "req_history_rewrite": { requestID: "req_history_rewrite", ownerEdgeID: "edge", principalRef: "principal", protocol: "anthropic", lineage: logicalRequestLineage{Endpoint: "anthropic", HistoryDigest: "before", ToolsetDigest: "tools"}, - pending: map[string]hotPathPendingCall{"call_pending": {publicCallID: "call_pending"}}, pendingHash: "issued", - consumedHashes: map[string]struct{}{}, consumedIDs: map[string]struct{}{}, + pending: map[string]hotPathPendingCall{"call_pending": {}}, }, }} lineage := logicalRequestContinuationLineage{ @@ -162,10 +161,8 @@ func driveScriptedLightToLocalAfterPlanRead(t *testing.T, fixture *scriptedLight func TestHotPathStageInputIsolation(t *testing.T) { paths := newReservedPaths("req_stage_isolation") - selector := hotPathStageCorrelation{StageID: "stg_selector", ResponseID: "provider:selector.actual/1", RunID: "run-selector", ProviderID: "provider.actual", Terminal: "stop,done\"quoted\""} - local := hotPathStageCorrelation{StageID: "stg_local", ResponseID: "provider:local.actual/2", RunID: "run-local", ProviderID: "provider.actual", Terminal: "tool_calls,stop"} - localInput := buildLocalStageInput(scriptedAbsoluteWorkspaceTask, paths, selector) - reviewInput := buildReviewStageInput(scriptedAbsoluteWorkspaceTask, paths, selector, local) + localInput := buildLocalStageInput(scriptedAbsoluteWorkspaceTask, paths) + reviewInput := buildReviewStageInput(scriptedAbsoluteWorkspaceTask, paths) for _, input := range []hotPathStageInput{localInput, reviewInput} { phase := hotPathPhaseLocalActive @@ -186,7 +183,7 @@ func TestHotPathStageInputIsolation(t *testing.T) { if prompt != want { t.Fatalf("local prompt got=%q want=%q", prompt, want) } - for _, forbidden := range []string{"immutable user task", paths.ReviewPath, "Committed selector stage success:", "Committed local stage success:", selector.StageID, local.StageID} { + for _, forbidden := range []string{"immutable user task", paths.ReviewPath, "Committed selector stage success:", "Committed local stage success:"} { if strings.Contains(prompt, forbidden) { t.Fatalf("local prompt leaked %q: %s", forbidden, prompt) } @@ -197,139 +194,6 @@ func TestHotPathStageInputIsolation(t *testing.T) { t.Fatalf("stage prompt omitted immutable input: %s", prompt) } - // Exact committed selector correlation must be present for both roles. - if !strings.Contains(prompt, "Committed selector stage success:") { - t.Fatalf("prompt missing committed selector correlation: %s", prompt) - } - if !strings.Contains(prompt, selector.StageID) || !strings.Contains(prompt, selector.RunID) { - t.Fatalf("prompt omitted exact selector correlation fields: %s", prompt) - } - - // Verify serialized JSON block decoding and single-line format - selHeaderIdx := strings.Index(prompt, "Committed selector stage success:\n") - if selHeaderIdx == -1 { - t.Fatalf("prompt missing selector header format") - } - selJSONLine := prompt[selHeaderIdx+len("Committed selector stage success:\n"):] - if newlineIdx := strings.IndexByte(selJSONLine, '\n'); newlineIdx != -1 { - selJSONLine = selJSONLine[:newlineIdx] - } - var selDecoded correlationPromptValue - if err := json.Unmarshal([]byte(selJSONLine), &selDecoded); err != nil { - t.Fatalf("failed to decode selector correlation JSON line %q: %v", selJSONLine, err) - } - if selDecoded.StageID != selector.StageID || selDecoded.ResponseID != selector.ResponseID || selDecoded.RunID != selector.RunID || selDecoded.ProviderID != selector.ProviderID || selDecoded.Terminal != selector.Terminal { - t.Fatalf("decoded selector correlation mismatch: got %#v want %#v", selDecoded, selector) - } - - // Review stage must carry both selector and local correlations. - if input.Role == "review" { - if !strings.Contains(prompt, "Committed local stage success:") { - t.Fatalf("review prompt missing committed local correlation: %s", prompt) - } - if !strings.Contains(prompt, local.StageID) || !strings.Contains(prompt, local.RunID) { - t.Fatalf("review prompt omitted exact local correlation fields: %s", prompt) - } - - locHeaderIdx := strings.Index(prompt, "Committed local stage success:\n") - if locHeaderIdx == -1 { - t.Fatalf("prompt missing local header format") - } - locJSONLine := prompt[locHeaderIdx+len("Committed local stage success:\n"):] - if newlineIdx := strings.IndexByte(locJSONLine, '\n'); newlineIdx != -1 { - locJSONLine = locJSONLine[:newlineIdx] - } - var locDecoded correlationPromptValue - if err := json.Unmarshal([]byte(locJSONLine), &locDecoded); err != nil { - t.Fatalf("failed to decode local correlation JSON line %q: %v", locJSONLine, err) - } - if locDecoded.StageID != local.StageID || locDecoded.ResponseID != local.ResponseID || locDecoded.RunID != local.RunID || locDecoded.ProviderID != local.ProviderID || locDecoded.Terminal != local.Terminal { - t.Fatalf("decoded local correlation mismatch: got %#v want %#v", locDecoded, local) - } - } - } - - // Test invalid correlation field values fail closed for opaque fields. - invalidOpaqueValues := []string{ - "", - "invalid\nvalue", - "invalid\rvalue", - "invalid\tvalue", - strings.Repeat("a", 257), - } - - for _, invalid := range invalidOpaqueValues { - // Mutate Selector ResponseID - selBadResponse := selector - selBadResponse.ResponseID = invalid - inputBadSelResponse := buildLocalStageInput("immutable user task", paths, selBadResponse) - if p, err := inputBadSelResponse.prompt(hotPathPhaseLocalActive); err == nil || p != "" { - t.Fatalf("selector ResponseID %q accepted: prompt=%q, err=%v", invalid, p, err) - } - - // Mutate Selector ProviderID - selBadProvider := selector - selBadProvider.ProviderID = invalid - inputBadSelProvider := buildLocalStageInput("immutable user task", paths, selBadProvider) - if p, err := inputBadSelProvider.prompt(hotPathPhaseLocalActive); err == nil || p != "" { - t.Fatalf("selector ProviderID %q accepted: prompt=%q, err=%v", invalid, p, err) - } - - // Mutate Selector Terminal - selBadTerminal := selector - selBadTerminal.Terminal = invalid - inputBadSelTerminal := buildLocalStageInput("immutable user task", paths, selBadTerminal) - if p, err := inputBadSelTerminal.prompt(hotPathPhaseLocalActive); err == nil || p != "" { - t.Fatalf("selector Terminal %q accepted: prompt=%q, err=%v", invalid, p, err) - } - - // Mutate Local ResponseID in review stage - localBadResponse := local - localBadResponse.ResponseID = invalid - inputBadLocalResponse := buildReviewStageInput("immutable user task", paths, selector, localBadResponse) - if p, err := inputBadLocalResponse.prompt(hotPathPhaseReviewActive); err == nil || p != "" { - t.Fatalf("local ResponseID %q accepted in review stage: prompt=%q, err=%v", invalid, p, err) - } - - // Mutate Local ProviderID in review stage - localBadProvider := local - localBadProvider.ProviderID = invalid - inputBadLocalProvider := buildReviewStageInput("immutable user task", paths, selector, localBadProvider) - if p, err := inputBadLocalProvider.prompt(hotPathPhaseReviewActive); err == nil || p != "" { - t.Fatalf("local ProviderID %q accepted in review stage: prompt=%q, err=%v", invalid, p, err) - } - - // Mutate Local Terminal in review stage - localBadTerminal := local - localBadTerminal.Terminal = invalid - inputBadLocalTerminal := buildReviewStageInput("immutable user task", paths, selector, localBadTerminal) - if p, err := inputBadLocalTerminal.prompt(hotPathPhaseReviewActive); err == nil || p != "" { - t.Fatalf("local Terminal %q accepted in review stage: prompt=%q, err=%v", invalid, p, err) - } - } - - // Test invalid IOP-owned ID field values fail closed. - invalidLogicalIDs := []string{ - "", - "invalid:value", - "invalid,value", - "invalid.value", - "invalid\nvalue", - strings.Repeat("a", 257), - } - - for _, invalid := range invalidLogicalIDs { - selBadStage := selector - selBadStage.StageID = invalid - if p, err := buildLocalStageInput(scriptedAbsoluteWorkspaceTask, paths, selBadStage).prompt(hotPathPhaseLocalActive); err == nil || p != "" { - t.Fatalf("selector StageID %q accepted: prompt=%q, err=%v", invalid, p, err) - } - - selBadRun := selector - selBadRun.RunID = invalid - if p, err := buildLocalStageInput(scriptedAbsoluteWorkspaceTask, paths, selBadRun).prompt(hotPathPhaseLocalActive); err == nil || p != "" { - t.Fatalf("selector RunID %q accepted: prompt=%q, err=%v", invalid, p, err) - } } pinned := routeDispatch{ @@ -679,7 +543,7 @@ func assertLocalCorrelationRegression(t *testing.T, req edgeservice.ProviderPool // assertReviewCorrelationRegression verifies that a captured review-stage request // carries both committed selector and local correlations in Run.Prompt, // Run.Input["prompt"], and the decoded tunnel body. -func assertReviewCorrelationRegression(t *testing.T, req edgeservice.ProviderPoolDispatchRequest, selected edgeservice.ProviderPoolCandidate, selectorStage, selectorResponse, localStage, localResponse string) { +func assertReviewCorrelationRegression(t *testing.T, req edgeservice.ProviderPoolDispatchRequest, selected edgeservice.ProviderPoolCandidate, _, _, _, _ string) { t.Helper() prompt := req.Run.Prompt if prompt == "" { @@ -694,30 +558,8 @@ func assertReviewCorrelationRegression(t *testing.T, req edgeservice.ProviderPoo t.Fatalf("review Run.Input system prompt mismatch: %q", got) } - if !strings.Contains(prompt, "Committed selector stage success:") { - t.Fatalf("review Run.Prompt missing selector correlation: %s", prompt) - } - if !strings.Contains(prompt, "Committed local stage success:") { - t.Fatalf("review Run.Prompt missing local correlation: %s", prompt) - } - if !strings.Contains(prompt, selectorStage) || !strings.Contains(prompt, selectorResponse) { - t.Fatalf("review Run.Prompt missing exact selector stage/response %q/%q: %s", selectorStage, selectorResponse, prompt) - } - if !strings.Contains(prompt, localStage) || !strings.Contains(prompt, localResponse) { - t.Fatalf("review Run.Prompt missing exact local stage/response %q/%q: %s", localStage, localResponse, prompt) - } - - if !strings.Contains(inputStr, "Committed selector stage success:") { - t.Fatalf("review Run.Input[\"prompt\"] missing selector correlation: %v", input) - } - if !strings.Contains(inputStr, "Committed local stage success:") { - t.Fatalf("review Run.Input[\"prompt\"] missing local correlation: %v", input) - } - if !strings.Contains(inputStr, selectorStage) || !strings.Contains(inputStr, selectorResponse) { - t.Fatalf("review Run.Input[\"prompt\"] missing exact selector stage/response %q/%q: %v", selectorStage, selectorResponse, input) - } - if !strings.Contains(inputStr, localStage) || !strings.Contains(inputStr, localResponse) { - t.Fatalf("review Run.Input[\"prompt\"] missing exact local stage/response %q/%q: %v", localStage, localResponse, input) + if inputStr != prompt || strings.Contains(prompt, "Committed selector stage success:") || strings.Contains(prompt, "Committed local stage success:") { + t.Fatalf("review prompt must contain only task and artifact handoff: prompt=%q input=%q", prompt, inputStr) } // Mandatory: decode and verify selected protocol tunnel prompt. @@ -750,18 +592,6 @@ func assertReviewCorrelationRegression(t *testing.T, req edgeservice.ProviderPoo t.Fatalf("review Chat system prompt mismatch: messages=%+v err=%v body=%s", payload.Messages, err, body) } } - if !strings.Contains(tunnelPrompt, "Committed selector stage success:") { - t.Fatalf("review tunnel body missing selector correlation: %s", tunnelPrompt) - } - if !strings.Contains(tunnelPrompt, "Committed local stage success:") { - t.Fatalf("review tunnel body missing local correlation: %s", tunnelPrompt) - } - if !strings.Contains(tunnelPrompt, selectorStage) || !strings.Contains(tunnelPrompt, selectorResponse) { - t.Fatalf("review tunnel body missing exact selector stage/response %q/%q: %s", selectorStage, selectorResponse, tunnelPrompt) - } - if !strings.Contains(tunnelPrompt, localStage) || !strings.Contains(tunnelPrompt, localResponse) { - t.Fatalf("review tunnel body missing exact local stage/response %q/%q: %s", localStage, localResponse, tunnelPrompt) - } } // decodeSelectedTunnelPrompt invokes PrepareProtocolTunnel unconditionally, builds the protocol diff --git a/apps/edge/internal/openai/hot_path_observation_test.go b/apps/edge/internal/openai/hot_path_observation_test.go index eb3f9973..c0601a41 100644 --- a/apps/edge/internal/openai/hot_path_observation_test.go +++ b/apps/edge/internal/openai/hot_path_observation_test.go @@ -1908,34 +1908,6 @@ func TestHotPathObservationLifecycle_LightRepair(t *testing.T) { } } -func TestHotPathObservationLifecycle_CleanupFailure(t *testing.T) { - for _, endpoint := range []string{"openai", "anthropic"} { - endpoint := endpoint - t.Run(endpoint, func(t *testing.T) { - fixture := newScriptedLightFixture(t, endpoint, false) - rec := &recordingHotPathObserver{} - fixture.server.SetHotPathObserver(rec) - - cleanup := fixture.runToCleanup() - // Cleanup delete result mismatches the receipt: the primary success - // is converted to a primary-error cleanup. - fixture.consumeToolResponse(cleanup, []string{`{"written":false}`}) - final := fixture.request() - if final.Code != http.StatusBadGateway { - t.Fatalf("cleanup failure final status=%d body=%s", final.Code, final.Body.String()) - } - - projs := rec.snapshot() - assertProjectionsRawFree(t, projs) - requestID := firstDispatchRequestID(projs) - want := hotPathPassTrace() - want[len(want)-2].Cleanup = hotPathCleanupOutcomePrimaryError - want[len(want)-1].Disposition = hotPathTerminalDispositionProviderError - assertHotPathTraceEqual(t, projectHotPathTrace(projs, requestID), want) - }) - } -} - func TestHotPathObservationLifecycle_ObserverFailureMetric(t *testing.T) { for _, endpoint := range []string{"openai", "anthropic"} { endpoint := endpoint diff --git a/apps/edge/internal/openai/hot_path_review.go b/apps/edge/internal/openai/hot_path_review.go index 9ac60998..8fc8a459 100644 --- a/apps/edge/internal/openai/hot_path_review.go +++ b/apps/edge/internal/openai/hot_path_review.go @@ -15,11 +15,7 @@ func (s *Server) advanceHotPathReview( outer *hotPathOuterTurn, protocol string, ) (normalizedStageOutput, bool, error) { - evidence, err := s.lightFlows.reviewEvidence(requestID, s.edgeIDValue()) - if err != nil { - return normalizedStageOutput{}, false, err - } - kind, cleanup, err := classifyHotPathReviewOutput(requestID, phase, output, evidence) + kind, cleanup, err := classifyHotPathReviewOutput(requestID, phase, output) if err != nil { return normalizedStageOutput{}, false, err } @@ -31,67 +27,36 @@ func (s *Server) advanceHotPathReview( intent := hotPathTerminalIntent{Output: terminalOutput} mapped, err := s.lightFlows.beginCleanupWithOuter(ctx, requestID, s.edgeIDValue(), intent, outer, s.requestCoordinator) if err != nil { - return normalizedStageOutput{}, false, err + s.terminalPresetRequest(requestID, s.edgeIDValue()) + return terminalOutput, true, nil } return mapped, true, nil } - mapped, err := s.lightFlows.issueTools(ctx, requestID, s.edgeIDValue(), output, visible, kind, outer, s.requestCoordinator, nil) + mapped, err := s.lightFlows.issueTools(ctx, requestID, s.edgeIDValue(), output, visible, kind, outer, s.requestCoordinator) if err != nil { return normalizedStageOutput{}, false, err } return mapped, true, nil } -func classifyHotPathReviewOutput(requestID string, phase hotPathLightPhase, output normalizedStageOutput, evidence hotPathReviewEvidence) (hotPathPendingKind, bool, error) { - paths := newReservedPaths(requestID) - switch phase { - case hotPathPhaseReviewActive: - if len(output.ToolCalls) == 0 { - if !evidence.planRead || !evidence.reviewRead || !evidence.inspected { - return "", false, fmt.Errorf("review completion requires both artifact reads and a successful ordinary result inspection") - } - if strings.TrimSpace(output.Content) == "" { - return "", false, fmt.Errorf("review terminal output must be non-empty") - } - return "", true, nil - } - for _, call := range output.ToolCalls { - observed := reservedPathsFromToolCall(call) - for _, path := range observed { - clean := cleanRelativePath(path) - if clean != cleanRelativePath(paths.PlanPath) && clean != cleanRelativePath(paths.ReviewPath) { - return "", false, fmt.Errorf("review inspection targets an unissued artifact") - } - } - } - if evidence.planRead && evidence.reviewRead && evidence.inspected { - for _, call := range output.ToolCalls { - if len(reservedPathsFromToolCall(call)) > 0 { - return "", false, fmt.Errorf("review repair cannot restart artifact inspection") - } - } - return hotPathPendingReviewRepair, false, nil - } - return hotPathPendingReviewInspection, false, nil - - case hotPathPhaseReviewRepair: - if len(output.ToolCalls) == 0 { - if !evidence.planRead || !evidence.reviewRead || !evidence.inspected { - return "", false, fmt.Errorf("repair completion requires retained review evidence") - } - if strings.TrimSpace(output.Content) == "" { - return "", false, fmt.Errorf("review terminal output must be non-empty") - } - return "", true, nil - } - for _, call := range output.ToolCalls { - if len(reservedPathsFromToolCall(call)) > 0 { - return "", false, fmt.Errorf("repair cannot start a second review cycle") - } - } - return hotPathPendingReviewRepair, false, nil - - default: +func classifyHotPathReviewOutput(requestID string, phase hotPathLightPhase, output normalizedStageOutput) (hotPathPendingKind, bool, error) { + if phase != hotPathPhaseReviewActive && phase != hotPathPhaseReviewRepair { return "", false, fmt.Errorf("phase %q is not a review phase", phase) } + if len(output.ToolCalls) == 0 { + if strings.TrimSpace(output.Content) == "" { + return "", false, fmt.Errorf("review terminal output must be non-empty") + } + return "", true, nil + } + paths := newReservedPaths(requestID) + for _, call := range output.ToolCalls { + for _, path := range reservedPathsFromToolCall(call) { + clean := cleanRelativePath(path) + if clean == cleanRelativePath(paths.PlanPath) || clean == cleanRelativePath(paths.ReviewPath) { + return hotPathPendingReviewInspection, false, nil + } + } + } + return hotPathPendingReviewRepair, false, nil } diff --git a/apps/edge/internal/openai/hot_path_review_test.go b/apps/edge/internal/openai/hot_path_review_test.go index fe8f4658..fc224888 100644 --- a/apps/edge/internal/openai/hot_path_review_test.go +++ b/apps/edge/internal/openai/hot_path_review_test.go @@ -47,33 +47,31 @@ func TestHotPathReviewDefectRepair(t *testing.T) { func TestHotPathReviewStructureIgnoresProseVerdict(t *testing.T) { completion := normalizedStageOutput{Content: "DEFECT FAIL words do not control state"} - evidence := hotPathReviewEvidence{planRead: true, reviewRead: true, inspected: true} - if kind, cleanup, err := classifyHotPathReviewOutput("req_review", hotPathPhaseReviewActive, completion, evidence); err != nil || kind != "" || !cleanup { + if kind, cleanup, err := classifyHotPathReviewOutput("req_review", hotPathPhaseReviewActive, completion); err != nil || kind != "" || !cleanup { t.Fatalf("completion structure did not pass: kind=%q cleanup=%t err=%v", kind, cleanup, err) } repair := normalizedStageOutput{ Content: "PASS words do not control state", ToolCalls: []normalizedToolCall{{ID: "provider_repair", Name: "run_command", Arguments: map[string]any{"command": "go test"}}}, } - if kind, cleanup, err := classifyHotPathReviewOutput("req_review", hotPathPhaseReviewActive, repair, evidence); err != nil || kind != hotPathPendingReviewRepair || cleanup { + if kind, cleanup, err := classifyHotPathReviewOutput("req_review", hotPathPhaseReviewActive, repair); err != nil || kind != hotPathPendingReviewRepair || cleanup { t.Fatalf("repair structure did not stay active: kind=%q cleanup=%t err=%v", kind, cleanup, err) } } -func TestHotPathReviewRequiresInspectionAndNonEmptyTerminal(t *testing.T) { +func TestHotPathReviewRequiresOnlyNonEmptyTerminal(t *testing.T) { completion := normalizedStageOutput{Content: "reviewed"} - if _, _, err := classifyHotPathReviewOutput("req_review", hotPathPhaseReviewActive, completion, hotPathReviewEvidence{}); err == nil || !strings.Contains(err.Error(), "artifact reads") { - t.Fatalf("skipped inspection error = %v", err) + if kind, cleanup, err := classifyHotPathReviewOutput("req_review", hotPathPhaseReviewActive, completion); err != nil || kind != "" || !cleanup { + t.Fatalf("review completion: kind=%q cleanup=%t err=%v", kind, cleanup, err) } - evidence := hotPathReviewEvidence{planRead: true, reviewRead: true, inspected: true} - if _, _, err := classifyHotPathReviewOutput("req_review", hotPathPhaseReviewActive, normalizedStageOutput{}, evidence); err == nil || !strings.Contains(err.Error(), "non-empty") { + if _, _, err := classifyHotPathReviewOutput("req_review", hotPathPhaseReviewActive, normalizedStageOutput{}); err == nil || !strings.Contains(err.Error(), "non-empty") { t.Fatalf("empty terminal error = %v", err) } reservedRepair := normalizedStageOutput{ToolCalls: []normalizedToolCall{{ ID: "provider_second_review", Name: "read_file", Arguments: map[string]any{"path": newReservedPaths("req_review").ReviewPath}, }}} - if _, _, err := classifyHotPathReviewOutput("req_review", hotPathPhaseReviewRepair, reservedRepair, evidence); err == nil || !strings.Contains(err.Error(), "second review cycle") { - t.Fatalf("second review cycle error = %v", err) + if kind, cleanup, err := classifyHotPathReviewOutput("req_review", hotPathPhaseReviewRepair, reservedRepair); err != nil || kind != hotPathPendingReviewInspection || cleanup { + t.Fatalf("review artifact read: kind=%q cleanup=%t err=%v", kind, cleanup, err) } } diff --git a/apps/edge/internal/openai/hot_path_stage_input.go b/apps/edge/internal/openai/hot_path_stage_input.go index 653506c3..d78dbf4c 100644 --- a/apps/edge/internal/openai/hot_path_stage_input.go +++ b/apps/edge/internal/openai/hot_path_stage_input.go @@ -1,21 +1,15 @@ package openai import ( - "encoding/json" "fmt" "path/filepath" "strings" - "unicode" ) const hotPathReviewSystemPrompt = `You are the Reviewer in a compact Plan -> Work -> Review -> Repair pipeline. -Read the exact issued Plan and worker-filled Review before judging the task. Treat Review as worker evidence, not proof. -Inspect the actual caller-workspace result and rerun the Plan's applicable verification with ordinary caller tools. -Your first response must issue tool calls that read both exact artifact paths and inspect the actual caller-workspace result; do not return terminal prose first. -Check correctness, completeness, requirement coverage, verification trust, and unjustified deviations. -If a defect exists, establish its evidence and root cause, choose one concrete fix, repair it with ordinary caller tools, and reverify in this same Review stage. -Do not rewrite the reserved Plan or Review artifacts. Do not create a separate Result or final-review document. -Return a concise non-empty final result only after the result is verified; use no tool call in that terminal response.` +Read the issued Plan and worker Review, inspect the caller-workspace result, and run the applicable checks. +If you find a defect, repair it with ordinary caller tools and verify the result. +Do not create a separate Result or final-review document. Return a concise final result.` type hotPathArtifactPaths struct { PlanPath string @@ -30,19 +24,15 @@ type hotPathStageCorrelation struct { Terminal string } -// hotPathStageInput is the complete cross-stage input boundary. It contains -// only caller-owned immutable task text, issued paths, and committed -// provider correlations. Workspace contents, credentials, provider targets, -// and prior control prompts never enter this value. +// hotPathStageInput carries only the task and issued artifact paths needed by +// the next stage. type hotPathStageInput struct { - Role string - ImmutableTask string - Artifacts hotPathArtifactPaths - SelectorCommit hotPathStageCorrelation - LocalCommit hotPathStageCorrelation + Role string + ImmutableTask string + Artifacts hotPathArtifactPaths } -func buildLocalStageInput(task string, paths reservedPaths, selector hotPathStageCorrelation) hotPathStageInput { +func buildLocalStageInput(task string, paths reservedPaths) hotPathStageInput { return hotPathStageInput{ Role: "local", ImmutableTask: strings.TrimSpace(task), @@ -50,11 +40,10 @@ func buildLocalStageInput(task string, paths reservedPaths, selector hotPathStag PlanPath: paths.PlanPath, ReviewPath: paths.ReviewPath, }, - SelectorCommit: selector, } } -func buildReviewStageInput(task string, paths reservedPaths, selector, local hotPathStageCorrelation) hotPathStageInput { +func buildReviewStageInput(task string, paths reservedPaths) hotPathStageInput { return hotPathStageInput{ Role: "review", ImmutableTask: strings.TrimSpace(task), @@ -62,8 +51,6 @@ func buildReviewStageInput(task string, paths reservedPaths, selector, local hot PlanPath: paths.PlanPath, ReviewPath: paths.ReviewPath, }, - SelectorCommit: selector, - LocalCommit: local, } } @@ -74,48 +61,9 @@ func (in hotPathStageInput) validate() error { if cleanRelativePath(in.Artifacts.PlanPath) == "" || cleanRelativePath(in.Artifacts.ReviewPath) == "" { return fmt.Errorf("issued artifact paths are unavailable") } - if err := validateStageCorrelation("selector", in.SelectorCommit); err != nil { - return err - } - if in.Role == "review" { - if err := validateStageCorrelation("local", in.LocalCommit); err != nil { - return err - } - } return nil } -func validateStageCorrelation(role string, correlation hotPathStageCorrelation) error { - if !validLogicalRequestID(correlation.StageID) { - return fmt.Errorf("%s commit correlation StageID %q is invalid", role, correlation.StageID) - } - if !validOpaqueStageCorrelation(correlation.ResponseID) { - return fmt.Errorf("%s commit correlation ResponseID is invalid", role) - } - if !validLogicalRequestID(correlation.RunID) { - return fmt.Errorf("%s commit correlation RunID %q is invalid", role, correlation.RunID) - } - if !validOpaqueStageCorrelation(correlation.ProviderID) { - return fmt.Errorf("%s commit correlation ProviderID is invalid", role) - } - if !validOpaqueStageCorrelation(correlation.Terminal) { - return fmt.Errorf("%s commit correlation Terminal is invalid", role) - } - return nil -} - -func validOpaqueStageCorrelation(value string) bool { - if value == "" || len(value) > 256 { - return false - } - for _, r := range value { - if unicode.IsControl(r) { - return false - } - } - return true -} - func (in hotPathStageInput) prompt(phase hotPathLightPhase) (string, error) { if err := in.validate(); err != nil { return "", err @@ -131,10 +79,6 @@ func (in hotPathStageInput) prompt(phase hotPathLightPhase) (string, error) { var b strings.Builder b.WriteString("User task:\n") b.WriteString(in.ImmutableTask) - writeStageCorrelation(&b, "selector", in.SelectorCommit) - if in.Role == "review" { - writeStageCorrelation(&b, "local", in.LocalCommit) - } b.WriteString("\n\nIssued workspace artifacts:\n- plan: ") b.WriteString(in.Artifacts.PlanPath) b.WriteString("\n- review: ") @@ -198,31 +142,3 @@ func callerWorkingDirectory(task string) (string, error) { } return "", fmt.Errorf("caller workspace absolute path is unavailable") } - -type correlationPromptValue struct { - StageID string `json:"stage"` - ResponseID string `json:"response"` - RunID string `json:"run"` - ProviderID string `json:"provider"` - Terminal string `json:"terminal"` -} - -// writeStageCorrelation appends an immutable predecessor-success correlation -// block to the prompt builder. Correlation values are provider-visible but -// never carry credentials, provider targets, workspace file contents, or -// prior internal prompts. -func writeStageCorrelation(b *strings.Builder, role string, correlation hotPathStageCorrelation) { - fmt.Fprintf(b, "\nCommitted %s stage success:\n", role) - encoded, err := json.Marshal(correlationPromptValue{ - StageID: correlation.StageID, - ResponseID: correlation.ResponseID, - RunID: correlation.RunID, - ProviderID: correlation.ProviderID, - Terminal: correlation.Terminal, - }) - if err != nil { - return - } - b.Write(encoded) - b.WriteString("\n") -} diff --git a/apps/edge/internal/openai/hot_path_terminal_control_test.go b/apps/edge/internal/openai/hot_path_terminal_control_test.go index f28ecb92..a20fee3e 100644 --- a/apps/edge/internal/openai/hot_path_terminal_control_test.go +++ b/apps/edge/internal/openai/hot_path_terminal_control_test.go @@ -991,10 +991,9 @@ func TestHotPathRejectedDispatchSelectorMatrix(t *testing.T) { func rejectedStageSnapshot(stream bool) hotPathDispatchSnapshot { paths := newReservedPaths("req-stage-reject") - selector := hotPathStageCorrelation{StageID: "stg-s", ResponseID: "r:s/1", RunID: "run-s", ProviderID: "p", Terminal: "t"} return hotPathDispatchSnapshot{ Protocol: "openai", Stream: stream, StageID: "stage-r", Stage: config.ExecutionRouteStage{Model: "m"}, - Input: buildLocalStageInput(scriptedAbsoluteWorkspaceTask, paths, selector), + Input: buildLocalStageInput(scriptedAbsoluteWorkspaceTask, paths), Route: routeDispatch{NodeRef: "node-stage", ProviderID: "p", Adapter: "a-stage", Target: "t-stage", SessionID: "s-stage", TimeoutSec: 5, ProviderPool: true}, } } diff --git a/apps/edge/internal/openai/request_coordinator.go b/apps/edge/internal/openai/request_coordinator.go index 1a7ec4df..efc2c905 100644 --- a/apps/edge/internal/openai/request_coordinator.go +++ b/apps/edge/internal/openai/request_coordinator.go @@ -427,9 +427,9 @@ func (c *logicalRequestCoordinator) consumeContinuationByLineage(ownerEdgeID, pr return target.snapshot(), nil } -func (c *logicalRequestCoordinator) consumeArtifactContinuationByLineage( +func (c *logicalRequestCoordinator) consumePendingResults( requestID, ownerEdgeID, principalRef string, - lineage logicalRequestContinuationLineage, + resultIDs []string, ) (logicalRequestSnapshot, error) { c.mu.Lock() defer c.mu.Unlock() @@ -443,19 +443,15 @@ func (c *logicalRequestCoordinator) consumeArtifactContinuationByLineage( if record.principalRef != principalRef { return logicalRequestSnapshot{}, errLogicalRequestPrincipal } - if record.lineage.Endpoint != lineage.Prefix.Endpoint || record.lineage.ToolsetDigest != lineage.Prefix.ToolsetDigest { - return logicalRequestSnapshot{}, errLogicalRequestLineage - } if record.expected == nil { return logicalRequestSnapshot{}, errLogicalRequestNoFrontier } - if record.expectedIssuedCallHash != lineage.IssuedCallHash || !sameLogicalRequestResultIDs(record.expected, lineage.ResultIDs) { + if !sameLogicalRequestResultIDs(record.expected, resultIDs) { return logicalRequestSnapshot{}, errLogicalRequestFrontier } record.expected = nil record.expectedIssuedCallHash = "" - record.lineage = lineage.Committed record.activeStageID = "" record.state = logicalRequestStateResumed record.updatedAt = c.now() diff --git a/apps/edge/internal/openai/request_identity_ingress.go b/apps/edge/internal/openai/request_identity_ingress.go index 88623faf..543f0600 100644 --- a/apps/edge/internal/openai/request_identity_ingress.go +++ b/apps/edge/internal/openai/request_identity_ingress.go @@ -31,10 +31,6 @@ func (s *Server) hotPathSelectorProviderInstruction(metadata map[string]string) func (s *Server) joinPresetChatIngress(r *http.Request, dispatch routeDispatch, rawBody []byte, runMeta map[string]string) (presetIngressResult, error) { delete(runMeta, hotPathInitialAdmissionMetadata) s.sweepLogicalRequestTTL() - requestContext := context.Background() - if r != nil { - requestContext = r.Context() - } ownerEdgeID := s.edgeIDValue() principalRef := runMeta[principalMetaRef] if principalRef == "" { @@ -78,27 +74,12 @@ func (s *Server) joinPresetChatIngress(r *http.Request, dispatch routeDispatch, } if s.artifactFrontiers != nil { snap, disposition, matched, err := s.artifactFrontiers.consumeChat( - ownerEdgeID, principalRef, rawBody, contLineage, s.requestCoordinator, s.lightFlows, + ownerEdgeID, principalRef, rawBody, contLineage, s.requestCoordinator, ) if matched { if err != nil { return presetIngressResult{}, fmt.Errorf("artifact continuation rejected: %w", err) } - if disposition.PrimaryError != nil { - if err := s.lightFlows.updateArtifactLineage(snap.ID, ownerEdgeID, contLineage.Committed, false); err != nil { - return presetIngressResult{}, err - } - cleanup, err := s.lightFlows.beginPrimaryErrorCleanup(requestContext, snap.ID, ownerEdgeID, *disposition.PrimaryError, nil, s.requestCoordinator) - if err != nil { - if contextErr := requestContext.Err(); contextErr != nil { - return presetIngressResult{}, contextErr - } - runMeta["iop_logical_request_id"] = snap.ID - return presetIngressResult{Terminal: s.retainHotPathPrimaryErrorForTTL(snap.ID, *disposition.PrimaryError)}, nil - } - s.observeHotPathCleanupTransition(requestContext, snap.ID, dispatch.Preset.ID) - return presetIngressResult{Cleanup: &hotPathCleanupTurn{RequestID: snap.ID, Output: cleanup}}, nil - } if err := s.applyArtifactDisposition(snap, disposition, runMeta); err != nil { return presetIngressResult{}, err } @@ -182,10 +163,6 @@ func (s *Server) joinPresetChatIngress(r *http.Request, dispatch routeDispatch, func (s *Server) joinPresetAnthropicIngress(r *http.Request, dispatch routeDispatch, rawBody []byte, metadata map[string]string) (presetIngressResult, error) { delete(metadata, hotPathInitialAdmissionMetadata) s.sweepLogicalRequestTTL() - requestContext := context.Background() - if r != nil { - requestContext = r.Context() - } ownerEdgeID := s.edgeIDValue() principalRef := metadata[principalMetaRef] if principalRef == "" { @@ -229,27 +206,12 @@ func (s *Server) joinPresetAnthropicIngress(r *http.Request, dispatch routeDispa } if s.artifactFrontiers != nil { snap, disposition, matched, err := s.artifactFrontiers.consumeAnthropic( - ownerEdgeID, principalRef, rawBody, contLineage, s.requestCoordinator, s.lightFlows, + ownerEdgeID, principalRef, rawBody, contLineage, s.requestCoordinator, ) if matched { if err != nil { return presetIngressResult{}, fmt.Errorf("artifact continuation rejected: %w", err) } - if disposition.PrimaryError != nil { - if err := s.lightFlows.updateArtifactLineage(snap.ID, ownerEdgeID, contLineage.Committed, false); err != nil { - return presetIngressResult{}, err - } - cleanup, err := s.lightFlows.beginPrimaryErrorCleanup(requestContext, snap.ID, ownerEdgeID, *disposition.PrimaryError, nil, s.requestCoordinator) - if err != nil { - if contextErr := requestContext.Err(); contextErr != nil { - return presetIngressResult{}, contextErr - } - metadata["iop_logical_request_id"] = snap.ID - return presetIngressResult{Terminal: s.retainHotPathPrimaryErrorForTTL(snap.ID, *disposition.PrimaryError)}, nil - } - s.observeHotPathCleanupTransition(requestContext, snap.ID, dispatch.Preset.ID) - return presetIngressResult{Cleanup: &hotPathCleanupTurn{RequestID: snap.ID, Output: cleanup}}, nil - } if err := s.applyArtifactDisposition(snap, disposition, metadata); err != nil { return presetIngressResult{}, err }