From 54754ef9c643618cb54b8ca482d1a8ff6ed41e0b Mon Sep 17 00:00:00 2001 From: tusharbhardwaj-bk Date: Thu, 20 Aug 2026 20:06:38 +0000 Subject: [PATCH 1/3] fix(server): stop the reaper cap from looping with durable recovery, and keep Codex readers alive Three threads in four days (907d4c74 on 08-17, ed3fc9f4 and mcp:1e019f68 on 08-20) sat in "Recovering" for 45 minutes and then held every later prompt in "Queued" until someone pressed Stop. The 5-minute cadence of their recovery attempts was the reaper's sweep interval. The reaper's 2h absolute cap fired on turn age, ignored that the agent had emitted an event seconds earlier, and only rewrote the projection (`thread.session.set interrupted`) - the provider turn kept running. Durable recovery then inspected the provider, found the same live turn, re-adopted it ("Agent work recovered"), and the next sweep settled it again. Ten successful recoveries later `recovery_attempts == maximum`; `observeSession` still parked the item in `recovering`, where nothing can claim it and nothing makes it terminal, so it blocked the thread's queue for good. The same sweep's inactivity pass then killed the live agent mid tool call, because the cap pass had just nulled the `activeTurnId` it relies on and `lastSeenAt` only moves on runtime operations. - ProviderSessionReaper: the cap measures silence (`lastActivityAt`), not age, and when it fires on a live session it dispatches `thread.turn.interrupt` first - the Stop path - so the work item is terminal and the provider is told to stop before the projection is settled. The inactivity pass asks the adapter (`inspectSession().activeProviderTurnId`) before terminating. - DurableExecutionIntentRepository.observeSession: an interrupted/stopped session exhausts an item whose budget is spent (`recovery-exhausted`, terminal) instead of parking it in `recovering`. Retry/Dismiss work, the next prompt runs. Separately, the second "continue" on mcp:1e019f68 dispatched fine and then the server went blind: Codex kept working (rollout grew for 30 minutes) while the provider event log stopped at sendTurn, until the inactivity reaper killed it. CodexAdapter forked its event reader with `Effect.forkChild`, i.e. as a child of whichever fiber called startSession. Since #124 runs each reactor command in a detached, short-lived fiber, a `thread.session-set` command whose inline `runDue` won the claim started the session and took the reader down with it when the command returned. - CodexAdapter / GrokAdapter: fork the reader into the session scope (`Effect.forkIn(sessionScope)`), as ClaudeAdapter already does. - ProviderCommandReactor: `session-set` and `session-restart-requested` wake the durable coordinator instead of dispatching inline, so turn starts only ever run on the coordinator's own long-lived fiber. Regression tests: reaper (silent turn is interrupted then settled; an old but busy turn is left alone; adapter-reported active turn is not terminated), repository (spent item exhausts terminally and unblocks the next prompt), CodexAdapter (reader survives the starting fiber ending). Co-Authored-By: Claude Fable 5 --- .../DurableExecutionIntentRepository.test.ts | 108 +++++++ .../DurableExecutionIntentRepository.ts | 33 ++- .../Layers/ProviderCommandReactor.ts | 9 +- .../orchestration/reconcileRunningTurns.ts | 66 +++++ .../src/provider/Layers/CodexAdapter.test.ts | 56 ++++ .../src/provider/Layers/CodexAdapter.ts | 39 ++- .../server/src/provider/Layers/GrokAdapter.ts | 3 +- .../Layers/ProviderSessionReaper.test.ts | 278 +++++++++++++++++- .../provider/Layers/ProviderSessionReaper.ts | 51 +++- 9 files changed, 616 insertions(+), 27 deletions(-) diff --git a/apps/server/src/execution/DurableExecutionIntentRepository.test.ts b/apps/server/src/execution/DurableExecutionIntentRepository.test.ts index 5eb53c1d369f..bc8a6db94fd8 100644 --- a/apps/server/src/execution/DurableExecutionIntentRepository.test.ts +++ b/apps/server/src/execution/DurableExecutionIntentRepository.test.ts @@ -463,4 +463,112 @@ layer("DurableExecutionIntentRepository", (it) => { ); }), ); + // T3-CUSTOM(expbkt3): an interrupted session must not park a spent work item + // in 'recovering' (unclaimable, non-terminal, blocks the thread's queue). + it.effect("exhausts a spent work item on session loss instead of parking it in recovering", () => + Effect.gen(function* () { + const repository = yield* DurableExecutionIntentRepository; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-spent-budget"); + const makeEvent = (suffix: string, sequence: number, occurredAt: string) => ({ + type: "thread.turn-start-requested" as const, + sequence, + eventId: EventId.make(`event-${suffix}`), + aggregateKind: "thread" as const, + aggregateId: threadId, + occurredAt, + commandId: CommandId.make(`command-${suffix}`), + causationEventId: null, + correlationId: CorrelationId.make(`command-${suffix}`), + metadata: {}, + payload: { + threadId, + messageId: MessageId.make(`message-${suffix}`), + runtimeMode: "full-access" as const, + interactionMode: "default" as const, + createdAt: occurredAt, + }, + }); + const accept = (suffix: string, sequence: number, occurredAt: string) => { + const event = makeEvent(suffix, sequence, occurredAt); + return repository.acceptFromEvent({ + event, + message: { + messageId: event.payload.messageId, + threadId, + turnId: null, + role: "user", + text: suffix, + attachments: [], + isStreaming: false, + sentByUserId: null, + createdAt: occurredAt, + updatedAt: occurredAt, + }, + }); + }; + + yield* accept("spent", 70, "2026-01-01T00:00:00.000Z"); + // Ten successful re-adoptions of a still-live provider turn leave the + // item running with its whole recovery budget consumed. + yield* sql` + UPDATE projection_thread_execution_intents + SET phase = 'running', delivery_certainty = 'provider-acknowledged', + provider_turn_id = 'provider-turn-spent', + recovery_attempts = maximum_recovery_attempts, + claim_owner = NULL, claim_expires_at = NULL + WHERE work_item_id = 'command-spent' + `; + + yield* repository.observeSession({ + threadId, + status: "interrupted", + providerTurnId: null, + error: "Interrupted: the turn produced no events for 120 minutes.", + at: "2026-01-01T02:00:00.000Z", + }); + + const spent = yield* repository.getByWorkItemId({ workItemId: "command-spent" }); + assert.isTrue(spent._tag === "Some"); + if (spent._tag === "None") return; + assert.strictEqual(spent.value.phase, "recovery-exhausted"); + assert.strictEqual(spent.value.desiredState, "stopped"); + assert.isFalse(spent.value.runnable); + assert.strictEqual(spent.value.terminalAt, "2026-01-01T02:00:00.000Z"); + assert.strictEqual(spent.value.exhaustedAt, "2026-01-01T02:00:00.000Z"); + + // The next prompt on the thread is not stuck behind it. + yield* accept("next", 71, "2026-01-01T02:05:00.000Z"); + const runnable = yield* repository.listRunnable({ + now: "2026-01-01T02:05:01.000Z", + limit: 10, + }); + assert.deepStrictEqual( + runnable.filter((item) => item.threadId === threadId).map((item) => item.workItemId), + ["command-next"], + ); + + // An item with budget left still goes through normal recovery. + yield* accept("fresh", 72, "2026-01-01T02:10:00.000Z"); + yield* sql` + UPDATE projection_thread_execution_intents + SET phase = 'running', delivery_certainty = 'provider-acknowledged', runnable = 1, + claim_owner = NULL, claim_expires_at = NULL + WHERE work_item_id = 'command-next' + `; + yield* repository.observeSession({ + threadId, + status: "interrupted", + providerTurnId: null, + error: "Session stopped", + at: "2026-01-01T02:11:00.000Z", + }); + const next = yield* repository.getByWorkItemId({ workItemId: "command-next" }); + assert.isTrue(next._tag === "Some"); + if (next._tag === "None") return; + assert.strictEqual(next.value.phase, "recovering"); + assert.strictEqual(next.value.desiredState, "running"); + assert.isNull(next.value.terminalAt); + }), + ); }); diff --git a/apps/server/src/execution/DurableExecutionIntentRepository.ts b/apps/server/src/execution/DurableExecutionIntentRepository.ts index 207f86ac535a..d7cdc984147d 100644 --- a/apps/server/src/execution/DurableExecutionIntentRepository.ts +++ b/apps/server/src/execution/DurableExecutionIntentRepository.ts @@ -1439,11 +1439,40 @@ const make = Effect.gen(function* () { input.status === "interrupted" || input.status === "stopped" ) { + // T3-CUSTOM(expbkt3): an item whose recovery budget is already spent + // cannot be claimed again (`recovery_attempts < maximum_recovery_attempts` + // gates every claim), so parking it in 'recovering' leaves a zombie + // that shows "Recovering" forever and head-of-line-blocks every later + // prompt on the thread. Exhaust it terminally instead; the user gets + // Retry/Dismiss and the next prompt runs. yield* sql` UPDATE projection_thread_execution_intents - SET phase = 'recovering', + SET phase = CASE + WHEN recovery_attempts >= maximum_recovery_attempts THEN 'recovery-exhausted' + ELSE 'recovering' + END, + desired_state = CASE + WHEN recovery_attempts >= maximum_recovery_attempts THEN 'stopped' + ELSE desired_state + END, + runnable = CASE + WHEN recovery_attempts >= maximum_recovery_attempts THEN 0 + ELSE runnable + END, delivery_certainty = CASE WHEN phase = 'starting' THEN 'uncertain' ELSE delivery_certainty END, - next_attempt_at = ${input.at}, last_failure_type = ${failureType}, + next_attempt_at = CASE + WHEN recovery_attempts >= maximum_recovery_attempts THEN NULL + ELSE ${input.at} + END, + exhausted_at = CASE + WHEN recovery_attempts >= maximum_recovery_attempts THEN ${input.at} + ELSE exhausted_at + END, + terminal_at = CASE + WHEN recovery_attempts >= maximum_recovery_attempts THEN ${input.at} + ELSE terminal_at + END, + last_failure_type = ${failureType}, last_failure_detail = ${input.error}, updated_at = ${input.at} WHERE work_item_id = ( SELECT work_item_id FROM projection_thread_execution_intents diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index ece7caf687b5..bfb7f630b448 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -2967,7 +2967,11 @@ const make = Effect.gen(function* () { } } yield* processSessionRestartRequested(event); - if (durableCoordinator !== null) yield* durableCoordinator.runDue; + // T3-CUSTOM(expbkt3): wake the coordinator instead of dispatching here. + // Dispatching inline runs the provider start under this bounded, + // short-lived command fiber; the coordinator's own fiber is the one + // place a turn start may run from. + if (durableCoordinator !== null) yield* durableCoordinator.wake(""); return; case "thread.archived": // T3-CUSTOM(expbkt3): archive fences the durable item transactionally @@ -2977,7 +2981,8 @@ const make = Effect.gen(function* () { .pipe(Effect.catchCause(Effect.logWarning), Effect.asVoid); return; case "thread.session-set": - if (durableCoordinator !== null) yield* durableCoordinator.runDue; + // T3-CUSTOM(expbkt3): see session-restart-requested above. + if (durableCoordinator !== null) yield* durableCoordinator.wake(""); return; } }); diff --git a/apps/server/src/orchestration/reconcileRunningTurns.ts b/apps/server/src/orchestration/reconcileRunningTurns.ts index 20042c49047c..6874bd30669d 100644 --- a/apps/server/src/orchestration/reconcileRunningTurns.ts +++ b/apps/server/src/orchestration/reconcileRunningTurns.ts @@ -22,12 +22,15 @@ */ import { CommandId, + EventId, IsoDateTime, ProviderInstanceId, type RuntimeMode, ThreadId, + TurnId, } from "@t3tools/contracts"; import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as SqlClient from "effect/unstable/sql/SqlClient"; @@ -42,6 +45,8 @@ export interface RunningSessionRow { readonly updatedAt: string; readonly lastActivityAt: string | null; readonly turnStartedAt: string | null; + /** T3-CUSTOM(expbkt3): the orchestration turn id the session projection is pinned to. */ + readonly activeTurnId: string | null; } /** @@ -58,6 +63,7 @@ export const listRunningSessionRows = Effect.gen(function* () { s.provider_instance_id AS "providerInstanceId", s.runtime_mode AS "runtimeMode", s.updated_at AS "updatedAt", + s.active_turn_id AS "activeTurnId", ( SELECT MAX(e.occurred_at) FROM orchestration_events e @@ -109,3 +115,63 @@ export const settleRunningSession = (input: { createdAt: settleAt, }); }); + +// T3-CUSTOM(expbkt3): BEGIN - interrupt a live turn for real before settling it. +/** + * Ask the provider to end a turn that is still alive in memory but has gone + * silent, through the same `thread.turn.interrupt` path the Stop button uses. + * + * `settleRunningSession` alone only rewrites the projection: the provider turn + * keeps running, durable recovery then sees that live turn, re-adopts it, and + * the reaper settles it again on its next sweep — ten "successful" recoveries + * later the work item is an unclaimable zombie that blocks every later prompt + * (2026-08-20, `mcp:1e019f68`). Dispatching the interrupt first makes the + * durable work item terminal (`stopThread`) and tells the provider to stop, so + * provider, projection and intent agree and there is nothing left to recover. + */ +export const interruptRunningSession = (input: { + readonly row: RunningSessionRow; + readonly reason: string; +}) => + Effect.gen(function* () { + const crypto = yield* Crypto.Crypto; + const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService; + const threadId = ThreadId.make(input.row.threadId); + const uuid = yield* crypto.randomUUIDv4; + const createdAt = IsoDateTime.make(DateTime.formatIso(yield* DateTime.now)); + + yield* orchestrationEngine.dispatch({ + type: "thread.turn.interrupt", + commandId: CommandId.make(`server:reconcile-running-turn-interrupt:${uuid}`), + threadId, + ...(input.row.activeTurnId !== null ? { turnId: TurnId.make(input.row.activeTurnId) } : {}), + createdAt, + }); + // Leave the reason in the thread feed; the interrupt itself carries none. + const activityUuid = yield* crypto.randomUUIDv4; + yield* orchestrationEngine.dispatch({ + type: "thread.activity.append", + commandId: CommandId.make(`server:reconcile-running-turn-activity:${activityUuid}`), + threadId, + activity: { + id: EventId.make(activityUuid), + tone: "info", + kind: "provider.turn.interrupted", + summary: input.reason, + payload: { + detail: input.reason, + activeTurnId: input.row.activeTurnId, + lastActivityAt: input.row.lastActivityAt, + }, + turnId: input.row.activeTurnId !== null ? TurnId.make(input.row.activeTurnId) : null, + createdAt, + }, + createdAt, + }); + yield* Effect.logInfo("provider.session.reaper.interrupted-silent-turn", { + threadId: input.row.threadId, + activeTurnId: input.row.activeTurnId, + reason: input.reason, + }); + }); +// T3-CUSTOM(expbkt3): END diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 3fbe0b9bf130..1ec426e39947 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -1499,3 +1499,59 @@ it.effect("flushes managed native logs when the adapter layer shuts down", () => } }), ); + +// T3-CUSTOM(expbkt3): the event reader belongs to the session scope, not to +// whichever fiber happened to call startSession. A short-lived caller (a bounded +// reactor command fiber) ending must not blind the server to a live runtime. +const readerRuntimeFactory = makeRuntimeFactory(); +const readerLayer = it.layer( + Layer.effect( + CodexAdapter, + Effect.gen(function* () { + const codexConfig = decodeCodexSettings({}); + return yield* makeCodexAdapter(codexConfig, { + makeRuntime: readerRuntimeFactory.factory, + }); + }), + ).pipe( + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), + Layer.provideMerge(ServerSettingsService.layerTest()), + Layer.provideMerge(providerSessionDirectoryTestLayer), + Layer.provideMerge(NodeServices.layer), + ), +); + +readerLayer("CodexAdapterLive event reader lifetime", (it) => { + it.effect("keeps delivering runtime events after the fiber that started the session ends", () => + Effect.gen(function* () { + const adapter = yield* CodexAdapter; + const starter = yield* adapter + .startSession({ + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-reader"), + runtimeMode: "full-access", + }) + .pipe(Effect.forkChild); + yield* Fiber.join(starter); + const runtime = readerRuntimeFactory.lastRuntime; + NodeAssert.ok(runtime); + + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + yield* runtime.emit({ + id: asEventId("evt-reader-closed"), + kind: "session", + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-reader"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "session/closed", + message: "Session stopped", + }); + const firstEvent = yield* Fiber.join(firstEventFiber).pipe(Effect.timeout("3 seconds")); + + NodeAssert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") return; + NodeAssert.equal(firstEvent.value.type, "session.exited"); + NodeAssert.equal(firstEvent.value.threadId, "thread-reader"); + }), + ); +}); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index d06394c6a3c0..9e8c1bbe8fba 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -1778,22 +1778,29 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( ), ); - const eventFiber = yield* Stream.runForEach(runtime.events, (event) => - Effect.gen(function* () { - yield* writeNativeEvent(event); - const runtimeEvents = mapToRuntimeEvents(event, event.threadId); - if (runtimeEvents.length === 0) { - yield* Effect.logDebug("ignoring unhandled Codex provider event", { - method: event.method, - threadId: event.threadId, - turnId: event.turnId, - itemId: event.itemId, - }); - return; - } - yield* Queue.offerAll(runtimeEventQueue, runtimeEvents); - }), - ).pipe(Effect.forkChild); + const eventFiber = yield* Stream.runForEach( + runtime.events, + (event) => + Effect.gen(function* () { + yield* writeNativeEvent(event); + const runtimeEvents = mapToRuntimeEvents(event, event.threadId); + if (runtimeEvents.length === 0) { + yield* Effect.logDebug("ignoring unhandled Codex provider event", { + method: event.method, + threadId: event.threadId, + turnId: event.turnId, + itemId: event.itemId, + }); + return; + } + yield* Queue.offerAll(runtimeEventQueue, runtimeEvents); + }), + // T3-CUSTOM(expbkt3): the reader lives as long as the session, never + // as long as the caller. `forkChild` tied it to whichever fiber ran + // startSession; when that was a short-lived reactor command fiber the + // reader died the moment sendTurn returned and the server went blind + // to a Codex process that kept working (2026-08-20, `mcp:1e019f68`). + ).pipe(Effect.forkIn(sessionScope)); const started = yield* runtime.start().pipe( Effect.mapError( diff --git a/apps/server/src/provider/Layers/GrokAdapter.ts b/apps/server/src/provider/Layers/GrokAdapter.ts index 647b4a8678f2..4405260848c5 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.ts @@ -930,7 +930,8 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte Effect.catch((cause) => Effect.logError("Failed to process Grok runtime notification.", { cause }), ), - Effect.forkChild, + // T3-CUSTOM(expbkt3): session-scoped, not caller-scoped (see CodexAdapter). + Effect.forkIn(sessionScope), ); ctx.notificationFiber = nf; diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 384b5e96fa3e..d02dd095db6b 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -16,6 +16,7 @@ import * as ManagedRuntime from "effect/ManagedRuntime"; import * as Option from "effect/Option"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; import { afterEach, describe, expect, it, vi } from "vite-plus/test"; import { OrchestrationEngineService } from "../../orchestration/Services/OrchestrationEngine.ts"; @@ -211,6 +212,11 @@ describe("ProviderSessionReaper", () => { readonly configEnv?: Record; readonly inactivityThresholdMs?: number; readonly sweepIntervalMs?: number; + // T3-CUSTOM(expbkt3): orphaned-turn pass doubles. + readonly turnAbsoluteCapMs?: number; + readonly listSessions?: ProviderServiceShape["listSessions"]; + readonly inspectSession?: ProviderServiceShape["inspectSession"]; + readonly dispatch?: OrchestrationEngineService["Service"]["dispatch"]; }) { const terminatedThreadIds = new Set(); const terminateSession = vi.fn( @@ -227,13 +233,13 @@ describe("ProviderSessionReaper", () => { startSession: () => unsupported(), sendTurn: () => unsupported(), interruptTurn: () => unsupported(), - inspectSession: () => Effect.succeed(null), + inspectSession: input.inspectSession ?? (() => Effect.succeed(null)), requestTurnInterrupt: () => unsupported(), terminateSession, respondToRequest: () => unsupported(), respondToUserInput: () => unsupported(), stopSession: () => unsupported(), - listSessions: () => Effect.succeed([]), + listSessions: input.listSessions ?? (() => Effect.succeed([])), // T3-CUSTOM(expbkt3): explicit durable execution behavior. getCapabilities: () => Effect.succeed({ @@ -267,6 +273,9 @@ describe("ProviderSessionReaper", () => { const layer = makeProviderSessionReaperLive({ inactivityThresholdMs: input.inactivityThresholdMs ?? 1_000, sweepIntervalMs: input.sweepIntervalMs ?? 60_000, + ...(input.turnAbsoluteCapMs !== undefined + ? { turnAbsoluteCapMs: input.turnAbsoluteCapMs } + : {}), }).pipe( Layer.provideMerge(providerSessionDirectoryLayer), Layer.provideMerge(runtimeRepositoryLayer), @@ -307,7 +316,7 @@ describe("ProviderSessionReaper", () => { Layer.provideMerge( Layer.succeed(OrchestrationEngineService, { readEvents: () => Stream.empty, - dispatch: () => Effect.die("unused"), + dispatch: input.dispatch ?? (() => Effect.die("unused")), streamDomainEvents: Stream.empty, latestSequence: Effect.succeed(0), }), @@ -800,4 +809,267 @@ describe("ProviderSessionReaper", () => { [{ threadId }], ]); }); + // T3-CUSTOM(expbkt3): BEGIN - orphaned-turn pass: silence, real interrupts, + // and no termination of a turn the adapter still reports as running. + const sqlQuote = (value: string | number | null) => + value === null + ? "NULL" + : typeof value === "number" + ? String(value) + : `'${value.replace(/'/g, "''")}'`; + + async function seedRow(table: string, values: Record) { + // The harness layer merges SqlitePersistenceMemory in, but the runtime's + // declared context is narrowed to the reaper services. + const sqlRuntime = runtime as unknown as ManagedRuntime.ManagedRuntime< + SqlClient.SqlClient, + never + >; + await sqlRuntime.runPromise( + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const columns = yield* sql.unsafe<{ + readonly name: string; + readonly notnull: number; + readonly dflt_value: string | null; + readonly pk: number; + }>(`PRAGMA table_info(${table})`); + const row: Record = { ...values }; + for (const column of columns) { + if (column.notnull === 1 && column.dflt_value === null && !(column.name in row)) { + row[column.name] = column.name.endsWith("_json") + ? "{}" + : column.name.endsWith("_at") + ? "2026-01-01T00:00:00.000Z" + : ""; + } + } + const names = Object.keys(row).join(", "); + const placeholders = Object.values(row).map(sqlQuote).join(", "); + yield* sql.unsafe(`INSERT INTO ${table} (${names}) VALUES (${placeholders})`); + }), + ); + } + + async function seedRunningTurn(input: { + readonly threadId: ThreadId; + readonly turnId: TurnId; + readonly startedAt: string; + readonly lastEventAt: string | null; + }) { + await seedRow("projection_threads", { + thread_id: input.threadId, + project_id: "project-provider-session-reaper", + title: "silent turn", + created_at: input.startedAt, + updated_at: input.startedAt, + }); + await seedRow("projection_thread_sessions", { + thread_id: input.threadId, + status: "running", + provider_name: "codex", + provider_instance_id: "codex", + active_turn_id: input.turnId, + updated_at: input.startedAt, + }); + await seedRow("projection_turns", { + thread_id: input.threadId, + turn_id: input.turnId, + state: "running", + requested_at: input.startedAt, + started_at: input.startedAt, + checkpoint_files_json: "[]", + }); + if (input.lastEventAt !== null) { + await seedRow("orchestration_events", { + event_id: `event-${input.threadId}`, + aggregate_kind: "thread", + stream_id: input.threadId, + stream_version: 1, + event_type: "thread.activity-appended", + occurred_at: input.lastEventAt, + actor_kind: "provider", + payload_json: "{}", + metadata_json: "{}", + }); + } + } + + function liveCodexSession(threadId: ThreadId, at: string) { + return () => + Effect.succeed([ + { + provider: ProviderDriverKind.make("codex"), + providerInstanceId: ProviderInstanceId.make("codex"), + status: "running" as const, + runtimeMode: "full-access" as const, + threadId, + createdAt: at, + updatedAt: at, + }, + ]); + } + + it("interrupts for real, then settles, a live turn that has been silent past the cap", async () => { + const threadId = ThreadId.make("thread-reaper-silent"); + const turnId = TurnId.make("turn-reaper-silent"); + const nowMs = await Effect.runPromise(Clock.currentTimeMillis); + const threeHoursAgo = DateTime.formatIso(DateTime.makeUnsafe(nowMs - 3 * 60 * 60 * 1000)); + const nowIso = DateTime.formatIso(DateTime.makeUnsafe(nowMs)); + const dispatched: Array<{ readonly type: string }> = []; + const harness = await createHarness({ + inactivityThresholdMs: 24 * 60 * 60 * 1000, + readModel: makeReadModel([ + { + id: threadId, + session: { + threadId, + status: "running", + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: turnId, + lastError: null, + updatedAt: threeHoursAgo, + }, + }, + ]), + listSessions: liveCodexSession(threadId, threeHoursAgo), + dispatch: (command) => { + dispatched.push(command); + return Effect.succeed({ sequence: dispatched.length }); + }, + }); + await seedRunningTurn({ + threadId, + turnId, + startedAt: threeHoursAgo, + lastEventAt: threeHoursAgo, + }); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); + await runtime!.runPromise( + repository.upsert({ + threadId, + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + adapterKey: "codex", + runtimeMode: "full-access", + status: "running", + lastSeenAt: nowIso, + resumeCursor: null, + runtimePayload: null, + }), + ); + + await startReaper(); + await waitFor(() => dispatched.length >= 3); + + expect(dispatched.map((command) => command.type)).toEqual([ + "thread.turn.interrupt", + "thread.activity.append", + "thread.session.set", + ]); + const interrupt = dispatched[0] as { readonly turnId?: string }; + expect(interrupt.turnId).toBe(turnId); + const settle = dispatched[2] as unknown as { readonly session: { readonly status: string } }; + expect(settle.session.status).toBe("interrupted"); + expect(harness.terminateSession).not.toHaveBeenCalled(); + }); + + it("leaves a live turn alone while it is still emitting events, however old it is", async () => { + const threadId = ThreadId.make("thread-reaper-busy"); + const turnId = TurnId.make("turn-reaper-busy"); + const nowMs = await Effect.runPromise(Clock.currentTimeMillis); + const threeHoursAgo = DateTime.formatIso(DateTime.makeUnsafe(nowMs - 3 * 60 * 60 * 1000)); + const oneMinuteAgo = DateTime.formatIso(DateTime.makeUnsafe(nowMs - 60 * 1000)); + const dispatched: Array<{ readonly type: string }> = []; + const harness = await createHarness({ + inactivityThresholdMs: 24 * 60 * 60 * 1000, + readModel: makeReadModel([ + { + id: threadId, + session: { + threadId, + status: "running", + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: turnId, + lastError: null, + updatedAt: threeHoursAgo, + }, + }, + ]), + listSessions: liveCodexSession(threadId, threeHoursAgo), + dispatch: (command) => { + dispatched.push(command); + return Effect.succeed({ sequence: dispatched.length }); + }, + }); + await seedRunningTurn({ + threadId, + turnId, + startedAt: threeHoursAgo, + lastEventAt: oneMinuteAgo, + }); + + await startReaper(); + await Effect.runPromise(Effect.sleep("200 millis")); + + expect(dispatched).toEqual([]); + expect(harness.terminateSession).not.toHaveBeenCalled(); + }); + + it("does not terminate an idle-looking session while the adapter reports an active turn", async () => { + const threadId = ThreadId.make("thread-reaper-adapter-turn"); + const now = "2026-01-01T00:00:00.000Z"; + const harness = await createHarness({ + configEnv: { T3CODE_PROVIDER_SESSION_INACTIVITY_MS: "0" }, + inactivityThresholdMs: 600_000, + readModel: makeReadModel([ + { + id: threadId, + session: { + threadId, + status: "ready", + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + }, + ]), + inspectSession: () => + Effect.succeed({ + threadId, + generation: 1, + state: "running" as const, + activeProviderTurnId: TurnId.make("turn-still-running"), + runtimeAlive: true, + }), + }); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); + await runtime!.runPromise( + repository.upsert({ + threadId, + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + adapterKey: "codex", + runtimeMode: "full-access", + status: "running", + lastSeenAt: "2026-04-14T00:00:00.000Z", + resumeCursor: null, + runtimePayload: null, + }), + ); + + await startReaper(); + await Effect.runPromise(Effect.sleep("200 millis")); + + expect(harness.terminateSession).not.toHaveBeenCalled(); + }); + // T3-CUSTOM(expbkt3): END }); diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.ts index 9017d667aaf5..83f126167498 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.ts @@ -12,6 +12,7 @@ import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as OrchestrationEngine from "../../orchestration/Services/OrchestrationEngine.ts"; import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; import { + interruptRunningSession, listRunningSessionRows, settleRunningSession, } from "../../orchestration/reconcileRunningTurns.ts"; @@ -67,6 +68,11 @@ const DEFAULT_SWEEP_INTERVAL_MS = 5 * 60 * 1000; * session but has clearly gone nowhere. Deliberately generous: silence is not * evidence of death (see reconcileRunningTurns), so this must never be the * mechanism that ends a normal turn. + * + * T3-CUSTOM(expbkt3): measured against the thread's last recorded event, not + * the turn start. Agents here legitimately run multi-hour monitoring turns that + * emit steadily; a turn that is still producing events has not "gone nowhere", + * and capping it by age alone started the settle/recover loop of 2026-08-20. */ const DEFAULT_TURN_ABSOLUTE_CAP_MS = 2 * 60 * 60 * 1000; /** @@ -144,24 +150,44 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = for (const row of rows) { const startedMs = Date.parse(row.turnStartedAt ?? row.updatedAt); const runningForMs = Number.isNaN(startedMs) ? 0 : now - startedMs; + // T3-CUSTOM(expbkt3): silence, not age, is what the cap measures. + const lastActivityMs = Date.parse(row.lastActivityAt ?? row.turnStartedAt ?? row.updatedAt); + const silentForMs = Number.isNaN(lastActivityMs) ? runningForMs : now - lastActivityMs; + const sessionLive = liveThreadIds.has(row.threadId); - const reason = !liveThreadIds.has(row.threadId) + const reason = !sessionLive ? runningForMs >= ORPHAN_EVIDENCE_GRACE_MS ? "Interrupted: the agent session is no longer running." : null - : runningForMs >= turnAbsoluteCapMs - ? "Interrupted: the turn exceeded the maximum run time." + : silentForMs >= turnAbsoluteCapMs + ? `Interrupted: the turn produced no events for ${Math.round(turnAbsoluteCapMs / 60_000)} minutes.` : null; if (reason === null) { continue; } + // T3-CUSTOM(expbkt3): a live-but-silent turn is interrupted for real + // first, so the durable work item is terminal and the provider is told + // to stop before the projection is settled. Settling alone re-armed + // durable recovery against the still-running provider turn. + if (sessionLive) { + yield* interruptRunningSession({ row, reason }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("provider.session.reaper.interrupt-silent-turn-failed", { + threadId: row.threadId, + cause, + }), + ), + ); + } + yield* settleRunningSession({ row, reason }).pipe( Effect.tap(() => Effect.logInfo("provider.session.reaper.settled-orphaned-turn", { threadId: row.threadId, runningForMs, + silentForMs, reason, }), ), @@ -308,6 +334,25 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = continue; } + // T3-CUSTOM(expbkt3): BEGIN - the projection's activeTurnId is not + // proof of idleness: the orphaned-turn pass above clears it in this + // same sweep, and lastSeenAt only moves on runtime operations, never on + // streamed events. Ask the adapter whether a turn is actually running + // before terminating a process that may be mid tool call (2026-08-20 + // 17:33:50 and 19:29:17, `mcp:1e019f68`). + const inspection = yield* providerService + .inspectSession(binding.threadId) + .pipe(Effect.catchCause(() => Effect.succeed(null))); + if (inspection?.activeProviderTurnId != null) { + yield* Effect.logDebug("provider.session.reaper.skipped-live-provider-turn", { + threadId: binding.threadId, + activeProviderTurnId: inspection.activeProviderTurnId, + idleDurationMs, + }); + continue; + } + // T3-CUSTOM(expbkt3): END + // The turn can settle while background work runs on (subagent // fleets, workflow runs, Monitor watch loops). Those live inside the // provider process, so stopping the session would kill them silently, From 38075d3b49c6310f31d9c29085e400798855c115 Mon Sep 17 00:00:00 2001 From: tusharbhardwaj-bk Date: Thu, 20 Aug 2026 20:12:06 +0000 Subject: [PATCH 2/3] fix(reaper): protect a held turn only while its thread still emits events A turn the adapter still holds but whose thread recorded no event inside the inactivity threshold is one the server has gone blind to; terminating it is right. One that is still emitting is alive, whatever the projection says. Co-Authored-By: Claude Fable 5 --- .../orchestration/reconcileRunningTurns.ts | 18 +++++ .../Layers/ProviderSessionReaper.test.ts | 75 ++++++++++++++++++- .../provider/Layers/ProviderSessionReaper.ts | 33 +++++--- 3 files changed, 114 insertions(+), 12 deletions(-) diff --git a/apps/server/src/orchestration/reconcileRunningTurns.ts b/apps/server/src/orchestration/reconcileRunningTurns.ts index 6874bd30669d..edad598d0f2b 100644 --- a/apps/server/src/orchestration/reconcileRunningTurns.ts +++ b/apps/server/src/orchestration/reconcileRunningTurns.ts @@ -175,3 +175,21 @@ export const interruptRunningSession = (input: { }); }); // T3-CUSTOM(expbkt3): END + +// T3-CUSTOM(expbkt3): BEGIN - event-based liveness for the inactivity pass. +/** + * When the thread last recorded any event. `lastSeenAt` on the provider + * binding only moves on runtime operations (start, sendTurn), so it cannot + * tell a streaming agent from a dead one; the event stream can. + */ +export const latestThreadEventAt = (threadId: string) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const rows = yield* sql<{ readonly lastActivityAt: string | null }>` + SELECT MAX(occurred_at) AS "lastActivityAt" + FROM orchestration_events + WHERE stream_id = ${threadId} + `; + return rows[0]?.lastActivityAt ?? null; + }); +// T3-CUSTOM(expbkt3): END diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index d02dd095db6b..f1b724bb774d 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -1020,7 +1020,21 @@ describe("ProviderSessionReaper", () => { expect(harness.terminateSession).not.toHaveBeenCalled(); }); - it("does not terminate an idle-looking session while the adapter reports an active turn", async () => { + async function seedThreadEvent(threadId: ThreadId, occurredAt: string) { + await seedRow("orchestration_events", { + event_id: `event-${threadId}-${occurredAt}`, + aggregate_kind: "thread", + stream_id: threadId, + stream_version: 1, + event_type: "thread.activity-appended", + occurred_at: occurredAt, + actor_kind: "provider", + payload_json: "{}", + metadata_json: "{}", + }); + } + + it("does not terminate an idle-looking session while the adapter reports a turn that is still emitting", async () => { const threadId = ThreadId.make("thread-reaper-adapter-turn"); const now = "2026-01-01T00:00:00.000Z"; const harness = await createHarness({ @@ -1065,11 +1079,70 @@ describe("ProviderSessionReaper", () => { runtimePayload: null, }), ); + const nowMs = await Effect.runPromise(Clock.currentTimeMillis); + await seedThreadEvent(threadId, DateTime.formatIso(DateTime.makeUnsafe(nowMs - 30_000))); await startReaper(); await Effect.runPromise(Effect.sleep("200 millis")); expect(harness.terminateSession).not.toHaveBeenCalled(); }); + + it("still terminates a session whose adapter holds a turn but that stopped producing events", async () => { + const threadId = ThreadId.make("thread-reaper-blind-turn"); + const now = "2026-01-01T00:00:00.000Z"; + const harness = await createHarness({ + configEnv: { T3CODE_PROVIDER_SESSION_INACTIVITY_MS: "0" }, + inactivityThresholdMs: 600_000, + readModel: makeReadModel([ + { + id: threadId, + session: { + threadId, + status: "ready", + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + }, + ]), + inspectSession: () => + Effect.succeed({ + threadId, + generation: 1, + state: "running" as const, + activeProviderTurnId: TurnId.make("turn-nobody-hears"), + runtimeAlive: true, + }), + }); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); + await runtime!.runPromise( + repository.upsert({ + threadId, + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + adapterKey: "codex", + runtimeMode: "full-access", + status: "running", + lastSeenAt: "2026-04-14T00:00:00.000Z", + resumeCursor: null, + runtimePayload: null, + }), + ); + const nowMs = await Effect.runPromise(Clock.currentTimeMillis); + await seedThreadEvent( + threadId, + DateTime.formatIso(DateTime.makeUnsafe(nowMs - 2 * 60 * 60 * 1000)), + ); + + await startReaper(); + await waitFor(() => harness.terminateSession.mock.calls.length === 1); + + expect(harness.terminatedThreadIds.has(threadId)).toBe(true); + }); // T3-CUSTOM(expbkt3): END }); diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.ts index 83f126167498..8bd8a03fe9e6 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.ts @@ -13,6 +13,7 @@ import * as OrchestrationEngine from "../../orchestration/Services/Orchestration import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; import { interruptRunningSession, + latestThreadEventAt, listRunningSessionRows, settleRunningSession, } from "../../orchestration/reconcileRunningTurns.ts"; @@ -335,21 +336,31 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = } // T3-CUSTOM(expbkt3): BEGIN - the projection's activeTurnId is not - // proof of idleness: the orphaned-turn pass above clears it in this - // same sweep, and lastSeenAt only moves on runtime operations, never on - // streamed events. Ask the adapter whether a turn is actually running - // before terminating a process that may be mid tool call (2026-08-20 - // 17:33:50 and 19:29:17, `mcp:1e019f68`). + // proof of idleness: it can be cleared (by the orphaned-turn pass, or a + // turn.started lost to the generation fence) while the agent streams, + // and lastSeenAt only moves on runtime operations. If the adapter still + // holds a turn AND the thread recorded an event inside the threshold, + // the agent is alive - do not terminate it mid tool call (2026-08-20 + // 17:33:50, `mcp:1e019f68`). A held turn with no events past the + // threshold is a session we have gone blind to; that one is reaped. const inspection = yield* providerService .inspectSession(binding.threadId) .pipe(Effect.catchCause(() => Effect.succeed(null))); if (inspection?.activeProviderTurnId != null) { - yield* Effect.logDebug("provider.session.reaper.skipped-live-provider-turn", { - threadId: binding.threadId, - activeProviderTurnId: inspection.activeProviderTurnId, - idleDurationMs, - }); - continue; + const lastEventAt = yield* latestThreadEventAt(binding.threadId).pipe( + Effect.provide(reconcileContext), + Effect.catchCause(() => Effect.succeed(null)), + ); + const lastEventMs = lastEventAt === null ? Number.NaN : Date.parse(lastEventAt); + if (!Number.isNaN(lastEventMs) && now - lastEventMs < inactivityThresholdMs) { + yield* Effect.logDebug("provider.session.reaper.skipped-live-provider-turn", { + threadId: binding.threadId, + activeProviderTurnId: inspection.activeProviderTurnId, + idleDurationMs, + lastEventAt, + }); + continue; + } } // T3-CUSTOM(expbkt3): END From 3b4637e99f7b2d88a7829f606f9184d66009fea4 Mon Sep 17 00:00:00 2001 From: tusharbhardwaj-bk Date: Fri, 21 Aug 2026 07:28:24 +0000 Subject: [PATCH 3/3] fix(server): use T3 sender for Claude userEmail Claude's native prompt reports the shared account email, which misattributes the user on Beknown's shared runtime. Append the session-scoped message sender as the authoritative userEmail, and explicitly preserve unknown identity when no sender is resolved. TEC-1031 --- .../src/provider/Layers/ClaudeAdapter.ts | 13 ++++++- .../claudeSessionIdentity.expbkt3.test.ts | 39 +++++++++++++++++++ .../provider/claudeSessionIdentity.expbkt3.ts | 34 ++++++++++++++++ docs/operations/expbkt3-customizations.md | 8 ++++ 4 files changed, 93 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/provider/claudeSessionIdentity.expbkt3.test.ts create mode 100644 apps/server/src/provider/claudeSessionIdentity.expbkt3.ts diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 9843b52f5399..0f26b54099ee 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -102,6 +102,8 @@ import { import { makeObservableLifecycle } from "../observableLifecycle.ts"; import { type ClaudeAdapterShape } from "../Services/ClaudeAdapter.ts"; import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; +// T3-CUSTOM(expbkt3): Claude's shared account email is not the T3 message sender. +import { claudeSessionIdentitySystemPrompt } from "../claudeSessionIdentity.expbkt3.ts"; const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.fromJsonString(Schema.Unknown)); const decodeUnknownJsonStringExit = Schema.decodeUnknownExit(Schema.fromJsonString(Schema.Unknown)); @@ -4193,11 +4195,20 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ...(input.cwd ? [input.cwd] : []), serverConfig.attachmentsDir, ]; + // T3-CUSTOM(expbkt3): BEGIN override Claude's shared-account userEmail context. + const sessionIdentitySystemPrompt = claudeSessionIdentitySystemPrompt(sessionEnvironment); + // T3-CUSTOM(expbkt3): END const queryOptions: ClaudeQueryOptions = { ...(input.cwd ? { cwd: input.cwd } : {}), ...(apiModelId ? { model: apiModelId } : {}), pathToClaudeCodeExecutable: claudeBinaryPath, - systemPrompt: { type: "preset", preset: "claude_code" }, + // T3-CUSTOM(expbkt3): BEGIN preserve the native prompt with T3 sender identity appended. + systemPrompt: { + type: "preset", + preset: "claude_code", + ...(sessionIdentitySystemPrompt ? { append: sessionIdentitySystemPrompt } : {}), + }, + // T3-CUSTOM(expbkt3): END settingSources: [...CLAUDE_SETTING_SOURCES], // `ultracode` is a Claude Code setting, not an API effort level. It is // normalized to `xhigh` above and paired with `settings.ultracode`. diff --git a/apps/server/src/provider/claudeSessionIdentity.expbkt3.test.ts b/apps/server/src/provider/claudeSessionIdentity.expbkt3.test.ts new file mode 100644 index 000000000000..86027153c011 --- /dev/null +++ b/apps/server/src/provider/claudeSessionIdentity.expbkt3.test.ts @@ -0,0 +1,39 @@ +import { assert, describe, it } from "@effect/vitest"; + +import { claudeSessionIdentitySystemPrompt } from "./claudeSessionIdentity.expbkt3.ts"; + +describe("claudeSessionIdentitySystemPrompt", () => { + it("uses the current T3 message sender instead of the shared Claude account", () => { + assert.equal( + claudeSessionIdentitySystemPrompt({ + BK_IDENTITY_RUNTIME: "t3-code", + BK_SESSION_OWNER_EMAIL: "owner@example.test", + BK_MESSAGE_SENDER_EMAIL: " sender@example.test ", + }), + [ + "T3 Code session identity:", + '- userEmail is "sender@example.test".', + "- This session-scoped value identifies the user who sent the current message and overrides the Claude account email for user attribution.", + ].join("\n"), + ); + }); + + it("keeps the user unknown when T3 cannot resolve the message sender", () => { + assert.include( + claudeSessionIdentitySystemPrompt({ + BK_IDENTITY_RUNTIME: "t3-code", + BK_SESSION_OWNER_EMAIL: "owner@example.test", + }) ?? "", + "userEmail is unavailable", + ); + }); + + it("does not change upstream Claude sessions", () => { + assert.equal( + claudeSessionIdentitySystemPrompt({ + BK_MESSAGE_SENDER_EMAIL: "sender@example.test", + }), + undefined, + ); + }); +}); diff --git a/apps/server/src/provider/claudeSessionIdentity.expbkt3.ts b/apps/server/src/provider/claudeSessionIdentity.expbkt3.ts new file mode 100644 index 000000000000..0dccfd740a66 --- /dev/null +++ b/apps/server/src/provider/claudeSessionIdentity.expbkt3.ts @@ -0,0 +1,34 @@ +// T3-CUSTOM(expbkt3): Claude Code derives its native `userEmail` context from +// the authenticated Claude account. Beknown runs that account on a shared +// machine, so it identifies the subscription rather than the person sending +// the current T3 message. Append the session-scoped identity to Claude's native +// system prompt without changing the account used for authentication. + +import { + MESSAGE_SENDER_EMAIL_KEY, + SESSION_IDENTITY_RUNTIME, + SESSION_IDENTITY_RUNTIME_KEY, +} from "../identity/SessionIdentityEnvironment.ts"; + +export function claudeSessionIdentitySystemPrompt( + environment: NodeJS.ProcessEnv, +): string | undefined { + if (environment[SESSION_IDENTITY_RUNTIME_KEY] !== SESSION_IDENTITY_RUNTIME) { + return undefined; + } + + const senderEmail = environment[MESSAGE_SENDER_EMAIL_KEY]?.trim(); + if (!senderEmail) { + return [ + "T3 Code session identity:", + "- userEmail is unavailable for the user who sent the current message.", + "- Do not use the Claude account email, operating-system identity, or Git identity to infer the user.", + ].join("\n"); + } + + return [ + "T3 Code session identity:", + `- userEmail is ${JSON.stringify(senderEmail)}.`, + "- This session-scoped value identifies the user who sent the current message and overrides the Claude account email for user attribution.", + ].join("\n"); +} diff --git a/docs/operations/expbkt3-customizations.md b/docs/operations/expbkt3-customizations.md index c6f206c2a584..a1e2bc96242a 100644 --- a/docs/operations/expbkt3-customizations.md +++ b/docs/operations/expbkt3-customizations.md @@ -395,6 +395,14 @@ Four rules carry the behaviour: live session was started with and restarts on an owner transfer or a new sender, next to the existing credential-actor restart. +Claude Code's native system prompt normally derives `userEmail` from the +authenticated Claude account. That account is shared in the Beknown runtime, so +the Claude adapter appends the resolved `BK_MESSAGE_SENDER_EMAIL` as the +authoritative `userEmail`. When the sender is unresolved, the appended context +explicitly leaves `userEmail` unknown and forbids inference from the shared +Claude account, operating-system identity, or Git identity. Non-T3 Claude +sessions keep the upstream system prompt unchanged. + The markers compose with source-control profiles rather than replacing them: `mergeSourceControlEnvironment` scrubs the machine's inherited Git and GitHub credentials only when the overlay carries a source-control identity of its own,