toki/iop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(openai): hybrid 검증 경계를 최소화한다

Browse source
This commit is contained in:
toki 2026-08-15 16:51:24 +09:00
parent 19b555c34d
commit eee4f883e7
14 changed files with 110 additions and 902 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

137
apps/edge/internal/openai/artifact_pair.go
View file

@ -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 {

30
apps/edge/internal/openai/artifact_pair_test.go
View file

@ -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)
}
})
}
}

62
apps/edge/internal/openai/hot_path_cleanup.go
View file

@ -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

96
apps/edge/internal/openai/hot_path_cleanup_test.go
View file

@ -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,

8
apps/edge/internal/openai/hot_path_dispatch.go
View file

@ -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
}

203
apps/edge/internal/openai/hot_path_light.go
View file

@ -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 {

192
apps/edge/internal/openai/hot_path_light_test.go
View file

@ -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

28
apps/edge/internal/openai/hot_path_observation_test.go
View file

@ -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

79
apps/edge/internal/openai/hot_path_review.go
View file

@ -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
}

18
apps/edge/internal/openai/hot_path_review_test.go
View file

@ -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)
}
}

104
apps/edge/internal/openai/hot_path_stage_input.go
View file

@ -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")
}

3
apps/edge/internal/openai/hot_path_terminal_control_test.go
View file

@ -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},
}
}

10
apps/edge/internal/openai/request_coordinator.go
View file

@ -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()

42
apps/edge/internal/openai/request_identity_ingress.go
View file

@ -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
}