feat: Turn steal

packages/trueforge-core/src/agent-session/SessionHandle.ts

@@ -18,6 +18,7 @@ import {
 import type { CreateDynamicSubAgentThread } from '../core/runtime/CreateDynamicSubAgentThread';
 import type { Sandbox } from '../core/sandbox/Sandbox';
 import type { AgentTracing } from '../core/tracing/AgentTracing';
+import { mintActiveExecutorId } from './activeExecutorId';
 import { builtinsFromSpec } from './builtinsFromSpec';
 import type { ITurnResourceResolver, ResolvedAgentDefinition } from './ITurnResourceResolver';
 import type { SessionRecord } from './models/SessionRecord';
@@ -285,7 +286,7 @@ export class SessionHandle<
         first_turn_id: previous?.first_turn_id ?? turnId,
         ancestor_ids: previous ? [...previous.ancestor_ids, previous.turn_id].slice(-MAX_TURN_ANCESTORS) : [],
         previous_turn_id: previousTurnId,
-        active_executor_id: input.active_executor_id,
+        active_executor_id: mintActiveExecutorId(input.active_executor_id),
         state: { status: 'running' },
         input: input.input ?? [],
         created_at: now,

packages/trueforge-core/src/agent-session/activeExecutorId.ts

@@ -0,0 +1,31 @@
+/**
+ * `active_executor_id` is `{executorId}.{generation}`. Generation is 4 hex
+ * chars from `randomBytes` so a stale in-memory run cannot persist after
+ * steal/rebuild. Rows minted before this grammar parse as executor-only.
+ */
+import { randomBytes } from 'node:crypto';
+
+const GENERATION = /^[0-9a-f]{4}$/;
+
+export function parseActiveExecutorId(value: string): { executorId: string; generation: string | undefined } {
+  const lastDot = value.lastIndexOf('.');
+  if (lastDot <= 0) {
+    return { executorId: value, generation: undefined };
+  }
+  const generation = value.slice(lastDot + 1);
+  if (!GENERATION.test(generation)) {
+    return { executorId: value, generation: undefined };
+  }
+  return { executorId: value.slice(0, lastDot), generation };
+}
+
+/** `{executorId}.{4 hex chars}`, never equal to `previous`. */
+export function mintActiveExecutorId(executorId: string, previous?: string): string {
+  const id = parseActiveExecutorId(executorId).executorId;
+  for (;;) {
+    const next = `${id}.${randomBytes(2).toString('hex')}`;
+    if (next !== previous) {
+      return next;
+    }
+  }
+}

packages/trueforge-core/src/agent-session/index.ts

@@ -71,6 +71,7 @@ export type {
 export { TokenPaginationSchema } from './schemas/pagination';
 export type { TokenPagination } from './schemas/pagination';
 
+export { mintActiveExecutorId, parseActiveExecutorId } from './activeExecutorId';
 export type { SessionRecord } from './models/SessionRecord';
 export { MAIN_THREAD_ID } from './models/TurnRecord';
 export type { TurnRecord, TurnSnapshot } from './models/TurnRecord';
@@ -81,6 +82,7 @@ export type {
   AddThreadsInput,
   AppendToEventsInput,
   AppendToThreadContextInput,
+  ClaimTurnOwnershipInput,
   CreateSessionInput,
   CreateTurnInput,
   DeleteSessionInput,

packages/trueforge-core/src/agent-session/store/ISessionStore.ts

@@ -164,6 +164,14 @@ export interface UpdateTurnStateInput {
   turn_done_event: PersistedTurnEvent;
 }
 
+/** Steal: one winner when two replicas claim a paused turn. */
+export interface ClaimTurnOwnershipInput {
+  session_id: string;
+  turn_id: string;
+  expected_active_executor_id: string;
+  new_active_executor_id: string;
+}
+
 export interface AppendToEventsInput {
   session_id: string;
   turn_id: string;
@@ -364,6 +372,13 @@ export interface ISessionStore<
    */
   updateTurnState(input: UpdateTurnStateInput): Promise<void>;
 
+  /**
+   * Claim `active_executor_id` when the turn is owned by
+   * `expected_active_executor_id`. True if this caller won. False if another
+   * replica already claimed or the tip is not stealable. Missing → {@link TurnNotFoundError}.
+   */
+  claimTurnOwnership(input: ClaimTurnOwnershipInput): Promise<boolean>;
+
   /**
    * Durable event log for the turn. MUST include lifecycle rows: a
    * `TurnCreatedEvent` at the start of the stream and a terminal `TurnDoneEvent`

packages/trueforge-core/src/agent-session/store/InMemorySessionStore.ts

@@ -1,5 +1,6 @@
 import type { AgentThreadSnapshot } from '../../core/runtime/AgentThread.types';
 import { getEmptyCurrentContextUsage } from '../../core/runtime/contextUsage';
+import { mintActiveExecutorId } from '../activeExecutorId';
 import type { SessionRecord } from '../models/SessionRecord';
 import type { TurnRecord, TurnSnapshot } from '../models/TurnRecord';
 import type { PersistedTurnEvent, SessionEventItem } from '../schemas/events';
@@ -10,6 +11,7 @@ import type {
   AddThreadsInput,
   AppendToEventsInput,
   AppendToThreadContextInput,
+  ClaimTurnOwnershipInput,
   CreateSessionInput,
   CreateTurnInput,
   DeleteSessionInput,
@@ -490,6 +492,16 @@ export class InMemorySessionStore<
     this.addTerminalSessionMetrics(input.session_id, turn.created_at, input.state);
   }
 
+  async claimTurnOwnership(input: ClaimTurnOwnershipInput): Promise<boolean> {
+    const turn = this.requireTurn(input.session_id, input.turn_id);
+    if (turn.state.status !== 'paused' || turn.active_executor_id !== input.expected_active_executor_id) {
+      return false;
+    }
+    turn.active_executor_id = mintActiveExecutorId(input.new_active_executor_id, turn.active_executor_id);
+    turn.updated_at = new Date();
+    return true;
+  }
+
   async appendToEvents(input: AppendToEventsInput): Promise<void> {
     this.requireRunningTurn(input.session_id, input.turn_id);
     const tKey = turnKey(input);

packages/trueforge-core/tests/agent-session/activeExecutorId.test.ts

@@ -0,0 +1,23 @@
+import { mintActiveExecutorId, parseActiveExecutorId } from '../../src/agent-session/activeExecutorId';
+
+describe('parseActiveExecutorId', () => {
+  it('splits executorId.generation', () => {
+    expect(parseActiveExecutorId('abc123.a1f0')).toEqual({ executorId: 'abc123', generation: 'a1f0' });
+  });
+
+  it('treats a bare id as executor-only', () => {
+    expect(parseActiveExecutorId('abc123')).toEqual({ executorId: 'abc123', generation: undefined });
+  });
+
+  it('does not treat a short suffix as generation', () => {
+    expect(parseActiveExecutorId('abc123.12')).toEqual({ executorId: 'abc123.12', generation: undefined });
+  });
+});
+
+describe('mintActiveExecutorId', () => {
+  it('appends 4 hex chars and strips an existing suffix', () => {
+    const minted = mintActiveExecutorId('abc123.a1f0', 'abc123.a1f0');
+    expect(minted).toMatch(/^abc123\.[0-9a-f]{4}$/);
+    expect(minted).not.toBe('abc123.a1f0');
+  });
+});

packages/trueforge-core/tests/agent-session/store/storeContractSuite.ts

@@ -2199,6 +2199,68 @@ export function runStoreContractSuite(createStore: () => ISessionStore) {
     });
   });
 
+  describe('claimTurnOwnership', () => {
+    it('missing turn → not found', async () => {
+      const store = createStore();
+      await seedSession(store);
+      await expect(
+        store.claimTurnOwnership({
+          session_id: sessionId,
+          turn_id: missingTurnId,
+          expected_active_executor_id: TEST_ACTIVE_EXECUTOR_ID,
+          new_active_executor_id: 'stealer',
+        }),
+      ).rejects.toBeInstanceOf(TurnNotFoundError);
+    });
+
+    it('running turn → false and owner unchanged', async () => {
+      const store = createStore();
+      await seedSession(store);
+      await store.createTurn(makeCreateTurnInput({ sessionId, turnId: 'turn-1' }));
+      await expect(
+        store.claimTurnOwnership({
+          session_id: sessionId,
+          turn_id: 'turn-1',
+          expected_active_executor_id: TEST_ACTIVE_EXECUTOR_ID,
+          new_active_executor_id: 'stealer',
+        }),
+      ).resolves.toBe(false);
+      const turn = await store.getTurn({ session_id: sessionId, turn_id: 'turn-1' });
+      expect(mustGet(turn).active_executor_id).toBe(TEST_ACTIVE_EXECUTOR_ID);
+    });
+
+    it('wrong expected owner → false', async () => {
+      const store = createStore();
+      await seedSession(store);
+      await store.createTurn(makeCreateTurnInput({ sessionId, turnId: 'turn-1' }));
+      await expect(
+        store.claimTurnOwnership({
+          session_id: sessionId,
+          turn_id: 'turn-1',
+          expected_active_executor_id: 'someone-else',
+          new_active_executor_id: 'stealer',
+        }),
+      ).resolves.toBe(false);
+      const turn = await store.getTurn({ session_id: sessionId, turn_id: 'turn-1' });
+      expect(mustGet(turn).active_executor_id).toBe(TEST_ACTIVE_EXECUTOR_ID);
+    });
+
+    it('terminal turn → false', async () => {
+      const store = createStore();
+      await seedSession(store);
+      await store.createTurn(makeCreateTurnInput({ sessionId, turnId: 'turn-1' }));
+      await finishTurn(store, 'turn-1');
+      await expect(
+        store.claimTurnOwnership({
+          session_id: sessionId,
+          turn_id: 'turn-1',
+          expected_active_executor_id: TEST_ACTIVE_EXECUTOR_ID,
+          new_active_executor_id: 'stealer',
+        }),
+      ).resolves.toBe(false);
+    });
+  });
+
   describe('events + threads + capability_state', () => {
     it('appendToEvents orders by monotonic event id, not append call order', async () => {
       const store = createStore();

packages/trueforge/package.json

@@ -69,6 +69,7 @@
     "@sentry/node": "^10.74.0",
     "@truefoundry/trueforge-core": "workspace:*",
     "@truefoundry/trueforge-sdk": "workspace:*",
+    "async-mutex": "^0.5.0",
     "better-sqlite3": "^13.0.3",
     "cron-parser": "^5.4.0",
     "env-paths": "^4.0.0",