fix(runtime): synchronize concurrent terminal state

internal/archtest/budgets.go

@@ -123,8 +123,8 @@ var LineBudgets = []LineBudget{
 	// aleg-cancellation-bleg-termination-hardening: single-use B-leg launch permit, concurrent bounded A-leg cancel fan-out, truthful physical CancelResult propagation, bounded attempt-owned sideband evidence accumulator, terminal stream drain, exactly-once terminal B-leg billing precedence, and bounded cancellation telemetry; measured 92771, bump to 92796 with 25 headroom.
 	// database-dialect-parity-enforcement: stable parity normalization across core components; measured 95020, bump to 95045 with 25 headroom.
 	// extension-plane-local-terminal: frozen identity accessor and turn-terminal carrier; measured 95070, bump to 95095 with 25 headroom.
-	// pre-oss-core-slimming: moved tool-call repair, secret-guard source/matcher, and concrete compaction detector out of core; measured 89845, ratchet downward to 89870 with 25 headroom.
-	{Dir: "internal/core", Max: 89870},
+	// pre-oss-core-slimming race remediation: measured 89936 after synchronizing attempt accounting, sideband teardown, and terminal provider identity; retain 25 lines of headroom.
+	{Dir: "internal/core", Max: 89961},
 	{Dir: "internal/pluginreg", Max: 1174},
 	{Dir: "internal/stdhttp", Max: 6693},
 	{Dir: "internal/infra/runtimebundle", Max: 12566},

internal/core/runtime/attempt_session.go

@@ -78,6 +78,8 @@ type attemptSession struct {
 	forceClose         chan struct{}
 	usageMu            sync.Mutex
 	billingMu          sync.Mutex
+	accountingMu       sync.Mutex
+	sidebandMu         sync.Mutex
 
 	internalUsageKeys  map[string]struct{}
 	accumulatedUsage   []lipapi.Event
@@ -130,6 +132,80 @@ func (a *attemptSession) claimBillingLegRecord() bool {
 	return true
 }
 
+func (a *attemptSession) accountingStartedAt() time.Time {
+	if a == nil {
+		return time.Time{}
+	}
+	a.accountingMu.Lock()
+	defer a.accountingMu.Unlock()
+	return a.accounting.requestStartedAt
+}
+
+func (a *attemptSession) accountingSnapshot() attemptAccountingSnapshot {
+	if a == nil {
+		return attemptAccountingSnapshot{}
+	}
+	a.accountingMu.Lock()
+	defer a.accountingMu.Unlock()
+	return a.accounting.snapshot()
+}
+
+func (a *attemptSession) observeAccountingBackendEvent(at time.Time, ev lipapi.Event) {
+	if a == nil {
+		return
+	}
+	a.accountingMu.Lock()
+	a.accounting.observeBackendEvent(at, ev)
+	a.accountingMu.Unlock()
+}
+
+func (a *attemptSession) observeAccountingClientEvent(at time.Time, ev lipapi.Event) {
+	if a == nil {
+		return
+	}
+	a.accountingMu.Lock()
+	a.accounting.observeClientEvent(at, ev)
+	a.accountingMu.Unlock()
+}
+
+func (a *attemptSession) observeAccountingUsage(ev lipapi.Event) {
+	if a == nil {
+		return
+	}
+	a.accountingMu.Lock()
+	a.accounting.observeUsage(ev)
+	a.accountingMu.Unlock()
+}
+
+func (a *attemptSession) toolCallAssembler() *toolCallAssembler {
+	if a == nil {
+		return nil
+	}
+	a.sidebandMu.Lock()
+	defer a.sidebandMu.Unlock()
+	return a.toolFinal
+}
+
+func (a *attemptSession) promptCacheSideband() (promptcache.ObservationSource, promptcache.Controller) {
+	if a == nil {
+		return nil, nil
+	}
+	a.sidebandMu.Lock()
+	defer a.sidebandMu.Unlock()
+	return a.promptCacheSource, a.promptCacheController
+}
+
+func (a *attemptSession) discardSidebandState() {
+	if a == nil {
+		return
+	}
+	a.sidebandMu.Lock()
+	a.toolFinal = nil
+	a.promptCacheSource = nil
+	a.promptCacheController = nil
+	a.sidebandMu.Unlock()
+}
+
 type attemptSessionInput struct {
 	inner          lipapi.ManagedEventStream
 	streamDisposed bool
@@ -303,7 +379,7 @@ func (a *attemptSession) terminalizeForCancel(ctx context.Context, cause lipapi.
 		BillingReason: detail,
 		TraceID:       a.traceID,
 		ALegID:        a.bleg.ALegID,
-		StartedAt:     a.accounting.requestStartedAt,
+		StartedAt:     a.accountingStartedAt(),
 	})
 }
 
@@ -340,7 +416,7 @@ func (a *attemptSession) terminalizeForClose(cmd sdkterminal.Command, relKind au
 		BillingReason: "aleg close",
 		TraceID:       a.traceID,
 		ALegID:        a.bleg.ALegID,
-		StartedAt:     a.accounting.requestStartedAt,
+		StartedAt:     a.accountingStartedAt(),
 	})
 }
 
