fix(server): use T3 sender for Claude userEmail (TEC-1031)

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);
+    }),
+  );
 });

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

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;
     }
   });

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

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`.

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");
+    }),
+  );
+});