feat: Turn ownership during writes

.changeset/turn-owner-write-fence.md

@@ -0,0 +1,6 @@
+---
+"@truefoundry/trueforge-core": patch
+"@truefoundry/trueforge": patch
+---
+
+Fence turn progress writes with `expected_active_executor_id` so only the owning replica can mutate a running turn; cancel/freeze stays unfenced. Wrong owner raises `TurnExecutorMismatchError`.

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

@@ -137,6 +137,19 @@ export class TurnHandle<TTurnCustom extends object = Record<string, never>> {
     return this.turn.turn_id;
   }
 
+  /** Keys for progress store writes: session, turn, and owning executor fence. */
+  private turnWriteScope(): {
+    session_id: string;
+    turn_id: string;
+    expected_active_executor_id: string;
+  } {
+    return {
+      session_id: this.turn.session_id,
+      turn_id: this.turn.turn_id,
+      expected_active_executor_id: this.turn.active_executor_id,
+    };
+  }
+
   get session_id(): string {
     return this.turn.session_id;
   }
@@ -224,8 +237,7 @@ export class TurnHandle<TTurnCustom extends object = Record<string, never>> {
         thread_id: null,
       };
       await this.store.appendToEvents({
-        session_id: this.turn.session_id,
-        turn_id: this.turn.turn_id,
+        ...this.turnWriteScope(),
         events: [turnCreated],
       });
       yield turnCreated;