@@ -1046,7 +1122,7 @@ func (r *readyAttempt) DisposeWithEvidence(ctx context.Context, intent attemptTe
 					evidence.ALegID = sess.bleg.ALegID
 				}
 				if evidence.StartedAt.IsZero() {
-					evidence.StartedAt = sess.accounting.requestStartedAt
+					evidence.StartedAt = sess.accountingStartedAt()
 				}
 				sess.TerminalizeAttempt(ctx, intent, evidence)
 			}
@@ -1076,7 +1152,7 @@ func (r *readyAttempt) DisposeWithEvidence(ctx context.Context, intent attemptTe
 			evidence.ALegID = sess.bleg.ALegID
 		}
 		if evidence.StartedAt.IsZero() {
-			evidence.StartedAt = sess.accounting.requestStartedAt
+			evidence.StartedAt = sess.accountingStartedAt()
 		}
 		sess.TerminalizeAttempt(ctx, intent, evidence)
 	}
@@ -1131,7 +1207,7 @@ func (r *readyAttempt) Dispose(ctx context.Context, err error) {
 					RecordReason: err.Error(),
 					TraceID:      sess.traceID,
 					ALegID:       sess.bleg.ALegID,
-					StartedAt:    sess.accounting.requestStartedAt,
+					StartedAt:    sess.accountingStartedAt(),
 				}
 				sess.TerminalizeAttempt(ctx, IntentPreReturnAbort, evidence)
 			}
@@ -1169,7 +1245,7 @@ func (r *readyAttempt) Dispose(ctx context.Context, err error) {
 			RecordReason: err.Error(),
 			TraceID:      sess.traceID,
 			ALegID:       sess.bleg.ALegID,
-			StartedAt:    sess.accounting.requestStartedAt,
+			StartedAt:    sess.accountingStartedAt(),
 		}
 		sess.TerminalizeAttempt(ctx, IntentPreReturnAbort, evidence)
 	}
