Fix Codex borrow observations and cancellation

.downstream/native-domains.md

@@ -118,6 +118,11 @@ Status adds `setup`, `observed_since` and `recent_usage`. The last actual dispat
 and in-process counters are retained per configured target/model/protocol/purpose;
 configuration publication resets this index. HTTP/SSE, pooled WS and native WS
 relay observations record application independently from upstream completion.
+HTTP application is checked against the immutable preparation proof and final
+cookie, including after restoring the caller context. A parsed successful terminal
+and visible text remain authoritative when read-ahead cleanup later reports
+cancellation; that detail remains available as `read_error`. Captures are snapshotted
+under a lock, and the usage index preserves the first terminal event.
 Diagnostic and business requests are distinguished. This index contains no request
 payloads or credential/cookie values; the existing administrator error records remain.
 

.github/workflows/downstream-verify.yml

@@ -49,7 +49,7 @@ jobs:
           go test -p=1 -tags integration ./internal/repository -run '^TestCodexGatewayBorrowTestRepositoryIntegrationLifecycle$' -count=1
       - name: Codex preparation coordination race tests
         working-directory: backend
-        run: go test -p=1 -race ./internal/service -run '^TestCodexBorrow(Flight|DifferentTargets)' -count=1
+        run: go test -p=1 -race -tags unit ./internal/service -run '^TestCodexBorrow(Flight|DifferentTargets|CancelledParent|DiagnoseHTTPTerminalAndDispatchEvidence)|^TestCodexGatewayBorrowRunnerCancellationStopsQueuedAndInflightAndRetainsRawOutput$' -count=1
       - name: Upstream automation classifier tests
         run: |
           python3 .github/scripts/test_upstream_risk.py

backend/internal/service/codex_gateway_borrow_coordination_test.go

@@ -161,3 +161,52 @@ func TestCodexBorrowDistinctRequestProofsRemainReusable(t *testing.T) {
 	require.NoError(t, err)
 	require.True(t, applied.Applied, "another request's validation must not replace this proof")
 }
+
+func TestCodexBorrowUsageKeepsFirstTerminal(t *testing.T) {
+	completed := []byte(`{"type":"response.completed","response":{"status":"completed"}}`)
+	failed := []byte(`{"type":"error","error":{"code":"synthetic_failure"}}`)
+	for _, firstError := range []bool{false, true} {
+		s := newBorrowCoreTest(t, nil)
+		tracker := s.beginUsage(context.Background(), 2, "gpt-6-astra", "http", true)
+		if firstError {
+			tracker.observe(failed)
+			tracker.observe(completed)
+		} else {
+			tracker.observe(completed)
+			tracker.observe(failed)
+		}
+		tracker.finish(context.Canceled)
+		want := "completed"
+		if firstError {
+			want = "upstream_error"
+		}
+		require.Equal(t, want, s.Status().RecentUsage[0].Outcome)
+	}
+}
+
+func TestCodexBorrowHTTPAppliedChecksFinalCookie(t *testing.T) {
+	req, err := http.NewRequest(http.MethodPost, chatgptCodexURL, nil)
+	require.NoError(t, err)
+	proof := codexGatewayBorrowHTTPPreparation{application: &CodexGatewayBorrowApplication{Applied: true, CookieFingerprint: borrowHash("qualified-cookie")}}
+	req = req.WithContext(context.WithValue(req.Context(), codexGatewayBorrowHTTPPreparationContextKey{}, proof))
+	require.False(t, codexBorrowHTTPApplied(req))
+	req.Header.Set("Cookie", "__oailb=changed-cookie")
+	require.False(t, codexBorrowHTTPApplied(req))
+	req.Header.Set("Cookie", "__oailb=qualified-cookie")
+	require.True(t, codexBorrowHTTPApplied(req))
+}
+
+func TestCodexBorrowCancelledParentCannotAdmitWaitingChild(t *testing.T) {
+	parent, cancel := context.WithCancel(context.Background())
+	coordinator := newPelicanExecutionCoordinator(1)
+	state := newPelicanExecutionState(parent, coordinator, time.Minute, time.Now)
+	defer state.finish()
+	// Represent the cancellation propagation gap: the parent is cancelled,
+	// but this child's cancellation signal is not visible to its waiter yet.
+	ctx := context.WithValue(context.WithoutCancel(state.parent), pelicanExecutionContextKey{}, state)
+	cancel()
+	_, release, err := AcquirePelicanExecution(ctx, 2)
+	require.ErrorIs(t, err, context.Canceled)
+	require.Nil(t, release)
+	require.Zero(t, coordinator.active)
+}