@@ -336,8 +348,7 @@ export class TurnHandle<TTurnCustom extends object = Record<string, never>> {
       if (!frozenByStore) {
         try {
           await this.store.updateTurnState({
-            session_id: this.turn.session_id,
-            turn_id: this.turn.turn_id,
+            ...this.turnWriteScope(),
             state: terminalState,
             turn_done_event: turnDone,
           });
@@ -401,10 +412,7 @@ export class TurnHandle<TTurnCustom extends object = Record<string, never>> {
    * the event should be emitted to the consumer (null = side-effect only / skip).
    */
   private async persistExecutionEvent(event: AgentThreadExecutionEvent): Promise<TurnStreamingEvent | null> {
-    const scope = {
-      session_id: this.turn.session_id,
-      turn_id: this.turn.turn_id,
-    };
+    const scope = this.turnWriteScope();
 
     switch (event.type) {
       case HarnessEventType.MODEL_MESSAGE:

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

@@ -105,6 +105,7 @@ export {
   SessionStoreInvariantError,
   SessionStoreNotFoundError,
   TurnAlreadyExistsError,
+  TurnExecutorMismatchError,
   TurnNotFoundError,
   TurnNotRunningError,
 } from './store/SessionStoreErrors';

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

@@ -159,6 +159,7 @@ export interface ListTurnsInput {
 export interface UpdateTurnStateInput {
   session_id: string;
   turn_id: string;
+  expected_active_executor_id: string;
   state: TerminalTurnState;
   /** Caller-built turn.done; written atomically with the state flip in the same tx. */
   turn_done_event: PersistedTurnEvent;
@@ -167,24 +168,28 @@ export interface UpdateTurnStateInput {
 export interface AppendToEventsInput {
   session_id: string;
   turn_id: string;
+  expected_active_executor_id: string;
   events: PersistedTurnEvent[];
 }
 
 export interface AddThreadsInput {
   session_id: string;
   turn_id: string;
+  expected_active_executor_id: string;
   threads: AgentThreadSnapshot[];
 }
 
 export interface RemoveThreadsInput {
   session_id: string;
   turn_id: string;
+  expected_active_executor_id: string;
   thread_ids: string[];
 }
 
 export interface AppendToThreadContextInput {
   session_id: string;
   turn_id: string;
+  expected_active_executor_id: string;
   thread_id: string;
   context: ContextMessage[];
   current_context_usage: CurrentContextUsage | null;
@@ -194,24 +199,28 @@ export interface AppendToThreadContextInput {
 export interface OverwriteThreadContextInput {
   session_id: string;
   turn_id: string;
+  expected_active_executor_id: string;
   event: ThreadOverwriteContextEvent;
 }
 
 export interface PatchMCPServersInput {
   session_id: string;
   turn_id: string;
+  expected_active_executor_id: string;
   mcp_servers: MCPServerInitInfo[];
 }
 
 export interface PatchSandboxInfoInput {
   session_id: string;
   turn_id: string;
+  expected_active_executor_id: string;
   sandbox_info: SandboxInfo;
 }
 
 export interface PatchThreadCapabilityStateInput {
   session_id: string;
   turn_id: string;
+  expected_active_executor_id: string;
   thread_id: string;
   key: string;
   state: JsonValue;

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

@@ -48,6 +48,7 @@ import {
   SessionNotFoundError,
   SessionStoreInvariantError,
   TurnAlreadyExistsError,
+  TurnExecutorMismatchError,
   TurnNotFoundError,
   TurnNotRunningError,
 } from './SessionStoreErrors';
@@ -465,6 +466,13 @@ export class InMemorySessionStore<
     if (turn.state.status !== 'running') {
       throw new TurnNotRunningError(input.turn_id, turn.state);
     }
+    if (turn.active_executor_id !== input.expected_active_executor_id) {
+      throw new TurnExecutorMismatchError({
+        turn_id: input.turn_id,
+        expected_active_executor_id: input.expected_active_executor_id,
+        active_executor_id: turn.active_executor_id,
+      });
+    }
     turn.state = deepCopy(input.state);
     turn.updated_at = new Date();
     const list = this.events.get(tKey);
@@ -475,7 +483,7 @@ export class InMemorySessionStore<
   }
 
   async appendToEvents(input: AppendToEventsInput): Promise<void> {
-    this.requireRunningTurn(input.session_id, input.turn_id);
+    this.requireTurnProgressAllowed(input);
     const tKey = turnKey(input);
     const list = this.events.get(tKey);
     if (!list) {
@@ -507,16 +515,27 @@ export class InMemorySessionStore<
     return turn;
   }
 
-  private requireRunningTurn(sessionId: string, turnId: string): TurnRecord<TTurnCustom> {
-    const turn = this.requireTurn(sessionId, turnId);
+  private requireTurnProgressAllowed(input: {
+    session_id: string;
+    turn_id: string;
+    expected_active_executor_id: string;
+  }): TurnRecord<TTurnCustom> {
+    const turn = this.requireTurn(input.session_id, input.turn_id);
     if (turn.state.status !== 'running') {
-      throw new TurnNotRunningError(turnId, turn.state);
+      throw new TurnNotRunningError(input.turn_id, turn.state);
+    }
+    if (turn.active_executor_id !== input.expected_active_executor_id) {
+      throw new TurnExecutorMismatchError({
+        turn_id: input.turn_id,
+        expected_active_executor_id: input.expected_active_executor_id,
+        active_executor_id: turn.active_executor_id,
+      });
     }
     return turn;
   }
 
   async addThreads(input: AddThreadsInput): Promise<void> {
-    const turn = this.requireRunningTurn(input.session_id, input.turn_id);
+    const turn = this.requireTurnProgressAllowed(input);
     for (const thread of input.threads) {
       turn.snapshot.threads[thread.thread_id] = deepCopy(thread);
     }
@@ -528,7 +547,7 @@ export class InMemorySessionStore<
     if (input.thread_ids.length === 0) {
       return;
     }
-    const turn = this.requireRunningTurn(input.session_id, input.turn_id);
+    const turn = this.requireTurnProgressAllowed(input);
     for (const id of input.thread_ids) {
       Reflect.deleteProperty(turn.snapshot.threads, id);
     }
@@ -537,7 +556,7 @@ export class InMemorySessionStore<
   }
 
   async appendToThreadContext(input: AppendToThreadContextInput): Promise<void> {
-    const turn = this.requireRunningTurn(input.session_id, input.turn_id);
+    const turn = this.requireTurnProgressAllowed(input);
     const thread = turn.snapshot.threads[input.thread_id];
     if (!thread) {
       throw new SessionStoreInvariantError(`Thread not found: ${input.thread_id}`);
@@ -554,7 +573,7 @@ export class InMemorySessionStore<
   }
 
   async overwriteThreadContext(input: OverwriteThreadContextInput): Promise<void> {
-    const turn = this.requireRunningTurn(input.session_id, input.turn_id);
+    const turn = this.requireTurnProgressAllowed(input);
     const threadId = input.event.thread_id;
     const thread = turn.snapshot.threads[threadId];
     if (!thread) {
@@ -567,7 +586,7 @@ export class InMemorySessionStore<
   }
 
   async patchMCPServers(input: PatchMCPServersInput): Promise<void> {
-    const turn = this.requireRunningTurn(input.session_id, input.turn_id);
+    const turn = this.requireTurnProgressAllowed(input);
     turn.snapshot.mcp_servers ??= {};
     for (const server of input.mcp_servers) {
       turn.snapshot.mcp_servers[server.id] = deepCopy(server);
@@ -577,14 +596,14 @@ export class InMemorySessionStore<
   }
 
   async patchSandboxInfo(input: PatchSandboxInfoInput): Promise<void> {
-    const turn = this.requireRunningTurn(input.session_id, input.turn_id);
+    const turn = this.requireTurnProgressAllowed(input);
     turn.snapshot.sandbox_info = deepCopy(input.sandbox_info);
     turn.updated_at = new Date();
     return;
   }
 
   async patchThreadCapabilityState(input: PatchThreadCapabilityStateInput): Promise<void> {
-    const turn = this.requireRunningTurn(input.session_id, input.turn_id);
+    const turn = this.requireTurnProgressAllowed(input);
     const thread = turn.snapshot.threads[input.thread_id];
     if (!thread) {
       throw new SessionStoreInvariantError(`Thread not found: ${input.thread_id}`);

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

@@ -98,6 +98,23 @@ export class TurnNotRunningError extends SessionStoreConflictError {
   }
 }
 
+/** Progress write rejected: turn is still running but the caller is not the active executor. */
+export class TurnExecutorMismatchError extends SessionStoreConflictError {
+  readonly turn_id: string;
+  readonly expected_active_executor_id: string;
+  readonly active_executor_id: string;
+
+  constructor(input: { turn_id: string; expected_active_executor_id: string; active_executor_id: string }) {
+    super(
+      `Turn ${input.turn_id} is owned by executor ${input.active_executor_id}, not ${input.expected_active_executor_id}`,
+    );
+    this.name = 'TurnExecutorMismatchError';
+    this.turn_id = input.turn_id;
+    this.expected_active_executor_id = input.expected_active_executor_id;
+    this.active_executor_id = input.active_executor_id;
+  }
+}
+
 export class InvalidPageTokenError extends SessionStoreConflictError {
   readonly token: string;