@@ -1260,7 +1336,7 @@ func (a *attemptSession) makeSwallowedEvidence(facts recvTurnFacts, p *responseP
 		Snapshot:       snapshot,
 		RecordReason:   reason,
 		Err:            err,
-		StartedAt:      a.accounting.requestStartedAt,
+		StartedAt:      a.accountingStartedAt(),
 		StreamFallback: fallback,
 		BillingState:   tFacts.billingState,
 		BillingCallID:  tFacts.billingCallID,
@@ -1530,7 +1606,7 @@ func (a *attemptSession) TerminalizeAttempt(ctx context.Context, intent attemptT
 			runGuarded("append billing leg", &errorsList, func() {
 				started := evidence.StartedAt
 				if started.IsZero() {
-					started = a.accounting.requestStartedAt
+					started = a.accountingStartedAt()
 				}
 				var nowFn func() time.Time
 				if a.now != nil {
@@ -1677,11 +1753,9 @@ func (a *attemptSession) TerminalizeAttempt(ctx context.Context, intent attemptT
 			})
 		}
 
-		// 10. Discard attempt-local state.
-		a.accounting = attemptAccountingTracker{}
-		a.toolFinal = nil
-		a.promptCacheSource = nil
-		a.promptCacheController = nil
+		// Detach sideband state without mutating an assembler that a concurrent
+		// Recv may already have snapshotted.
+		a.discardSidebandState()
 
 		if evidence.Err != nil && intent == IntentSurfacedFailure {
 			errorsList = append(errorsList, evidence.Err)

internal/core/runtime/attempt_session_test.go

@@ -404,8 +404,6 @@ func testInstallSlot(slot *attemptSlot, s *attemptSession) {
 	}
 	slot.mu.Lock()
 	old := slot.current
-	slot.current = s
-	slot.mu.Unlock()
 	if s != nil && old != nil {
 		if s.billingEnabled == nil {
 			s.billingEnabled = old.billingEnabled
@@ -441,6 +439,8 @@ func testInstallSlot(slot *attemptSlot, s *attemptSession) {
 			s.aScope = old.aScope
 		}
 	}
+	slot.current = s
+	slot.mu.Unlock()
 }
 
 // testReadySession returns the underlying session for test assertions.

internal/core/runtime/attempt_usage_evidence.go

@@ -46,7 +46,7 @@ func (a *attemptSession) recordUsageEvidence(ev lipapi.Event) {
 	if !a.rememberUsageEvidenceOnce(ev) {
 		return
 	}
-	a.accounting.observeUsage(ev)
+	a.observeAccountingUsage(ev)
 }
 
 func (a *attemptSession) aggregatedUsageEvidence() lipapi.Event {
@@ -77,7 +77,7 @@ func (a *attemptSession) drainStreamUsageEvidence(inner lipapi.ManagedEventStrea
 				continue
 			}
 			if a.rememberUsageEvidenceOnce(ev) {
-				a.accounting.observeUsage(ev)
+				a.observeAccountingUsage(ev)
 			}
 		}
 		return nil

internal/core/runtime/executor_recv_loop.go

@@ -103,7 +103,7 @@ func (s *retryRecvStream) Recv(ctx context.Context) (lipapi.Event, error) {
 				}
 				terminal.finishResponseAtBoundary(p, attempt, false)
 			}
-			attempt.accounting.observeClientEvent(p.nowTime(), ev)
+			attempt.observeAccountingClientEvent(p.nowTime(), ev)
 			if recovery != nil && recovery.recoverPolicy != nil {
 				recovery.recoverPolicy.ObserveClientEvent(ev, p.nowTime())
 			}
@@ -129,7 +129,7 @@ func (s *retryRecvStream) Recv(ctx context.Context) (lipapi.Event, error) {
 			}
 			return out, false, nil
 		}
-		attempt.accounting.observeClientEvent(p.nowTime(), ev)
+		attempt.observeAccountingClientEvent(p.nowTime(), ev)
 		if recovery != nil && recovery.recoverPolicy != nil {
 			recovery.recoverPolicy.ObserveClientEvent(ev, p.nowTime())
 		}
@@ -413,8 +413,8 @@ func (s *retryRecvStream) Recv(ctx context.Context) (lipapi.Event, error) {
 		// receive or terminal decision; never carry the retired B-leg identity
 		// into the replacement.
 		attempt = slot.require()
-		if attempt.toolFinal != nil {
-			if ev, ok := attempt.toolFinal.popDrain(); ok {
+		if toolFinal := attempt.toolCallAssembler(); toolFinal != nil {
+			if ev, ok := toolFinal.popDrain(); ok {
 				out, cont, err := dispatchClientFacingEvent(ev, recvEventPreparation{event: ev})
 				if cont {
 					continue
@@ -465,7 +465,7 @@ func (s *retryRecvStream) Recv(ctx context.Context) (lipapi.Event, error) {
 				}
 				terminal.finishResponseAtBoundary(p, attempt, false)
 			}
-			attempt.accounting.observeClientEvent(p.nowTime(), ev)
+			attempt.observeAccountingClientEvent(p.nowTime(), ev)
 			pm, _ := facts.hookMeta(attempt.bleg, attempt.cand)
 			out, recording, emitErr := p.observeClientFacing(ctx, ev, responseEventInput{
 				facts: facts, attempt: attempt, recovery: recovery,

internal/core/runtime/executor_settlement.go

@@ -123,7 +123,7 @@ func (t *turnTerminal) finalizeBillingAfterCancel(ctx context.Context, attempt *
 	}
 	persistCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), billingFinalizeTimeout)
 	defer cancel()
-	attempt.accounting.observeUsage(ev)
+	attempt.observeAccountingUsage(ev)
 	p.rememberClientEvent(ev)
 	recording := p.recordClientFacingTerminal(persistCtx, request, attempt, ev, t.committed())
 	if recording.err != nil && p.log != nil {
@@ -238,7 +238,7 @@ func (t *turnTerminal) finalizeResponseFinishedAuthority(ctx context.Context, ev
 		ALegID:         request.aLegID,
 		Snapshot:       &snapshot,
 		RecordOutcome:  lipapi.AttemptSuccess,
-		StartedAt:      attempt.accounting.requestStartedAt,
+		StartedAt:      attempt.accountingStartedAt(),
 		StreamFallback: p.billingEvidenceFallback(),
 		BillingState:   request.billingState,
 		BillingCallID:  request.billingCallID,

internal/core/runtime/interleaved_stream.go

@@ -190,7 +190,7 @@ func (s *interleavedContinuationStream) recordVisibleOutput(ev lipapi.Event) {
 		return
 	}
 	s.thinker.terminal.markOutputCommittedForAttempt(ev, s.thinker.attempt.snapshot(), s.thinker.recovery)
-	s.thinker.attempt.require().accounting.observeClientEvent(s.thinker.responsePipeline.nowTime(), ev)
+	s.thinker.attempt.require().observeAccountingClientEvent(s.thinker.responsePipeline.nowTime(), ev)
 	if s.thinker.recovery != nil && s.thinker.recovery.recoverPolicy != nil {
 		s.thinker.recovery.recoverPolicy.ObserveClientEvent(ev, s.thinker.responsePipeline.nowTime())
 	}

internal/core/runtime/keepwarm_integration.go

@@ -45,10 +45,11 @@ func (p *responsePipeline) commitSuccessfulTurn(facts recvTurnFacts, attempt *at
 		return
 	}
 	p.keepwarmArmOnce.Do(func() {
-		if attempt.promptCacheSource == nil || attempt.promptCacheController == nil {
+		source, controller := attempt.promptCacheSideband()
+		if source == nil || controller == nil {
 			return
 		}
-		observations := attempt.promptCacheSource.DrainPromptCacheObservations()
+		observations := source.DrainPromptCacheObservations()
 		if len(observations) == 0 {
 			return
 		}
@@ -60,7 +61,7 @@ func (p *responsePipeline) commitSuccessfulTurn(facts recvTurnFacts, attempt *at
 			Observations:        observations,
 			BackendInstanceID:   attempt.cand.Primary.Backend,
 			CanonicalModelID:    attempt.cand.Primary.Model,
-			Controller:          attempt.promptCacheController,
+			Controller:          controller,
 		})
 	})
 }

internal/core/runtime/response_pipeline.go

@@ -251,8 +251,8 @@ func (p *responsePipeline) clearToolClassification() {
 }
 
 func clearAttemptToolState(p *responsePipeline, attempt *attemptSession) {
-	if attempt != nil && attempt.toolFinal != nil {
-		attempt.toolFinal.clear()
+	if attempt != nil {
+		attempt.discardSidebandState()
 	}
 	if p != nil {
 		p.clearToolClassification()

internal/core/runtime/response_pipeline_observations.go

@@ -63,18 +63,18 @@ func (p *responsePipeline) prepareRecvEvent(ctx context.Context, facts recvTurnF
 		return prepared
 	}
 	at := p.nowTime()
-	attempt.accounting.observeBackendEvent(at, ev)
+	attempt.observeAccountingBackendEvent(at, ev)
 	if ev.Kind == lipapi.EventUsageDelta && ev.Accounting.DedupeKey != "" && !attempt.rememberUsageEvidenceOnce(ev) {
 		prepared.swallowed = true
 		return prepared
 	}
-	attempt.accounting.observeUsage(ev)
+	attempt.observeAccountingUsage(ev)
 	prepared.partMeta, _ = facts.hookMeta(attempt.bleg, attempt.cand)
 	p.emitTraffic(ctx, attempt, sdktraffic.LegBTP, ev, prepared.partMeta)
 	p.emitUsage(ctx, facts, attempt, ev)
-	if attempt.toolFinal != nil && attempt.toolFinal.enabled() {
+	if toolFinal := attempt.toolCallAssembler(); toolFinal != nil && toolFinal.enabled() {
 		meta := toolcall.Meta{TraceID: facts.traceID, ALegID: facts.aLegID, BLegID: attempt.bleg.BLegID, AttemptSeq: attempt.bleg.Seq}
-		held, err := attempt.toolFinal.ingest(ctx, ev, meta)
+		held, err := toolFinal.ingest(ctx, ev, meta)
 		if err != nil {
 			p.clearToolClassification()
 			prepared.err = err
@@ -370,7 +370,7 @@ func (p *responsePipeline) consumeBackendUsageEvidenceForAttempt(ctx context.Con
 			continue
 		}
 		p.rememberInternalUsage(ev)
-		attempt.accounting.observeUsage(ev)
+		attempt.observeAccountingUsage(ev)
 		p.emitUsage(ctx, facts, attempt, ev)
 	}
 }

internal/core/runtime/secure_session_stream_record.go

@@ -127,7 +127,7 @@ func buildStreamEventRecordInputEvidence(evidence responseRequestEvidence, attem
 		}
 	}
 	if ev.Kind == lipapi.EventResponseFinished {
-		acct := attempt.accounting.snapshot()
+		acct := attempt.accountingSnapshot()
 		in.RequestStartedAt = acct.RequestStartedAt
 		in.FirstRemoteEventAt = acct.FirstRemoteEventAt
 		in.FirstMeaningfulTokenAt = acct.FirstMeaningfulTokenAt

internal/core/runtime/terminal_decision.go

@@ -61,7 +61,8 @@ func (t *turnTerminal) sharedTerminalDecision(ctx context.Context, provider term
 		}
 		t.terminalDecisionMu.Unlock()
 		cancel()
-		return evaluateTerminalDecisionWithFrozenIdentity(ctx, provider, t.terminalDecisionProviderID, t.terminalDecisionProviderHasID, input, t.log)
+		providerID, providerHasID := t.terminalDecisionProviderIdentity()
+		return evaluateTerminalDecisionWithFrozenIdentity(ctx, provider, providerID, providerHasID, input, t.log)
 	}
 	key := terminalDecisionKey{
 		cause:               input.Candidate.Cause,
@@ -116,7 +117,8 @@ func (t *turnTerminal) sharedTerminalDecision(ctx context.Context, provider term
 		t.terminalDecisionFlight = flight
 		t.terminalDecisionMu.Unlock()
 
-		outcome := evaluateTerminalDecisionWithFrozenIdentity(sharedCtx, provider, t.terminalDecisionProviderID, t.terminalDecisionProviderHasID, input, t.log)
+		providerID, providerHasID := t.terminalDecisionProviderIdentity()
+		outcome := evaluateTerminalDecisionWithFrozenIdentity(sharedCtx, provider, providerID, providerHasID, input, t.log)
 		cancel()
 		t.terminalDecisionMu.Lock()
 		flight.outcome = outcome

internal/core/runtime/terminal_decision_continuation.go

@@ -188,7 +188,7 @@ func continuationTransactionWithOverlay(ctx context.Context, t *turnTerminal, s
 		RecordReason:  continuationPendingReason,
 		TraceID:       b1.traceID,
 		ALegID:        b1.bleg.ALegID,
-		StartedAt:     b1.accounting.requestStartedAt,
+		StartedAt:     b1.accountingStartedAt(),
 	})
 	if !settled.Result.Won || settled.Result.Err != nil || !continuationSettlementSucceeded(b1) {
 		return true, errors.New("continuation: prior attempt settlement unavailable")