backend/internal/service/codex_gateway_borrow_diagnose.go

@@ -74,6 +74,7 @@ type CodexBorrowDiagnosticRequest struct {
 }
 
 type CodexBorrowDiagnosticResult struct {
+	ReadError     string `json:"read_error,omitempty"`
 	Dispatched    bool   `json:"dispatched"`
 	Mode          string `json:"mode"`
 	Turn          int    `json:"turn"`
@@ -208,12 +209,7 @@ func (s *CodexGatewayBorrowService) diagnoseTurn(ctx context.Context, account *A
 		result.ResponseID = firstNonEmpty(forward.ResponseID, forward.RequestID)
 		result.ReportedModel = forward.UpstreamResponseModel
 	}
-	capture.mu.Lock()
-	var last *pelicanHTTPAttempt
-	if len(capture.attempts) > 0 {
-		last = capture.attempts[len(capture.attempts)-1]
-	}
-	capture.mu.Unlock()
+	last := snapshotPelicanAttempt(capture)
 	if last == nil {
 		result.RawResponse = w.Body.String()
 		if sendErr != nil {
@@ -224,6 +220,9 @@ func (s *CodexGatewayBorrowService) diagnoseTurn(ctx context.Context, account *A
 		return result
 	}
 	result.Dispatched = last.dispatched
+	if last.err != nil {
+		result.ReadError = last.err.Error()
+	}
 	result.RawResponse = last.raw.String()
 	if last.wsFrames && last.raw.Len() == 0 && sendErr != nil {
 		var eventErr *openAIWSUpstreamEventError
@@ -246,7 +245,9 @@ func (s *CodexGatewayBorrowService) diagnoseTurn(ctx context.Context, account *A
 	state.mu.Lock()
 	result.Applied = state.invocation.BorrowApplied
 	state.mu.Unlock()
-	result.Completed = sendErr == nil && parseErr == nil && !limited && last.err == nil && last.status < 400 && strings.TrimSpace(result.Answer) != ""
+	// A valid completed terminal with text is authoritative. Read-ahead cleanup
+	// (often context cancellation after that terminal) cannot turn it into failure.
+	result.Completed = sendErr == nil && parseErr == nil && !limited && last.status >= 200 && last.status < 300 && strings.TrimSpace(result.Answer) != ""
 	if sendErr != nil {
 		result.Error = sendErr.Error()
 	} else if parseErr != nil {

backend/internal/service/codex_gateway_borrow_diagnose_test.go

@@ -5,14 +5,81 @@ package service
 import (
 	"context"
 	"encoding/json"
+	"io"
 	"net/http"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint"
 	"github.com/stretchr/testify/require"
 )
 
+// A real HTTP stream can deliver its final bytes together with cancellation
+// while the gateway shuts down read-ahead after committing response.completed.
+type borrowDiagnosticCancelledTail struct{ *strings.Reader }
+
+func (r *borrowDiagnosticCancelledTail) Read(p []byte) (int, error) {
+	n, err := r.Reader.Read(p)
+	if r.Reader.Len() == 0 {
+		return n, context.Canceled
+	}
+	return n, err
+}
+func (*borrowDiagnosticCancelledTail) Close() error { return nil }
+
+func TestCodexBorrowDiagnoseHTTPTerminalAndDispatchEvidence(t *testing.T) {
+	for _, complete := range []bool{true, false} {
+		t.Run(map[bool]string{true: "completed then read cancellation", false: "missing terminal"}[complete], func(t *testing.T) {
+			account := borrowCoreAccount(2)
+			generator, repo := newPelicanGeneratorForTest(account, &pelicanGeneratorUpstream{})
+			gateway := generator.openaiGatewayService
+			raw := borrowCoreSSE("gpt-6-astra", "OK")
+			if !complete {
+				raw = "data: {\"type\":\"response.output_text.delta\",\"delta\":\"OK\"}\n\n"
+			}
+			gateway.httpUpstream = borrowCoreProbe(func(req *http.Request, _ string, id int64, n int, _ *tlsfingerprint.Profile) (*http.Response, error) {
+				_, finish, err := PreparePelicanHTTPRequest(req, id, n, "http")
+				if err != nil {
+					return nil, err
+				}
+				return finish(&http.Response{StatusCode: 200, Header: http.Header{"Content-Type": {"text/event-stream"}}, Body: &borrowDiagnosticCancelledTail{strings.NewReader(raw)}}, nil)
+			})
+			s := NewCodexGatewayBorrowService(nil, repo, gateway, borrowCoreProbe(func(*http.Request, string, int64, int, *tlsfingerprint.Profile) (*http.Response, error) {
+				return borrowCoreResponse("gpt-6-astra", "OK", "stable", ""), nil
+			}), nil)
+			s.publishConfig(CodexGatewayBorrowConfig{Enabled: true, SourceAccountIDs: []int64{1}, TargetAccountIDs: []int64{2}, Models: []string{"gpt-6-astra"}}, false)
+			defer s.Stop()
+			borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL))
+			var results []CodexBorrowDiagnosticResult
+			err := s.Diagnose(context.Background(), CodexBorrowDiagnosticRequest{AccountID: 2, Model: "gpt-6-astra", Transport: "http"}, func(e CodexBorrowDiagnosticEvent) {
+				if e.Result != nil {
+					results = append(results, *e.Result)
+				}
+			})
+			require.NoError(t, err)
+			require.Len(t, results, 2)
+			for _, r := range results {
+				require.Equal(t, complete, r.Completed, r.Error)
+				require.True(t, r.Dispatched)
+			}
+			require.False(t, results[0].Applied)
+			require.True(t, results[1].Applied)
+			usage := s.Status().RecentUsage
+			require.Len(t, usage, 1)
+			require.True(t, usage[0].Applied)
+			require.EqualValues(t, 1, usage[0].AppliedCount)
+			if complete {
+				require.Equal(t, "completed", usage[0].Outcome)
+			} else {
+				require.NotEqual(t, "completed", usage[0].Outcome)
+			}
+		})
+	}
+}
+
+var _ io.ReadCloser = (*borrowDiagnosticCancelledTail)(nil)
+
 func TestCodexBorrowDiagnoseHTTPUsesRealSenderAndIsolatesModes(t *testing.T) {
 	account := borrowCoreAccount(2)
 	upstream := &pelicanGeneratorUpstream{body: pelicanResponsesSSE("gpt-6-astra", "completed", "")}

backend/internal/service/codex_gateway_borrow_http_observation.go

@@ -0,0 +1,21 @@
+package service
+
+import "net/http"
+
+// Prepared forwarding restores the caller's context after qualification. Its
+// immutable application proof survives that restoration; the transient Apply
+// marker does not. Compare the proof against the cookie at the actual boundary.
+func codexBorrowHTTPApplied(req *http.Request) bool {
+	if req == nil {
+		return false
+	}
+	cookie, err := req.Cookie("__oailb")
+	if err != nil || cookie.Value == "" {
+		return false
+	}
+	if prepared, ok := req.Context().Value(codexGatewayBorrowHTTPPreparationContextKey{}).(codexGatewayBorrowHTTPPreparation); ok {
+		application := prepared.application
+		return application != nil && application.Applied && application.CookieFingerprint == borrowHash(cookie.Value)
+	}
+	return req.Context().Value(codexBorrowAppliedContextKey{}) == true
+}

backend/internal/service/codex_gateway_borrow_usage.go

@@ -66,6 +66,9 @@ func (s *CodexGatewayBorrowService) beginUsage(ctx context.Context, accountID in
 }
 
 type codexBorrowUsageTracker struct {
+	mu       sync.Mutex
+	terminal bool
+	finished bool
 	revision uint64
 	service  *CodexGatewayBorrowService
 	key      codexBorrowUsageKey
@@ -77,6 +80,11 @@ func (t *codexBorrowUsageTracker) observe(message []byte) {
 	if t == nil {
 		return
 	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if t.finished || t.terminal {
+		return
+	}
 	typeName := gjson.GetBytes(message, "type").String()
 	if id := gjson.GetBytes(message, "response.id").String(); id != "" {
 		t.row.RequestID = id
@@ -86,6 +94,7 @@ func (t *codexBorrowUsageTracker) observe(message []byte) {
 	}
 	switch typeName {
 	case "response.completed", "response.done":
+		t.terminal = true
 		if gjson.GetBytes(message, "response.status").String() == "completed" && !gjson.GetBytes(message, "response.error").IsObject() {
 			t.row.Outcome = "completed"
 		} else if gjson.GetBytes(message, "response.status").String() != "" || gjson.GetBytes(message, "response.error").IsObject() {
@@ -94,6 +103,7 @@ func (t *codexBorrowUsageTracker) observe(message []byte) {
 			t.row.Outcome = "unobserved"
 		}
 	case "error", "response.failed", "response.incomplete", "response.cancelled", "response.canceled":
+		t.terminal = true
 		t.row.Outcome = "upstream_error"
 	}
 }
@@ -103,23 +113,29 @@ func (t *codexBorrowUsageTracker) finish(err error) {
 		return
 	}
 	t.once.Do(func() {
-		if err != nil && t.row.Outcome != "completed" {
-			t.row.Outcome = "transport_error"
-		} else if t.row.Outcome == "sent" {
-			t.row.Outcome = "incomplete"
+		t.mu.Lock()
+		if !t.terminal {
+			if err != nil {
+				t.row.Outcome = "transport_error"
+			} else {
+				t.row.Outcome = "incomplete"
+			}
 		}
 		now := time.Now()
 		t.row.FinishedAt = &now
+		t.finished = true
+		row := t.row
+		t.mu.Unlock()
 		t.service.mu.Lock()
 		defer t.service.mu.Unlock()
-		// An older completion must not overwrite a newer attempt or cleared config.
-		if current, ok := t.service.usage[t.key]; ok && t.service.revision == t.revision && current.Count == t.row.Count {
-			t.service.usage[t.key] = t.row
+		if current, ok := t.service.usage[t.key]; ok && t.service.revision == t.revision && current.Count == row.Count {
+			t.service.usage[t.key] = row
 		}
 	})
 }
 
 type codexBorrowUsageBody struct {
+	mu sync.Mutex
 	io.ReadCloser
 	tracker *codexBorrowUsageTracker
 	line    []byte
@@ -128,6 +144,8 @@ type codexBorrowUsageBody struct {
 
 func (b *codexBorrowUsageBody) Read(p []byte) (int, error) {
 	n, err := b.ReadCloser.Read(p)
+	b.mu.Lock()
+	defer b.mu.Unlock()
 	for _, ch := range p[:n] {
 		if ch == '\n' {
 			if strings.HasPrefix(string(b.line), "data:") {
@@ -153,6 +171,8 @@ func (b *codexBorrowUsageBody) Read(p []byte) (int, error) {
 
 func (b *codexBorrowUsageBody) Close() error {
 	err := b.ReadCloser.Close()
+	b.mu.Lock()
+	defer b.mu.Unlock()
 	if b.err == nil {
 		b.err = err
 	}
@@ -168,14 +188,26 @@ func (s *CodexGatewayBorrowService) trackHTTPResponse(t *codexBorrowUsageTracker
 		t.finish(err)
 		return resp, err
 	}
+	t.mu.Lock()
 	t.row.RequestID = resp.Header.Get("X-Request-ID")
 	if resp.StatusCode >= 400 {
 		t.row.Outcome = "upstream_error"
+		t.terminal = true
 	}
+	t.mu.Unlock()
 	if err != nil || resp.Body == nil {
 		t.finish(err)
 	} else {
 		resp.Body = &codexBorrowUsageBody{ReadCloser: resp.Body, tracker: t}
 	}
 	return resp, err
 }
+
+func (t *codexBorrowUsageTracker) terminalObserved() bool {
+	if t == nil {
+		return false
+	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.terminal
+}

backend/internal/service/codex_gateway_borrow_usage_ws.go

@@ -43,7 +43,7 @@ func (c *codexBorrowUsageFrameConn) ReadFrame(ctx context.Context) (coderws.Mess
 	c.current.observe(payload)
 	if err != nil {
 		c.current.finish(err)
-	} else if c.current != nil && c.current.row.Outcome != "sent" {
+	} else if c.current.terminalObserved() {
 		c.current.finish(nil)
 	}
 	c.mu.Unlock()

backend/internal/service/openai_codex_request_compression.go

@@ -157,7 +157,7 @@ func (s *OpenAIGatewayService) doOpenAICodexUpstream(req *http.Request, account
 	if err != nil {
 		return nil, err
 	}
-	applied := wire.Context().Value(codexBorrowAppliedContextKey{}) == true
+	applied := codexBorrowHTTPApplied(wire)
 	if err := consumeBorrowDiagnosticRequest(wire.Context()); err != nil {
 		return nil, err
 	}

backend/internal/service/openai_ws_forwarder_v2.go

@@ -447,7 +447,10 @@ func (s *OpenAIGatewayService) forwardOpenAIWSV2(
 	if capture := pelicanCaptureFromContext(ctx); capture != nil {
 		capture.mu.Lock()
 		if len(capture.attempts) > 0 {
-			capture.attempts[len(capture.attempts)-1].dispatched = true
+			attempt := capture.attempts[len(capture.attempts)-1]
+			attempt.mu.Lock()
+			attempt.dispatched = true
+			attempt.mu.Unlock()
 		}
 		capture.mu.Unlock()
 	}