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..edad598d0f2b 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,81 @@ 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 + +// 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/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/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..f1b724bb774d 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,340 @@ 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(); + }); + + 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({ + 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, + }), + ); + 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 9017d667aaf5..8bd8a03fe9e6 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.ts @@ -12,6 +12,8 @@ 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, + latestThreadEventAt, listRunningSessionRows, settleRunningSession, } from "../../orchestration/reconcileRunningTurns.ts"; @@ -67,6 +69,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 +151,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 +335,35 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = continue; } + // T3-CUSTOM(expbkt3): BEGIN - the projection's activeTurnId is not + // 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) { + 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 + // 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, 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,