diff --git a/.downstream/native-domains.md b/.downstream/native-domains.md index 05f928193937..927094699dbc 100644 --- a/.downstream/native-domains.md +++ b/.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. diff --git a/.github/workflows/downstream-verify.yml b/.github/workflows/downstream-verify.yml index 51d944a4d009..a6246bc3525d 100644 --- a/.github/workflows/downstream-verify.yml +++ b/.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 diff --git a/backend/internal/service/codex_gateway_borrow_coordination_test.go b/backend/internal/service/codex_gateway_borrow_coordination_test.go index 43eaeded7b86..f8cd58665831 100644 --- a/backend/internal/service/codex_gateway_borrow_coordination_test.go +++ b/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) +} diff --git a/backend/internal/service/codex_gateway_borrow_diagnose.go b/backend/internal/service/codex_gateway_borrow_diagnose.go index 82c1a75fde92..633831155402 100644 --- a/backend/internal/service/codex_gateway_borrow_diagnose.go +++ b/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 { diff --git a/backend/internal/service/codex_gateway_borrow_diagnose_test.go b/backend/internal/service/codex_gateway_borrow_diagnose_test.go index d9af9417cae4..6d7aca0fd5d6 100644 --- a/backend/internal/service/codex_gateway_borrow_diagnose_test.go +++ b/backend/internal/service/codex_gateway_borrow_diagnose_test.go @@ -5,7 +5,9 @@ package service import ( "context" "encoding/json" + "io" "net/http" + "strings" "testing" "time" @@ -13,6 +15,71 @@ import ( "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", "")} diff --git a/backend/internal/service/codex_gateway_borrow_http_observation.go b/backend/internal/service/codex_gateway_borrow_http_observation.go new file mode 100644 index 000000000000..dd81075f45fd --- /dev/null +++ b/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 +} diff --git a/backend/internal/service/codex_gateway_borrow_usage.go b/backend/internal/service/codex_gateway_borrow_usage.go index a76bbd6ae844..7cfca73202a5 100644 --- a/backend/internal/service/codex_gateway_borrow_usage.go +++ b/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,10 +188,13 @@ 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 { @@ -179,3 +202,12 @@ func (s *CodexGatewayBorrowService) trackHTTPResponse(t *codexBorrowUsageTracker } return resp, err } + +func (t *codexBorrowUsageTracker) terminalObserved() bool { + if t == nil { + return false + } + t.mu.Lock() + defer t.mu.Unlock() + return t.terminal +} diff --git a/backend/internal/service/codex_gateway_borrow_usage_ws.go b/backend/internal/service/codex_gateway_borrow_usage_ws.go index 963282bd12e5..080cb1d2282c 100644 --- a/backend/internal/service/codex_gateway_borrow_usage_ws.go +++ b/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() diff --git a/backend/internal/service/openai_codex_request_compression.go b/backend/internal/service/openai_codex_request_compression.go index d426ea0a8b18..4bb36b9b0e64 100644 --- a/backend/internal/service/openai_codex_request_compression.go +++ b/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 } diff --git a/backend/internal/service/openai_ws_forwarder_v2.go b/backend/internal/service/openai_ws_forwarder_v2.go index 62399ac3fb7c..5e02102b3d76 100644 --- a/backend/internal/service/openai_ws_forwarder_v2.go +++ b/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() } diff --git a/backend/internal/service/pelican_account_generator.go b/backend/internal/service/pelican_account_generator.go index d199d36b2dce..13764daa5cb5 100644 --- a/backend/internal/service/pelican_account_generator.go +++ b/backend/internal/service/pelican_account_generator.go @@ -89,6 +89,7 @@ type pelicanGenerationCapture struct { } type pelicanHTTPAttempt struct { + mu sync.Mutex dispatched bool protocol string contentType string @@ -189,10 +190,7 @@ func PreparePelicanHTTPRequest(req *http.Request, accountID int64, accountConcur firstOutputPolicy, _ := ctx.Value(pelicanFirstOutputPolicyContextKey{}).(*pelicanFirstOutputPolicy) ctx = firstOutputPolicy.arm(ctx) req = req.WithContext(ctx) - borrowApplied := ctx.Value(codexBorrowAppliedContextKey{}) == true - if prepared, ok := ctx.Value(codexGatewayBorrowHTTPPreparationContextKey{}).(codexGatewayBorrowHTTPPreparation); ok { - borrowApplied = prepared.application != nil && prepared.application.Applied - } + borrowApplied := codexBorrowHTTPApplied(req) invocation := PelicanInvocation{Platform: capture.account.Platform, Model: model, Effort: effort, Endpoint: req.URL.Scheme + "://" + req.URL.Host + req.URL.EscapedPath(), Protocol: protocol, Transport: transport, BorrowApplied: borrowApplied} RecordPelicanInvocation(ctx, invocation) @@ -201,6 +199,8 @@ func PreparePelicanHTTPRequest(req *http.Request, accountID int64, accountConcur capture.attempts = append(capture.attempts, attempt) capture.mu.Unlock() finish = func(resp *http.Response, sendErr error) (*http.Response, error) { + attempt.mu.Lock() + defer attempt.mu.Unlock() firstOutputPolicy.stopHeaderWait() attempt.err = sendErr if resp == nil || resp.Body == nil { @@ -285,6 +285,8 @@ func preparePelicanWSSend(ctx context.Context, account *Account, endpoint string capture.attempts = append(capture.attempts, attempt) capture.mu.Unlock() observe = func(message []byte) error { + attempt.mu.Lock() + defer attempt.mu.Unlock() if attempt.raw.Len()+len(message)+1 > int(codexGatewayBorrowPelicanMaxBody) { attempt.err = errors.New("upstream response exceeds the pelican record size limit") return attempt.err @@ -305,6 +307,8 @@ type pelicanCapturingBody struct { func (b *pelicanCapturingBody) Read(data []byte) (int, error) { n, err := b.ReadCloser.Read(data) + b.attempt.mu.Lock() + defer b.attempt.mu.Unlock() if n > 0 { if b.attempt.raw.Len()+n > int(codexGatewayBorrowPelicanMaxBody) { b.attempt.err = errors.New("upstream response exceeds the pelican record size limit") @@ -443,12 +447,7 @@ func (s *AccountTestService) GeneratePelican(ctx context.Context, accountID int6 // This represents the server's internal generation operation. No client // UA, protocol handshake, API-key subject or group policy is synthesized. sendErr := s.sendPelicanWithAccount(ctx, c, account, model, option.UpstreamModel, effort) - 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 { @@ -668,3 +667,23 @@ func parsePelicanWSFrames(raw string) (answer string, limited bool, model string } return text.String(), p.limited, p.upstreamModel, p.incomplete() } + +// Forward can commit a terminal event while its read-ahead goroutine is being +// cancelled. Snapshot under the attempt lock; later read cleanup cannot race +// diagnostic parsing or change the bytes that were already observed. +func snapshotPelicanAttempt(capture *pelicanGenerationCapture) *pelicanHTTPAttempt { + capture.mu.Lock() + var last *pelicanHTTPAttempt + if len(capture.attempts) > 0 { + last = capture.attempts[len(capture.attempts)-1] + } + capture.mu.Unlock() + if last == nil { + return nil + } + last.mu.Lock() + defer last.mu.Unlock() + snapshot := &pelicanHTTPAttempt{protocol: last.protocol, contentType: last.contentType, status: last.status, model: last.model, effort: last.effort, err: last.err, wsFrames: last.wsFrames, dispatched: last.dispatched} + _, _ = snapshot.raw.Write(last.raw.Bytes()) + return snapshot +} diff --git a/backend/internal/service/pelican_execution.go b/backend/internal/service/pelican_execution.go index 167df5903cd9..8d83d3132fd0 100644 --- a/backend/internal/service/pelican_execution.go +++ b/backend/internal/service/pelican_execution.go @@ -131,6 +131,7 @@ type pelicanExecutionSnapshot struct { } type pelicanExecutionState struct { + admission context.Context mu sync.Mutex coordinator *pelicanExecutionCoordinator parent context.Context @@ -153,7 +154,7 @@ type pelicanExecutionState struct { func newPelicanExecutionState(parent context.Context, coordinator *pelicanExecutionCoordinator, budget time.Duration, now func() time.Time) *pelicanExecutionState { operationCtx, operationCancel := context.WithCancelCause(parent) - return &pelicanExecutionState{parent: operationCtx, operationCancel: operationCancel, coordinator: coordinator, budget: budget, now: now, + return &pelicanExecutionState{admission: parent, parent: operationCtx, operationCancel: operationCancel, coordinator: coordinator, budget: budget, now: now, withTimeout: context.WithTimeout} } @@ -188,6 +189,11 @@ func AcquirePelicanExecution(ctx context.Context, accountID int64) (context.Cont if err := ctx.Err(); err != nil { return ctx, nil, err } + if state.admission != nil { + if err := state.admission.Err(); err != nil { + return ctx, nil, err + } + } if lease, _ := ctx.Value(pelicanExecutionLeaseContextKey{}).(*pelicanExecutionLease); lease != nil && lease.active.Load() { if lease.accountID == accountID { return ctx, func() {}, nil @@ -203,6 +209,16 @@ func AcquirePelicanExecution(ctx context.Context, accountID int64) (context.Cont if err != nil { return ctx, nil, err } + // A task cancellation marks its parent before walking the per-target + // children. A released slot can wake a sibling during that walk, while the + // sibling's own Err is still nil. Do not admit that cancelled task. + if state.admission != nil { + if err := state.admission.Err(); err != nil { + release() + return ctx, nil, err + } + } + lease := &pelicanExecutionLease{accountID: accountID} lease.active.Store(true) return context.WithValue(ctx, pelicanExecutionLeaseContextKey{}, lease), func() { diff --git a/frontend/src/api/admin/codexGatewayBorrow.ts b/frontend/src/api/admin/codexGatewayBorrow.ts index 20742638e51f..efbb5e4e60d0 100644 --- a/frontend/src/api/admin/codexGatewayBorrow.ts +++ b/frontend/src/api/admin/codexGatewayBorrow.ts @@ -39,6 +39,7 @@ export interface CodexGatewayBorrowUsage { } export interface CodexBorrowDiagnosticResult { dispatched?: boolean + read_error?: string mode: string; turn: number; applied: boolean; completed: boolean; response_id?: string reported_model?: string; answer: string; raw_response: string; error?: string; duration_ms: number } diff --git a/frontend/src/components/admin/codex/CodexBorrowActivity.vue b/frontend/src/components/admin/codex/CodexBorrowActivity.vue index ec2ebaf7878a..946eeb57d188 100644 --- a/frontend/src/components/admin/codex/CodexBorrowActivity.vue +++ b/frontend/src/components/admin/codex/CodexBorrowActivity.vue @@ -35,7 +35,7 @@

{{ result.dispatched === false ? text('准备阶段结束，未发出对照请求', 'Stopped before dispatching this comparison') : result.applied ? text('实际已使用借用', 'Borrow was applied') : text('未使用借用', 'Borrow was not applied') }} · {{ result.reported_model || '—' }} · {{ result.duration_ms }} ms

{{ result.answer || '—' }}

{{ result.error }}
-
{{ text('完整响应', 'Complete response') }}
{{ result.raw_response }}
+
{{ text('完整响应', 'Complete response') }}

{{ text('流读取结束信息（完成终态优先）', 'Read cleanup detail (a completed terminal takes precedence)') }}: {{ result.read_error }}

{{ result.raw_response }}

{{ text('这些结果用于确认链路和本次回答，不证明模型的真实身份或智力水平。WS 仅在账号已经启用时可用。', 'These results establish transport behaviour and this answer, not physical model identity or intelligence. WS requires an already enabled account.') }}