diff --git a/packages/runtime-host/src/__tests__/goal-root-authority.test.ts b/packages/runtime-host/src/__tests__/goal-root-authority.test.ts
index 02f4acf334..7f13ca5250 100644
--- a/packages/runtime-host/src/__tests__/goal-root-authority.test.ts
+++ b/packages/runtime-host/src/__tests__/goal-root-authority.test.ts
@@ -642,6 +642,7 @@ async function createFixture(options: { recoverAdmissions?: boolean } = {}): Pro
     executeRoot: (input) =>
       executeHostedExecutionToSettlement(requireCoordinator(coordinator), input),
     stopRoot: (identity, input) => requireCoordinator(coordinator).stopRoot(identity, input),
+    stopRootRun: (identity, input) => requireCoordinator(coordinator).stopRootRun(identity, input),
     stopSession: (sessionId, input) =>
       requireCoordinator(coordinator).stopSession(sessionId, input),
   };
diff --git a/packages/runtime-host/src/__tests__/interaction-coordinator.test.ts b/packages/runtime-host/src/__tests__/interaction-coordinator.test.ts
index 680f4395ff..3946604d5e 100644
--- a/packages/runtime-host/src/__tests__/interaction-coordinator.test.ts
+++ b/packages/runtime-host/src/__tests__/interaction-coordinator.test.ts
@@ -1827,6 +1827,7 @@ async function withGraphInteractionActivity(
         runClaimedAgentGraphIntent: async () => {
           throw new Error('An activity read cannot dispatch work');
         },
+        stopAgentGraphActivation: async () => {},
         stopSession: async () => {},
       },
       newId: randomUUID,
diff --git a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts
index 102c15a343..7696373ec1 100644
--- a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts
+++ b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts
@@ -27,6 +27,9 @@ import {
 } from '@maka/runtime/goal-continuation';
 import { HostGoalExecutionCoordinator } from '../server/goal-execution-coordinator.js';
 import { runtimeInvocationOutcome } from '@maka/core/runtime-invocation';
+import type { AgentGraphIntentClaim } from '@maka/core/agent-graph-control';
+import type { AgentGraphRunnableIntent } from '@maka/runtime/stream-graph-readiness';
+import { fingerprintAgentGraphRunnableIntent } from '@maka/runtime/stream-graph-admission';
 import { createRunCompositionSnapshot } from '@maka/core/run-composition';
 import {
   readLogicalRuntimeExecution,
@@ -3364,6 +3367,8 @@ test('hosted linked child roots share admission, message, terminal, and stop aut
       executeRoot: (input) =>
         executeHostedExecutionToSettlement(requireCoordinator(coordinator), input),
       stopRoot: (identity, input) => requireCoordinator(coordinator).stopRoot(identity, input),
+      stopRootRun: (identity, input) =>
+        requireCoordinator(coordinator).stopRootRun(identity, input),
       stopSession: (sessionId, input) =>
         requireCoordinator(coordinator).stopSession(sessionId, input),
     };
@@ -5963,6 +5968,379 @@ test('public turn.stop takes over an earlier closure claim queued behind its lea
   }
 });
 
+for (const mode of ['scoped', 'session', 'retry'] as const) {
+  const scoped = mode !== 'session';
+  test(`Hosted graph ${mode === 'retry' ? 'scoped stop redelivers after failure while preserving Host drain' : scoped ? 'scoped stop preserves queued work' : 'Session stop cancels queued work'}`, {
+    timeout: 10_000,
+  }, async () => {
+    const graphClaims: AgentGraphIntentClaim[] = [];
+    const gates = [deferred<void>(), deferred<void>(), deferred<void>()];
+    const entered = [deferred<void>(), deferred<void>(), deferred<void>()];
+    let sends = 0;
+    let stops = 0;
+    let current = -1;
+    const fixture = await createFailureFixture({
+      graphClaims,
+      withInteractions: true,
+      childTools: IMPLEMENTATION_AGENT_DEFINITION.tools.map(testTool),
+      registerBackend: (backends) =>
+        backends.register(
+          'ai-sdk',
+          (context) =>
+            new (class extends FakeBackend {
+              override async *send(input: BackendSendInput): AsyncIterable<SessionEvent> {
+                const index = sends++;
+                current = index;
+                entered[index]!.resolve();
+                yield {
+                  type: 'text_delta',
+                  id: randomUUID(),
+                  turnId: input.turnId,
+                  ts: Date.now(),
+                  messageId: randomUUID(),
+                  text: `result-${index}`,
+                };
+                await gates[index]!.promise;
+                yield {
+                  type: 'complete',
+                  id: randomUUID(),
+                  turnId: input.turnId,
+                  ts: Date.now(),
+                  stopReason: 'end_turn',
+                };
+              }
+              override async stop(): Promise<void> {
+                stops += 1;
+                gates[current]?.resolve();
+              }
+            })(context),
+        ),
+    });
+    if (mode === 'retry') {
+      const deliver = fixture.manager.deliverHostedRootStop.bind(fixture.manager);
+      let failDelivery = true;
+      fixture.manager.deliverHostedRootStop = async (...args) => {
+        if (failDelivery) {
+          failDelivery = false;
+          throw new Error('temporary stop delivery failure');
+        }
+        return deliver(...args);
+      };
+    }
+    let releaseFence: (() => void) | undefined;
+    try {
+      const { child, makeInput } = await createHostedGraphChild(fixture, graphClaims);
+      const a = makeInput(0);
+      const b = makeInput(1);
+      const first = fixture.manager.runClaimedAgentGraphIntent(a.input);
+      await entered[0]!.promise;
+      const second = fixture.manager.runClaimedAgentGraphIntent(b.input);
+      const settled = Promise.allSettled([first, second]);
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      const identity = {
+        sessionId: child.id,
+        runId: a.claim.targetRunId,
+        turnId: a.claim.targetTurnId,
+      };
+      const fenceEntered = deferred<void>();
+      const fenceRelease = deferred<void>();
+      releaseFence = () => fenceRelease.resolve();
+      const fence = fixture.sessionAdmission.run(child.id, async () => {
+        fenceEntered.resolve();
+        await fenceRelease.promise;
+      });
+      await fenceEntered.promise;
+      const stopping = scoped
+        ? fixture.manager.stopAgentGraphActivation(identity, {
+            source: 'graph_supervisor',
+          })
+        : fixture.manager.stopSession(child.id, { source: 'stop_button' });
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      assert.equal(stops, 0, 'Runtime stop must wait for the Host admission fence');
+      fenceRelease.resolve();
+      await fence;
+      if (mode === 'retry') {
+        await assert.rejects(stopping, /temporary stop delivery failure/);
+        assert.equal(sends, 1);
+        await fixture.manager.stopAgentGraphActivation(identity, { source: 'graph_supervisor' });
+      } else await stopping;
+      gates[1]!.resolve();
+      const outcomes = await settled;
+      assert.equal(outcomes[0]!.status, 'fulfilled');
+      if (outcomes[0]!.status === 'fulfilled') assert.equal(outcomes[0]!.value.status, 'cancelled');
+      assert.equal(outcomes[1]!.status, mode === 'scoped' ? 'fulfilled' : 'rejected');
+      if (mode === 'scoped' && outcomes[1]!.status === 'fulfilled')
+        assert.equal(outcomes[1]!.value.status, 'completed');
+      assert.equal(sends, mode === 'scoped' ? 2 : 1);
+      if (mode === 'scoped') {
+        const c = makeInput(2);
+        const third = fixture.manager.runClaimedAgentGraphIntent(c.input);
+        await entered[2]!.promise;
+        await fixture.manager.stopAgentGraphActivation(identity, {
+          source: 'graph_supervisor',
+        });
+        assert.equal(stops, 1);
+        gates[2]!.resolve();
+        assert.equal((await third).status, 'completed');
+        assert.equal(
+          runtimeInvocationOutcome(
+            (await fixture.stores.runtimeEventStore.listSessionInvocations(child.id)).find(
+              (run) => run.runId === c.claim.targetRunId,
+            )!,
+          ),
+          'completed',
+        );
+      }
+      assert.equal(fixture.drainRequested(), mode === 'retry');
+    } finally {
+      releaseFence?.();
+      gates.forEach((gate) => gate.resolve());
+      await fixture.coordinator.close();
+      await fixture.messages.close();
+      await fixture.interactions?.close();
+      await fixture.dispose();
+    }
+  });
+}
+
+test('public turn.stop cancels a root Turn whose Run attaches after the stop', {
+  timeout: 10_000,
+}, async () => {
+  const attachEntered = deferred<void>();
+  const releaseAttach = deferred<void>();
+  let holdAttach = true;
+  let backend: LinkedChildAuthorityBackend | undefined;
+  const fixture = await createFailureFixture({
+    resolveFreshTurnToolMode: async () => {
+      if (holdAttach) {
+        holdAttach = false;
+        attachEntered.resolve();
+        await releaseAttach.promise;
+      }
+      return undefined;
+    },
+    registerBackend: (backends) => {
+      backends.register('ai-sdk', (context) => {
+        backend = new LinkedChildAuthorityBackend(context.sessionId);
+        return backend;
+      });
+    },
+  });
+  try {
+    const turnId = 'turn-stop-before-run-attach';
+    const starting = fixture.interactiveTurns.handlers['turn.start'](
+      { sessionId: fixture.sessionId, turnId, content: { text: HOLD_EXTERNAL_PROMPT } },
+      operationContext(fixture.hostEpoch, fixture.acquireResidency),
+    );
+    await completesWithin(attachEntered.promise, 2_000, 'Host activation before Run attach');
+    const admission = await fixture.stores.agentRunStore.readRootTurnAdmission(
+      fixture.sessionId,
+      turnId,
+    );
+    assert.ok(admission);
+    const delivered = deferred<void>();
+    const deliver = fixture.manager.deliverHostedRootStop.bind(fixture.manager);
+    fixture.manager.deliverHostedRootStop = (...args) => {
+      const delivery = deliver(...args);
+      delivered.resolve();
+      return delivery;
+    };
+    const stopping = fixture.turnControl.handlers['turn.stop'](
+      { sessionId: fixture.sessionId, turnId, runId: admission.runId },
+      operationContext(fixture.hostEpoch, fixture.acquireResidency),
+    );
+    await completesWithin(delivered.promise, 2_000, 'Runtime stop delivery before attach');
+    releaseAttach.resolve();
+    const stopped = await completesWithin(stopping, 2_000, 'stop of a Run attached after it');
+    assert.equal(stopped.ok, true);
+    await completesWithin(starting, 2_000, 'stopped Turn start');
+    const run = await readInvocation(fixture.stores, fixture.sessionId, admission.runId);
+    const terminal = classifyTerminalRuntimeLedger(
+      run,
+      await fixture.stores.runtimeEventStore.readImmutableRuntimeEvents(
+        fixture.sessionId,
+        admission.runId,
+      ),
+    );
+    assert.equal(terminal.kind, 'fact');
+    if (terminal.kind === 'fact') assert.equal(terminal.fact.runStatus, 'cancelled');
+  } finally {
+    releaseAttach.resolve();
+    backend?.release();
+    await fixture.coordinator.close();
+    await fixture.messages.close();
+    await fixture.dispose();
+  }
+});
+
+test('Hosted graph stop before durable admission refuses the submitted activation', {
+  timeout: 10_000,
+}, async () => {
+  const graphClaims: AgentGraphIntentClaim[] = [];
+  const submitted = deferred<void>();
+  const releaseSubmission = deferred<void>();
+  let sends = 0;
+  const fixture = await createFailureFixture({
+    graphClaims,
+    beforeExecuteRoot: async () => {
+      submitted.resolve();
+      await releaseSubmission.promise;
+    },
+    childTools: IMPLEMENTATION_AGENT_DEFINITION.tools.map(testTool),
+    registerBackend: (backends) =>
+      backends.register(
+        'ai-sdk',
+        (context) =>
+          new (class extends FakeBackend {
+            override async *send(input: BackendSendInput): AsyncIterable<SessionEvent> {
+              sends += 1;
+              yield* super.send(input);
+            }
+          })(context),
+      ),
+  });
+  let running: Promise<unknown> | undefined;
+  try {
+    const { child, makeInput } = await createHostedGraphChild(fixture, graphClaims);
+    const stoppedInput = makeInput(0);
+    running = fixture.manager
+      .runClaimedAgentGraphIntent(stoppedInput.input)
+      .catch((error) => error);
+    await completesWithin(submitted.promise, 2_000, 'Host submission');
+    const stopping = fixture.manager.stopAgentGraphActivation(
+      {
+        sessionId: child.id,
+        runId: stoppedInput.claim.targetRunId,
+        turnId: stoppedInput.claim.targetTurnId,
+      },
+      { source: 'graph_supervisor' },
+    );
+    // Let a Host stop that cannot see the pending admission settle before the
+    // submission reaches the Host gate; a local stop stays pending until then.
+    await Promise.race([
+      stopping.catch(() => undefined),
+      new Promise<void>((resolve) => setTimeout(resolve, 100)),
+    ]);
+    releaseSubmission.resolve();
+    await completesWithin(stopping, 2_000, 'stop before durable admission');
+    assert.match(String(await running), /cancelled before Runtime admission/);
+    assert.equal(sends, 0);
+    assert.equal(
+      await fixture.stores.agentRunStore.readRootTurnAdmission(
+        child.id,
+        stoppedInput.claim.targetTurnId,
+      ),
+      undefined,
+    );
+    const next = makeInput(1);
+    assert.equal(
+      (await fixture.manager.runClaimedAgentGraphIntent(next.input)).status,
+      'completed',
+    );
+    assert.equal(sends, 1);
+  } finally {
+    releaseSubmission.resolve();
+    await running;
+    await fixture.coordinator.close();
+    await fixture.messages.close();
+    await fixture.dispose();
+  }
+});
+
+/** A linked graph child whose claims the Hosted fixture resolves from `graphClaims`. */
+async function createHostedGraphChild(
+  fixture: Awaited<ReturnType<typeof createFailureFixture>>,
+  graphClaims: AgentGraphIntentClaim[],
+) {
+  const parent = await fixture.stores.sessionStore.readHeaderSnapshot(fixture.sessionId);
+  const { header: child } = await fixture.stores.sessionStore.createSubagent({
+    cwd: parent.cwd,
+    llmConnectionId: parent.llmConnectionId,
+    llmConnectionSlug: 'fake',
+    model: 'fake-model',
+    permissionMode: 'ask',
+    collaborationMode: 'agent',
+    orchestrationMode: 'default',
+    subagentParent: {
+      kind: 'subagent',
+      parentSessionId: parent.id,
+      spawnedBy: {
+        parentRunId: 'parent-run',
+        parentTurnId: 'parent-turn',
+        toolCallId: 'graph-create',
+      },
+      lifecycle: 'foreground',
+    },
+    subagentRuntime: {
+      schemaVersion: 1,
+      definitionVersion: IMPLEMENTATION_AGENT_DEFINITION.definitionVersion,
+      agentId: IMPLEMENTATION_AGENT_DEFINITION.id,
+      agentName: IMPLEMENTATION_AGENT_DEFINITION.name,
+      profile: 'implementation',
+      systemPrompt: IMPLEMENTATION_AGENT_DEFINITION.systemPrompt,
+      toolNames: [...IMPLEMENTATION_AGENT_DEFINITION.tools],
+      categoryPolicy: {},
+    },
+    subagentSpawn: {
+      schemaVersion: 1,
+      requestFingerprint: 'c'.repeat(64),
+      initialTurnId: 'scoped-turn-0',
+      initialRunId: 'scoped-run-0',
+    },
+  });
+  const makeInput = (index: number) => {
+    const char = String(index + 1);
+    const prompt = `bounded task ${index}`;
+    const intent: AgentGraphRunnableIntent = {
+      schemaVersion: 1,
+      graphId: 'host-scoped-graph',
+      intentId: `graph_intent_${char.repeat(32)}`,
+      readinessContextFingerprint: `sha256:${'a'.repeat(64)}`,
+      policyFingerprint: `sha256:${'b'.repeat(64)}`,
+      readinessId: `work-${index}`,
+      operatorId: 'operator',
+      targetSessionId: child.id,
+      policyKind: 'map',
+      triggerRouteIds: [],
+      triggerRecordIds: [],
+    };
+    const claim: AgentGraphIntentClaim = {
+      schemaVersion: 1,
+      claimId: `graph_claim_${char.repeat(32)}`,
+      graphId: intent.graphId,
+      intentId: intent.intentId,
+      intentFingerprint: fingerprintAgentGraphRunnableIntent({
+        intent,
+        executionInput: { prompt },
+      }),
+      readinessContextFingerprint: intent.readinessContextFingerprint,
+      targetOperatorId: 'operator',
+      targetSessionId: child.id,
+      targetTurnId: `scoped-turn-${index}`,
+      targetRunId: `scoped-run-${index}`,
+      claimedAt: Date.now(),
+    };
+    graphClaims.push(claim);
+    return {
+      claim,
+      input: {
+        intent,
+        graphId: claim.graphId,
+        intentId: claim.intentId,
+        prompt,
+        claimStore: {
+          readAgentGraphIntentClaim: async () => claim,
+          listAgentGraphIntentClaims: async () => [claim],
+          claimAgentGraphIntent: async () => {
+            throw new Error('already claimed');
+          },
+        },
+      },
+    };
+  };
+  return { child, makeInput };
+}
+
 function executeClaimedGraphRoot(
   fixture: Awaited<ReturnType<typeof createFailureFixture>>,
   input: {
@@ -6636,6 +7014,10 @@ for (const decision of ['cancel', 'resume', 'detach', 'blocked'] as const) {
 }
 
 async function createFailureFixture(options: {
+  graphClaims?: AgentGraphIntentClaim[];
+  beforeExecuteRoot?(): Promise<void>;
+  /** Runtime calls this in `startTurn` after the Host activated the Turn, before Run attach. */
+  resolveFreshTurnToolMode?(): Promise<undefined>;
   registerBackend(backends: BackendRegistry): void;
   afterHandoffSeal?(): Promise<void>;
   directoryHostId?: string;
@@ -6822,9 +7204,47 @@ async function createFailureFixture(options: {
       : stores.runtimeEventStore,
     backends,
     ...(options.childTools ? { childTools: options.childTools } : {}),
+    ...(options.resolveFreshTurnToolMode
+      ? { resolveFreshTurnToolMode: options.resolveFreshTurnToolMode }
+      : {}),
     newId: randomUUID,
     now: Date.now,
-    messageAuthority: options.wrapMessageAuthority?.(messages) ?? messages,
+    messageAuthority: options.graphClaims
+      ? {
+          bindRun: messages.bindRun.bind(messages),
+          executeRoot: async (input: Parameters<RuntimeHostedRootAuthority['executeRoot']>[0]) => {
+            await options.beforeExecuteRoot?.();
+            return executeHostedExecutionToSettlement(requireCoordinator(coordinator), input);
+          },
+          stopRoot: (...args: Parameters<RuntimeHostedRootAuthority['stopRoot']>) =>
+            requireCoordinator(coordinator).stopRoot(...args),
+          stopRootRun: (...args: Parameters<RuntimeHostedRootAuthority['stopRootRun']>) =>
+            requireCoordinator(coordinator).stopRootRun(...args),
+          stopSession: (
+            sessionId: string,
+            input: Parameters<RuntimeHostedRootAuthority['stopSession']>[1],
+          ) => requireCoordinator(coordinator).stopSession(sessionId, input),
+        }
+      : (options.wrapMessageAuthority?.(messages) ?? messages),
+    ...(options.graphClaims
+      ? {
+          hostedAgentGraphExecution: {
+            readAgentGraphIntentClaim: async (graphId: string, intentId: string) =>
+              options.graphClaims!.find(
+                (claim) => claim.graphId === graphId && claim.intentId === intentId,
+              ),
+            readRootTurnAdmissionIdentity: async (sessionId: string, turnId: string) => {
+              const admission = await stores.agentRunStore.readRootTurnAdmission(sessionId, turnId);
+              return admission
+                ? {
+                    runId: admission.runId,
+                    userMessageId: admission.userMessageId,
+                  }
+                : undefined;
+            },
+          },
+        }
+      : {}),
     ...(options.continuationSafety
       ? {
           safeBoundaryResumeEnabled: true,
diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts
index 82a82fd142..6e62b69bd4 100644
--- a/packages/runtime-host/src/server/execution-composition.ts
+++ b/packages/runtime-host/src/server/execution-composition.ts
@@ -1266,6 +1266,8 @@ export async function createExecutionRuntimeHostComposition(
         executeHostedExecutionToSettlement(requireRootCoordinator(rootCoordinator), input),
       stopRoot: (identity, input) =>
         requireRootCoordinator(rootCoordinator).stopRoot(identity, input),
+      stopRootRun: (identity, input) =>
+        requireRootCoordinator(rootCoordinator).stopRootRun(identity, input),
       stopSession: (sessionId, input) =>
         requireRootCoordinator(rootCoordinator).stopSession(sessionId, input),
     };
diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts
index 1b3f59f458..6ef935bedc 100644
--- a/packages/runtime-host/src/server/root-turn-coordinator.ts
+++ b/packages/runtime-host/src/server/root-turn-coordinator.ts
@@ -1267,6 +1267,50 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
     });
   }
 
+  /**
+   * Activation-scoped graph stop: the same durable Host fence as `stopRoot`,
+   * but the Runtime stop reaches only this exact Run's owner, is redelivered
+   * after a failed attempt, and retries retained cleanup of a completed Run.
+   */
+  stopRootRun(identity: RuntimeMessageRunIdentity, input: StopSessionInput = {}): Promise<void> {
+    normalizeStopSessionSource(input.source, input.workHubActionId);
+    return this.runCommand(async () => {
+      const declared = await this.sessionAdmission.run(identity.sessionId, (lease) =>
+        this.declareStopFence(
+          identity,
+          () => this.messages.commitStopFence(identity),
+          lease,
+          input,
+          true,
+        ),
+      );
+      await declared?.deliverStop();
+      await declared?.active.startSettled.promise;
+      const disposition = await this.sessionAdmission.run(identity.sessionId, (lease) =>
+        this.prepareStopDisposition(identity, () => this.messages.commitStopFence(identity), lease),
+      );
+      if (disposition.kind === 'complete') {
+        if (!disposition.outcome.ok) throwHostedStopError(identity.sessionId, disposition.outcome);
+        const logical = await readLogicalRuntimeExecutionForRun(
+          this.stores.runtimeEventStore,
+          identity,
+        );
+        await this.deliverRuntimeStopIntent(identity.sessionId, input, {
+          ...identity,
+          runId: logical?.tip.runId ?? identity.runId,
+        });
+        return;
+      }
+      if (disposition.kind === 'request_stop') {
+        await this.deliverRuntimeStopIntent(identity.sessionId, input, {
+          ...identity,
+          runId: disposition.active.continuation?.runId ?? identity.runId,
+        });
+      }
+      await disposition.active.done;
+    });
+  }
+
   stopSession(sessionId: string, input: StopSessionInput = {}): Promise<void> {
     normalizeStopSessionSource(input.source, input.workHubActionId);
     return this.runCommand(async () => {
@@ -2700,6 +2744,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
     commitQueueFence: () => QueueFenceResult,
     admission: SessionAdmissionLease,
     stopInput: StopSessionInput = {},
+    scoped = false,
   ): Promise<DeclaredStopFence | undefined> {
     const active = this.#executions.get(input.sessionId);
     if (!active || active.turnId !== input.turnId || active.runId !== input.runId) {
@@ -2718,13 +2763,22 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
       'turn_stopped',
       admission,
     );
-    const shouldDeliverStop = !active.stopRequested;
+    const shouldDeliverStop = scoped || !active.stopRequested;
     active.stopRequested = stopInput;
     return {
       active,
       deliverStop: () =>
         shouldDeliverStop
-          ? this.deliverRuntimeStopIntent(input.sessionId, stopInput)
+          ? this.deliverRuntimeStopIntent(
+              input.sessionId,
+              stopInput,
+              scoped
+                ? {
+                    ...input,
+                    runId: active.continuation?.runId ?? active.runId,
+                  }
+                : undefined,
+            )
           : Promise.resolve(),
     };
   }
@@ -3376,8 +3430,9 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
   private async deliverRuntimeStopIntent(
     sessionId: string,
     input: StopSessionInput = { source: 'stop_button' },
+    identity?: RuntimeMessageRunIdentity,
   ): Promise<void> {
-    await this.manager.deliverHostedRootStop(sessionId, input);
+    await this.manager.deliverHostedRootStop(sessionId, input, identity);
   }
 
   private async stopActiveTurn(sessionId: string, active: ActiveRootTurn): Promise<void> {
diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts
index 997cf7076b..261dde8164 100644
--- a/packages/runtime/src/__tests__/session-manager.test.ts
+++ b/packages/runtime/src/__tests__/session-manager.test.ts
@@ -2084,6 +2084,428 @@ describe('SessionManager claimed graph intent execution', () => {
     assert.strictEqual((await runStore.listSessionInvocations(child.id)).length, 2);
   });
 
+  test('scoped graph stop preserves another queued claim on the same Session', {
+    timeout: 2_000,
+  }, async () => {
+    const { manager, child, runStore, first, claims } = await createQueuedGraphScenario();
+    const [firstClaim, secondClaim] = claims;
+    const second = manager.runClaimedAgentGraphIntent(
+      graphExecutionInput(secondClaim, 'queued activation'),
+    );
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    await manager.stopAgentGraphActivation(
+      {
+        sessionId: child.id,
+        runId: firstClaim.targetRunId,
+        turnId: firstClaim.targetTurnId,
+      },
+      { source: 'graph_supervisor' },
+    );
+    assert.strictEqual((await first).status, 'cancelled');
+    assert.strictEqual((await second).status, 'completed');
+    assert.deepStrictEqual(
+      (await runStore.listSessionInvocations(child.id)).map((run) => [
+        run.turnId,
+        runtimeInvocationOutcome(run),
+      ]),
+      [
+        [firstClaim.targetTurnId, 'cancelled'],
+        [secondClaim.targetTurnId, 'completed'],
+      ],
+    );
+  });
+
+  test('scoped graph stop cancels only a queued capability before runtime admission', {
+    timeout: 2_000,
+  }, async () => {
+    const { manager, child, runStore, activeGate, first, claims } =
+      await createQueuedGraphScenario();
+    const [firstClaim, secondClaim, thirdClaim] = claims;
+    const queued = manager.runClaimedAgentGraphIntent(
+      graphExecutionInput(secondClaim, 'queued activation'),
+    );
+    const rejected = assert.rejects(queued, /cancelled before dispatch/);
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    await manager.stopAgentGraphActivation({
+      sessionId: child.id,
+      runId: secondClaim.targetRunId,
+      turnId: secondClaim.targetTurnId,
+    });
+    await rejected;
+    assert.strictEqual(
+      (await readInvocation(runStore, child.id, firstClaim.targetRunId)).terminalEvent,
+      undefined,
+    );
+    activeGate.release();
+    assert.strictEqual((await first).status, 'completed');
+    assert.strictEqual(
+      (
+        await manager.runClaimedAgentGraphIntent(
+          graphExecutionInput(thirdClaim, 'third activation'),
+        )
+      ).status,
+      'completed',
+    );
+    assert.deepStrictEqual(
+      (await runStore.listSessionInvocations(child.id)).map((run) => run.turnId),
+      [firstClaim.targetTurnId, thirdClaim.targetTurnId],
+    );
+  });
+
+  test('scoped graph stop retries failed terminal persistence on its captured owner', {
+    timeout: 2_000,
+  }, async () => {
+    const store = new MemorySessionStore();
+    let rejectAbort = true;
+    const runStore = new MemoryAgentRunStore({
+      beforeRuntimeEventAppend: (_sessionId, _runId, event) => {
+        if (rejectAbort && event.status === 'aborted')
+          throw new Error('scoped abort persistence unavailable');
+      },
+    });
+    const backends = new BackendRegistry();
+    const sendGate = makeGate();
+    const ready = makeGate();
+    backends.register('ai-sdk', (ctx) => new CountingStopBackend(ctx, sendGate));
+    const manager = new SessionManager({
+      store,
+      runStore,
+      runtimeEventStore: runStore,
+      backends,
+      childTools: [testTool('Read'), testTool('Glob'), testTool('Grep')],
+      newId: nextId(),
+      now: nextNow(80),
+    });
+    const parent = await manager.createSession(makeInput());
+    const child = await createGraphOperatorSession(store, parent.id);
+    const claim = graphIntentClaim({ targetSessionId: child.id }, 'retry scoped stop');
+    const running = manager.runClaimedAgentGraphIntent({
+      ...graphExecutionInput(claim, 'retry scoped stop'),
+      onReady: () => ready.release(),
+    });
+    await ready.promise;
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    const identity = {
+      sessionId: child.id,
+      runId: claim.targetRunId,
+      turnId: claim.targetTurnId,
+    };
+    await assert.rejects(
+      manager.stopAgentGraphActivation(identity),
+      /scoped abort persistence unavailable/,
+    );
+    rejectAbort = false;
+    await manager.stopAgentGraphActivation(identity);
+    sendGate.release();
+    assert.strictEqual((await running).status, 'cancelled');
+    assert.strictEqual(
+      runtimeInvocationOutcome(await readInvocation(runStore, child.id, claim.targetRunId)),
+      'cancelled',
+    );
+    assert.strictEqual(
+      (await runStore.readRuntimeEvents(child.id, claim.targetRunId)).filter(
+        (event) => event.status === 'aborted',
+      ).length,
+      1,
+    );
+  });
+
+  test('a scoped graph stop whose backend delivery fails keeps the child result authoritative', {
+    timeout: 2_000,
+  }, async () => {
+    const store = new MemorySessionStore();
+    const runStore = new MemoryAgentRunStore();
+    const backends = new BackendRegistry();
+    const sendGate = makeGate();
+    const ready = makeGate();
+    backends.register(
+      'ai-sdk',
+      (ctx) =>
+        new (class extends TestBackend {
+          override async stop(): Promise<void> {
+            this.stopCalls += 1;
+            sendGate.release();
+            throw new Error('backend stop delivery failed');
+          }
+        })(ctx, sendGate),
+    );
+    const manager = new SessionManager({
+      store,
+      runStore,
+      runtimeEventStore: runStore,
+      backends,
+      childTools: [testTool('Read'), testTool('Glob'), testTool('Grep')],
+      newId: nextId(),
+      now: nextNow(85),
+    });
+    const parent = await manager.createSession(makeInput());
+    const child = await createGraphOperatorSession(store, parent.id);
+    const claim = graphIntentClaim({ targetSessionId: child.id }, 'failed stop delivery');
+    const running = manager.runClaimedAgentGraphIntent({
+      ...graphExecutionInput(claim, 'failed stop delivery'),
+      onReady: () => ready.release(),
+    });
+    void running.catch(() => undefined);
+    await ready.promise;
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    const identity = {
+      sessionId: child.id,
+      runId: claim.targetRunId,
+      turnId: claim.targetTurnId,
+    };
+    // The stopper owns the cleanup failure; a retry settles the retained
+    // operation as failed without delivering to the backend again.
+    await assert.rejects(manager.stopAgentGraphActivation(identity), /stop delivery failed/);
+    await assert.rejects(manager.stopAgentGraphActivation(identity), /stop delivery failed/);
+    const result = await running;
+    assert.strictEqual(result.status, 'cancelled');
+    assert.strictEqual(
+      runtimeInvocationOutcome(await readInvocation(runStore, child.id, claim.targetRunId)),
+      'cancelled',
+    );
+  });
+
+  for (const recovery of ['retry', 'session_stop'] as const) {
+    test(`scoped graph stop cleanup failure keeps the Session queue fenced until ${recovery}`, {
+      timeout: 2_000,
+    }, async () => {
+      const store = new MemorySessionStore();
+      let rejectAbort = true;
+      const runStore = new MemoryAgentRunStore({
+        beforeRuntimeEventAppend: (_sessionId, _runId, event) => {
+          if (rejectAbort && event.status === 'aborted')
+            throw new Error('scoped cleanup unavailable');
+        },
+      });
+      const backends = new BackendRegistry();
+      const sendGate = makeGate();
+      const ready = makeGate();
+      backends.register('ai-sdk', (ctx) => new CountingStopBackend(ctx, sendGate));
+      const manager = new SessionManager({
+        store,
+        runStore,
+        runtimeEventStore: runStore,
+        backends,
+        childTools: [testTool('Read'), testTool('Glob'), testTool('Grep')],
+        newId: nextId(),
+        now: nextNow(90),
+      });
+      const parent = await manager.createSession(makeInput());
+      const child = await createGraphOperatorSession(store, parent.id);
+      const firstClaim = graphIntentClaim({ targetSessionId: child.id }, 'stop before successor');
+      const secondClaim = graphIntentClaim(
+        {
+          targetSessionId: child.id,
+          claimId: `graph_claim_${'c'.repeat(32)}`,
+          intentId: `graph_intent_${'d'.repeat(32)}`,
+          targetRunId: 'successor-run',
+          targetTurnId: 'successor-turn',
+        },
+        'successor',
+      );
+      const first = manager.runClaimedAgentGraphIntent({
+        ...graphExecutionInput(firstClaim, 'stop before successor'),
+        onReady: () => ready.release(),
+      });
+      await ready.promise;
+      const second = manager.runClaimedAgentGraphIntent(
+        graphExecutionInput(secondClaim, 'successor'),
+      );
+      const settled = Promise.allSettled([first, second]);
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      const identity = {
+        sessionId: child.id,
+        runId: firstClaim.targetRunId,
+        turnId: firstClaim.targetTurnId,
+      };
+      await assert.rejects(
+        manager.stopAgentGraphActivation(identity),
+        /scoped cleanup unavailable/,
+      );
+      sendGate.release();
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      assert.deepStrictEqual(
+        (await runStore.listSessionInvocations(child.id)).map((run) => run.turnId),
+        [firstClaim.targetTurnId],
+      );
+      rejectAbort = false;
+      if (recovery === 'retry') await manager.stopAgentGraphActivation(identity);
+      else await manager.stopSession(child.id);
+      const outcomes = await settled;
+      assert.strictEqual(
+        runtimeInvocationOutcome(await readInvocation(runStore, child.id, firstClaim.targetRunId)),
+        'cancelled',
+      );
+      assert.strictEqual(outcomes[1]!.status, recovery === 'retry' ? 'fulfilled' : 'rejected');
+      if (outcomes[1]!.status === 'fulfilled')
+        assert.strictEqual(outcomes[1]!.value.status, 'completed');
+    });
+  }
+
+  test('a retained graph stop quarantines an unrelated Turn until its cleanup succeeds', {
+    timeout: 2_000,
+  }, async () => {
+    const store = new MemorySessionStore();
+    let rejectAbort = true;
+    const runStore = new MemoryAgentRunStore({
+      beforeRuntimeEventAppend: (_sessionId, _runId, event) => {
+        if (rejectAbort && event.status === 'aborted') throw new Error('stop cleanup unavailable');
+      },
+    });
+    const backends = new BackendRegistry();
+    const sendGate = makeGate();
+    const ready = makeGate();
+    backends.register('ai-sdk', (ctx) => new CountingStopBackend(ctx, sendGate));
+    const manager = new SessionManager({
+      store,
+      runStore,
+      runtimeEventStore: runStore,
+      backends,
+      childTools: [testTool('Read'), testTool('Glob'), testTool('Grep')],
+      newId: nextId(),
+      now: nextNow(97),
+    });
+    const parent = await manager.createSession(makeInput());
+    const child = await createGraphOperatorSession(store, parent.id);
+    const claim = graphIntentClaim({ targetSessionId: child.id }, 'retained stop');
+    // The stopped Run's own finalization reports the same persistence fault.
+    const running = manager
+      .runClaimedAgentGraphIntent({
+        ...graphExecutionInput(claim, 'retained stop'),
+        onReady: () => ready.release(),
+      })
+      .catch((error: unknown) => error);
+    await ready.promise;
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    const identity = {
+      sessionId: child.id,
+      runId: claim.targetRunId,
+      turnId: claim.targetTurnId,
+    };
+    await assert.rejects(manager.stopAgentGraphActivation(identity), /stop cleanup unavailable/);
+    sendGate.release();
+    assert.strictEqual(manager.hasPendingAgentGraphActivationStop(identity), true);
+    // A Turn that holds no stop intent of the retained operation cannot use
+    // the stop owner's quarantine bypass.
+    await assert.rejects(
+      drain(manager.sendMessage(child.id, { turnId: 'unrelated-turn', text: 'must wait' })),
+      /quarantined by a retained stop operation/,
+    );
+    rejectAbort = false;
+    await manager.stopAgentGraphActivation(identity);
+    assert.strictEqual(manager.hasPendingAgentGraphActivationStop(identity), false);
+    await running;
+    await drain(manager.sendMessage(child.id, { turnId: 'after-cleanup', text: 'runs now' }));
+    const runs = await runStore.listSessionInvocations(child.id);
+    assert.strictEqual(
+      runtimeInvocationOutcome(runs.find((run) => run.turnId === claim.targetTurnId)!),
+      'cancelled',
+    );
+    assert.strictEqual(
+      runtimeInvocationOutcome(runs.find((run) => run.turnId === 'after-cleanup')!),
+      'completed',
+    );
+  });
+
+  test('scoped graph stop for a completed identity leaves its active successor intact', {
+    timeout: 2_000,
+  }, async () => {
+    const successorGate = makeGate();
+    const { manager, child, runStore, activeGate, first, claims } = await createQueuedGraphScenario(
+      undefined,
+      successorGate,
+    );
+    const [firstClaim, secondClaim] = claims;
+    activeGate.release();
+    assert.strictEqual((await first).status, 'completed');
+    const ready = makeGate();
+    const second = manager.runClaimedAgentGraphIntent({
+      ...graphExecutionInput(secondClaim, 'queued activation'),
+      onReady: () => ready.release(),
+    });
+    await ready.promise;
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    try {
+      await manager.stopAgentGraphActivation(
+        {
+          sessionId: child.id,
+          runId: firstClaim.targetRunId,
+          turnId: firstClaim.targetTurnId,
+        },
+        { source: 'graph_supervisor' },
+      );
+      assert.strictEqual(
+        (await readInvocation(runStore, child.id, secondClaim.targetRunId)).terminalEvent,
+        undefined,
+      );
+    } finally {
+      successorGate.release();
+    }
+    assert.strictEqual((await second).status, 'completed');
+    assert.deepStrictEqual(
+      (await runStore.listSessionInvocations(child.id)).map((run) => [
+        run.turnId,
+        runtimeInvocationOutcome(run),
+      ]),
+      [
+        [firstClaim.targetTurnId, 'completed'],
+        [secondClaim.targetTurnId, 'completed'],
+      ],
+    );
+  });
+
+  for (const source of ['graph_supervisor', 'stop_button'] as const) {
+    test(`a graph operator stopped by a ${source} Session stop ${source === 'graph_supervisor' ? 'accepts' : 'refuses'} its next claim`, {
+      timeout: 2_000,
+    }, async () => {
+      const store = new MemorySessionStore();
+      const runStore = new MemoryAgentRunStore();
+      const backends = new BackendRegistry();
+      const sendGate = makeGate();
+      const ready = makeGate();
+      backends.register('ai-sdk', (ctx) => new TestBackend(ctx, sendGate));
+      const manager = new SessionManager({
+        store,
+        runStore,
+        runtimeEventStore: runStore,
+        backends,
+        childTools: [testTool('Read'), testTool('Glob'), testTool('Grep')],
+        newId: nextId(),
+        now: nextNow(95),
+      });
+      const parent = await manager.createSession(makeInput());
+      const child = await createGraphOperatorSession(store, parent.id);
+      const stoppedClaim = graphIntentClaim({ targetSessionId: child.id }, 'stopped activation');
+      const stopped = manager.runClaimedAgentGraphIntent({
+        ...graphExecutionInput(stoppedClaim, 'stopped activation'),
+        onReady: () => ready.release(),
+      });
+      await ready.promise;
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      await manager.stopSession(child.id, { source });
+      sendGate.release();
+      assert.strictEqual((await stopped).status, 'cancelled');
+      assert.strictEqual((await store.readHeader(child.id)).status, 'aborted');
+      // A whole-graph stop keeps durable schedule facts so the supervisor can
+      // wake the same graph again; a user stop still terminates the operator.
+      const nextClaim = graphIntentClaim(
+        {
+          targetSessionId: child.id,
+          claimId: `graph_claim_${'e'.repeat(32)}`,
+          intentId: `graph_intent_${'f'.repeat(32)}`,
+          targetRunId: 'resumed-run',
+          targetTurnId: 'resumed-turn',
+        },
+        'resumed activation',
+      );
+      const next = manager.runClaimedAgentGraphIntent(
+        graphExecutionInput(nextClaim, 'resumed activation'),
+      );
+      if (source === 'graph_supervisor') assert.strictEqual((await next).status, 'completed');
+      else await assert.rejects(next, /target child session is terminated/);
+    });
+  }
+
   test('evaluates execution admission only after a claimed child-session slot is available', async () => {
     const store = new MemorySessionStore();
     const runStore = new MemoryAgentRunStore();
@@ -2142,69 +2564,78 @@ describe('SessionManager claimed graph intent execution', () => {
     assert.strictEqual((await runStore.listSessionInvocations(child.id)).length, 1);
   });
 
-  test('keeps a stop pending across graph admission with an idle cached backend', async () => {
-    const store = new MemorySessionStore();
-    const runStore = new MemoryAgentRunStore();
-    const backends = new BackendRegistry();
-    const admissionStarted = makeGate();
-    const releaseAdmission = makeGate();
-    let backend: TestBackend | undefined;
-    backends.register('ai-sdk', (ctx) => {
-      backend = new TestBackend(ctx);
-      return backend;
-    });
-    const manager = new SessionManager({
-      store,
-      runStore,
-      runtimeEventStore: runStore,
-      backends,
-      childTools: [testTool('Read'), testTool('Glob'), testTool('Grep')],
-      newId: nextId(),
-      now: nextNow(77),
-    });
-    const parent = await manager.createSession(makeInput());
-    const child = await createGraphOperatorSession(store, parent.id);
-    await drain(
-      manager.sendMessage(child.id, {
-        turnId: 'completed-before-admission',
-        text: 'warm the cached operator backend',
-      }),
-    );
-    assert.strictEqual(backend?.sendInputs?.length, 1);
-    const claim = graphIntentClaim(
-      { targetSessionId: child.id },
-      'activation stopped during admission',
-    );
-    const execution = manager.runClaimedAgentGraphIntent({
-      ...graphExecutionInput(claim, 'activation stopped during admission'),
-      async admitExecution() {
-        admissionStarted.release();
-        await releaseAdmission.promise;
-        return 'executing';
-      },
-    });
-    await admissionStarted.promise;
+  for (const scoped of [false, true])
+    test(`keeps a ${scoped ? 'scoped graph stop' : 'session stop'} pending across graph admission with an idle cached backend`, async () => {
+      const store = new MemorySessionStore();
+      const runStore = new MemoryAgentRunStore();
+      const backends = new BackendRegistry();
+      const admissionStarted = makeGate();
+      const releaseAdmission = makeGate();
+      let backend: TestBackend | undefined;
+      backends.register('ai-sdk', (ctx) => {
+        backend = new TestBackend(ctx);
+        return backend;
+      });
+      const manager = new SessionManager({
+        store,
+        runStore,
+        runtimeEventStore: runStore,
+        backends,
+        childTools: [testTool('Read'), testTool('Glob'), testTool('Grep')],
+        newId: nextId(),
+        now: nextNow(77),
+      });
+      const parent = await manager.createSession(makeInput());
+      const child = await createGraphOperatorSession(store, parent.id);
+      await drain(
+        manager.sendMessage(child.id, {
+          turnId: 'completed-before-admission',
+          text: 'warm the cached operator backend',
+        }),
+      );
+      assert.strictEqual(backend?.sendInputs?.length, 1);
+      const claim = graphIntentClaim(
+        { targetSessionId: child.id },
+        'activation stopped during admission',
+      );
+      const execution = manager.runClaimedAgentGraphIntent({
+        ...graphExecutionInput(claim, 'activation stopped during admission'),
+        async admitExecution() {
+          admissionStarted.release();
+          await releaseAdmission.promise;
+          return 'executing';
+        },
+      });
+      await admissionStarted.promise;
 
-    let stopSettled = false;
-    const stop = manager.stopSession(child.id, { source: 'graph_supervisor' }).finally(() => {
-      stopSettled = true;
-    });
-    await Promise.resolve();
-    assert.strictEqual(stopSettled, false);
-    assert.strictEqual(backend?.sendInputs?.length, 1);
+      let stopSettled = false;
+      const stop = (
+        scoped
+          ? manager.stopAgentGraphActivation({
+              sessionId: child.id,
+              runId: claim.targetRunId,
+              turnId: claim.targetTurnId,
+            })
+          : manager.stopSession(child.id, { source: 'graph_supervisor' })
+      ).finally(() => {
+        stopSettled = true;
+      });
+      await Promise.resolve();
+      assert.strictEqual(stopSettled, false);
+      assert.strictEqual(backend?.sendInputs?.length, 1);
 
-    releaseAdmission.release();
-    await stop;
-    const result = await execution;
+      releaseAdmission.release();
+      await stop;
+      const result = await execution;
 
-    assert.strictEqual(result.status, 'cancelled');
-    assert.strictEqual(backend?.stopCalls, 1);
-    assert.strictEqual(backend?.sendInputs?.length, 1);
-    assert.strictEqual(
-      runtimeInvocationOutcome(await readInvocation(runStore, child.id, claim.targetRunId)),
-      'cancelled',
-    );
-  });
+      assert.strictEqual(result.status, 'cancelled');
+      assert.strictEqual(backend?.stopCalls, 1);
+      assert.strictEqual(backend?.sendInputs?.length, 1);
+      assert.strictEqual(
+        runtimeInvocationOutcome(await readInvocation(runStore, child.id, claim.targetRunId)),
+        'cancelled',
+      );
+    });
 
   test('runtime stop settles queued graph claims without letting their slots pass the active claim', async () => {
     const firstAbort = new AbortController();
@@ -3633,6 +4064,118 @@ describe('SessionManager child-session runtime primitive', () => {
     assert.strictEqual(childTwoResult.status, 'cancelled');
   });
 
+  test('scoped Hosted stop includes earlier runs in its turn and preserves other turn children', async () => {
+    const store = new MemorySessionStore();
+    const runStore = new MemoryAgentRunStore();
+    const backends = new BackendRegistry();
+    const parentGate = makeGate();
+    const childGates = [makeGate(), makeGate()];
+    let childGateIndex = 0;
+    const backendsBySession = new Map<string, TestBackend>();
+    backends.register('ai-sdk', (ctx) => {
+      const gate = ctx.header.subagentRuntime ? childGates[childGateIndex++] : parentGate;
+      const backend = new TestBackend(ctx, gate);
+      backendsBySession.set(ctx.sessionId, backend);
+      return backend;
+    });
+    const manager = new SessionManager({
+      store,
+      runStore,
+      runtimeEventStore: runStore,
+      isParentRunActive: (_sessionId, runId) =>
+        ['earlier-physical-run', 'other-parent-run'].includes(runId),
+      backends,
+      childTools: [testTool('Read'), testTool('Glob'), testTool('Grep')],
+      newId: nextId(),
+      now: nextNow(300),
+    });
+    const parent = await manager.createSession(makeInput());
+    const parentTurn = manager
+      .sendMessage(parent.id, { turnId: 'parent-turn', text: 'coordinate children' })
+      [Symbol.asyncIterator]();
+    await parentTurn.next();
+    const [parentRun] = await runStore.listSessionInvocations(parent.id);
+    if (!parentRun) throw new Error('parent run was not recorded');
+
+    await seedInvocationFromHeader(
+      runStore,
+      makeRunHeader({
+        sessionId: parent.id,
+        runId: 'earlier-physical-run',
+        status: 'running',
+        turnId: parentRun.turnId,
+      }),
+    );
+    await seedInvocationFromHeader(
+      runStore,
+      makeRunHeader({
+        sessionId: parent.id,
+        runId: 'other-parent-run',
+        status: 'running',
+        turnId: 'other-logical-turn',
+      }),
+    );
+    const childOneStarted = makeGate();
+    let childOneId = '';
+    const childOne = manager.spawnChildSession(parent.id, {
+      name: 'Child one',
+      spawnedBy: {
+        parentRunId: 'earlier-physical-run',
+        parentTurnId: parentRun.turnId,
+        toolCallId: 'tool-call-1',
+      },
+      agentProfile: LOCAL_READ_AGENT_PROFILE,
+      prompt: 'first child',
+      onReady: ({ childSessionId }) => {
+        childOneId = childSessionId;
+      },
+      onEvent: (event) => {
+        if (event.type === 'text_delta') childOneStarted.release();
+      },
+    });
+    await childOneStarted.promise;
+
+    const childTwoStarted = makeGate();
+    let childTwoId = '';
+    const childTwo = manager.spawnChildSession(parent.id, {
+      name: 'Child two',
+      spawnedBy: {
+        parentRunId: 'other-parent-run',
+        parentTurnId: 'other-logical-turn',
+        toolCallId: 'tool-call-2',
+      },
+      agentProfile: LOCAL_READ_AGENT_PROFILE,
+      prompt: 'second child',
+      onReady: ({ childSessionId }) => {
+        childTwoId = childSessionId;
+      },
+      onEvent: (event) => {
+        if (event.type === 'text_delta') childTwoStarted.release();
+      },
+    });
+    await childTwoStarted.promise;
+
+    await manager.deliverHostedRootStop(
+      parent.id,
+      { source: 'graph_supervisor' },
+      {
+        sessionId: parent.id,
+        runId: parentRun.runId,
+        turnId: parentRun.turnId,
+      },
+    );
+    assert.strictEqual(backendsBySession.get(parent.id)?.stopCalls, 1);
+    assert.strictEqual(backendsBySession.get(childOneId)?.stopCalls, 1);
+    assert.strictEqual(backendsBySession.get(childTwoId)?.stopCalls, 0);
+
+    parentGate.release();
+    for (const gate of childGates) gate.release();
+    while (!(await parentTurn.next()).done) {}
+    const [childOneResult, childTwoResult] = await Promise.all([childOne, childTwo]);
+    assert.strictEqual(childOneResult.status, 'cancelled');
+    assert.strictEqual(childTwoResult.status, 'completed');
+  });
+
   for (const { name, stopOptions, stop } of [
     {
       name: 'observes a rejected hosted stop while child lookup is pending',
@@ -14793,7 +15336,12 @@ class MemoryAgentRunStore
     sessionId: string,
     turnId: string,
   ): Promise<
-    { runId: string; userMessageId: string | null; execution: RootExecutionDescriptor } | undefined
+    | {
+        runId: string;
+        userMessageId: string | null;
+        execution: RootExecutionDescriptor;
+      }
+    | undefined
   > {
     return this.rootTurnAdmissions.get(key(sessionId, turnId));
   }
@@ -15361,6 +15909,7 @@ function hostedRootAuthority(): RuntimeHostedRootAuthority {
       }
     },
     stopRoot: async () => {},
+    stopRootRun: async () => {},
     stopSession: async () => {},
   };
 }
@@ -15501,7 +16050,7 @@ function graphRunnableIntentForClaim(claim: AgentGraphIntentClaim): AgentGraphRu
   };
 }
 
-async function createQueuedGraphScenario(firstAbortSignal?: AbortSignal) {
+async function createQueuedGraphScenario(firstAbortSignal?: AbortSignal, successorGate?: Gate) {
   const store = new MemorySessionStore();
   const runStore = new MemoryAgentRunStore();
   const backends = new BackendRegistry();
@@ -15511,6 +16060,10 @@ async function createQueuedGraphScenario(firstAbortSignal?: AbortSignal) {
   let backend!: TestBackend;
   backends.register('ai-sdk', (ctx) => {
     backend = new (class extends TestBackend {
+      override async *send(input: BackendSendInput): AsyncIterable<SessionEvent> {
+        if (input.turnId === 'graph-turn-2') await successorGate?.promise;
+        yield* super.send(input);
+      }
       override async stop(
         reason: 'user_stop' | 'redirect',
         mode: BackendStopMode = 'immediate',
diff --git a/packages/runtime/src/__tests__/stream-graph-coordinator.test.ts b/packages/runtime/src/__tests__/stream-graph-coordinator.test.ts
index 8b68ed5dd6..fc9215e4c4 100644
--- a/packages/runtime/src/__tests__/stream-graph-coordinator.test.ts
+++ b/packages/runtime/src/__tests__/stream-graph-coordinator.test.ts
@@ -213,6 +213,7 @@ describe('host-managed agent graph coordinator', () => {
         runClaimedAgentGraphIntent: async () => {
           throw new Error('test does not dispatch new work');
         },
+        stopAgentGraphActivation: async () => {},
         stopSession: async () => {},
       },
       newId: randomUUID,
@@ -1166,6 +1167,7 @@ describe('host-managed agent graph coordinator', () => {
         runClaimedAgentGraphIntent: async () => {
           throw new Error('unexpected operator dispatch');
         },
+        stopAgentGraphActivation: async () => {},
         stopSession: async () => {},
       },
       newId: randomUUID,
@@ -1252,6 +1254,7 @@ describe('host-managed agent graph coordinator', () => {
         runClaimedAgentGraphIntent: async () => {
           throw new Error('unexpected operator dispatch');
         },
+        stopAgentGraphActivation: async () => {},
         stopSession: async () => {},
       },
       newId: randomUUID,
@@ -1324,6 +1327,7 @@ describe('host-managed agent graph coordinator', () => {
         runClaimedAgentGraphIntent: async () => {
           throw new Error('unexpected operator dispatch');
         },
+        stopAgentGraphActivation: async () => {},
         stopSession: async () => {},
       },
       newId: randomUUID,
@@ -1374,6 +1378,7 @@ describe('host-managed agent graph coordinator', () => {
         runClaimedAgentGraphIntent: async () => {
           throw new Error('epoch cutover cannot dispatch operators');
         },
+        stopAgentGraphActivation: async () => {},
         stopSession: async () => {
           stopStarted.resolve();
           await releaseStop.promise;
@@ -1490,6 +1495,7 @@ describe('host-managed agent graph coordinator', () => {
         runClaimedAgentGraphIntent: async () => {
           throw new Error('epoch cutover cannot dispatch operators');
         },
+        stopAgentGraphActivation: async () => {},
         stopSession: async () => {},
       },
       newId: randomUUID,
@@ -1578,6 +1584,7 @@ describe('host-managed agent graph coordinator', () => {
           throw new Error('preset provider unavailable');
         },
         runClaimedAgentGraphIntent: manager.runClaimedAgentGraphIntent.bind(manager),
+        stopAgentGraphActivation: manager.stopAgentGraphActivation.bind(manager),
         stopSession: manager.stopSession.bind(manager),
       };
       const create = () =>
@@ -1956,6 +1963,7 @@ describe('host-managed agent graph coordinator', () => {
           runClaimedAgentGraphIntent: async () => {
             throw new Error('lifecycle reads cannot dispatch operators');
           },
+          stopAgentGraphActivation: async () => {},
           stopSession: async () => {},
         },
         newId: randomUUID,
@@ -2321,6 +2329,7 @@ describe('host-managed agent graph coordinator', () => {
         ],
       },
       runtime: {
+        stopAgentGraphActivation: async () => {},
         stopSession: async () => {
           started.resolve();
           await release.promise;
@@ -2429,6 +2438,7 @@ describe('host-managed agent graph coordinator', () => {
           })),
       },
       runtime: {
+        stopAgentGraphActivation: async () => {},
         stopSession: async (sessionId: string) => {
           stopped.push(sessionId);
           if (stopped.length === 2) markStopsStarted();
@@ -2507,6 +2517,7 @@ describe('host-managed agent graph coordinator', () => {
         runClaimedAgentGraphIntent: async () => {
           throw new Error('missing input must prevent runtime dispatch');
         },
+        stopAgentGraphActivation: async () => {},
         stopSession: async () => {},
       },
       newId: randomUUID,
diff --git a/packages/runtime/src/__tests__/stream-graph-schedule-reconcile.test.ts b/packages/runtime/src/__tests__/stream-graph-schedule-reconcile.test.ts
index 311ccb5892..7153d46626 100644
--- a/packages/runtime/src/__tests__/stream-graph-schedule-reconcile.test.ts
+++ b/packages/runtime/src/__tests__/stream-graph-schedule-reconcile.test.ts
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-import { deferred, nextId } from '@maka/core/test-only/async-primitives';
+import { deferred, nextId, waitFor, withTimeout } from '@maka/core/test-only/async-primitives';
 import assert from 'node:assert/strict';
 import { describe, test } from 'node:test';
 import type {
@@ -38,6 +38,7 @@ import type { AgentGraphInputHandoff } from '../stream-graph-handoff.js';
 import {
   reconcileAgentGraphSchedule,
   type RenderAgentGraphScheduledWorkPromptInput,
+  type AgentGraphScheduleStopController,
 } from '../stream-graph-schedule-reconcile.js';
 import {
   compileAgentGraphScheduleUpdate,
@@ -561,7 +562,12 @@ describe('stream graph schedule reconciliation', () => {
 
       assert.equal(stopped.status, 'reconciled');
       assert.deepEqual(stopController.calls, [
-        { sessionId: 'session-writer', source: 'graph_supervisor' },
+        {
+          sessionId: 'session-writer',
+          runId: first.dispatches[0]!.claim.targetRunId,
+          turnId: first.dispatches[0]!.claim.targetTurnId,
+          source: 'graph_supervisor',
+        },
       ]);
       assert.deepEqual(
         stopped.stops.map((result) => [result.targetId, result.status, result.activationId]),
@@ -574,6 +580,61 @@ describe('stream graph schedule reconciliation', () => {
     }
   });
 
+  test('a historical terminal stop target neither reaches the runtime nor blocks new work', async () => {
+    const store = createSqliteSessionMetadataStore(':memory:', { now: nextNumber(250) });
+    const observation = new MemoryGraphObservation();
+    const executor = new MemoryScheduleExecutor(store, observation, 'running');
+    const stopController = new MemoryStopController(observation);
+    try {
+      const added = await commitSchedule(store, 'tool-add', {
+        add_work: [{ operator_id: 'writer', instruction: 'Draft until stopped.', input_ids: [] }],
+      });
+      const stoppedWorkId = added.addWork[0]!.workId;
+      const newId = nextId();
+      const reconcile = (controller: AgentGraphScheduleStopController) =>
+        reconcileAgentGraphSchedule({
+          topology: topology(),
+          controlStore: store,
+          executor,
+          stopController: controller,
+          newId,
+          maxNewActivations: 1,
+          observeGraph: () => observation.read(),
+          renderPrompt: ({ work }) => work.instruction,
+        });
+      await reconcile(stopController);
+      await commitSchedule(store, 'tool-stop', {
+        stop: [{ target_id: stoppedWorkId, reason: 'The draft is obsolete.' }],
+      });
+      assert.equal((await reconcile(stopController)).stops[0]?.status, 'stopped');
+      assert.equal(stopController.calls.length, 1);
+
+      const next = await commitSchedule(store, 'tool-add-next', {
+        add_work: [{ operator_id: 'writer', instruction: 'Draft the replacement.', input_ids: [] }],
+      });
+      let runtimeStops = 0;
+      const result = await reconcile({
+        async stopAgentGraphActivation() {
+          runtimeStops += 1;
+          throw new Error('a settled historical stop must not reach the runtime again');
+        },
+      });
+
+      assert.equal(runtimeStops, 0);
+      assert.equal(result.status, 'reconciled');
+      assert.deepEqual(
+        result.stops.map((stop) => [stop.targetId, stop.status]),
+        [[stoppedWorkId, 'already_terminal']],
+      );
+      assert.deepEqual(
+        result.dispatches.map((dispatch) => dispatch.intent.readinessId),
+        [next.addWork[0]!.workId],
+      );
+    } finally {
+      store.close();
+    }
+  });
+
   test('does not admit stale work when a stop wins the SQLite revision race', async () => {
     const store = createSqliteSessionMetadataStore(':memory:', { now: nextNumber(300) });
     const observation = new MemoryGraphObservation();
@@ -843,6 +904,405 @@ describe('stream graph schedule reconciliation', () => {
       store.close();
     }
   });
+
+  test('retains a schedule commit that arrives before the next wave wait is installed', async () => {
+    const wave = await gatedScheduleWave(2);
+    const enteredStop = deferred();
+    const releaseStop = deferred();
+    let first = true;
+    wave.beforeStop = async () => {
+      if (!first) return;
+      first = false;
+      enteredStop.resolve();
+      await releaseStop.promise;
+    };
+    try {
+      await wave.commit('stop-first', {
+        stop: [{ target_id: wave.workIds[0]!, reason: 'First target.' }],
+      });
+      await withTimeout(enteredStop.promise, 1000, 'first stop must start');
+      // The reconciler is still awaiting the first stop, not waiting for a wake.
+      await wave.commit('stop-second', {
+        stop: [{ target_id: wave.workIds[1]!, reason: 'Second target.' }],
+      });
+      releaseStop.resolve();
+      const result = await withTimeout(
+        wave.reconciliation,
+        1000,
+        'the retained wake must stop the second child',
+      );
+      assert.equal(result.dispatches.length, 2);
+      assert.equal(wave.executor.backendInvocations, 2);
+      assert.deepEqual(
+        result.dispatches.map((dispatch) => dispatch.result.status),
+        ['cancelled', 'cancelled'],
+      );
+      assert.equal((await wave.store.listAgentGraphIntentClaims(GRAPH_ID)).length, 2);
+      assert.equal(wave.subscriptions, 0);
+    } finally {
+      releaseStop.resolve();
+      await wave.close();
+    }
+  });
+
+  test('more than eight schedule wakes preserve the running wave and its single dispatch', async () => {
+    const wave = await gatedScheduleWave(1);
+    try {
+      for (let index = 0; index < 12; index += 1) {
+        const observed = wave.observations;
+        await wave.commit(`wake-${index}`, {
+          stop: [{ target_id: `unknown-target-${index}`, reason: 'An unrelated obsolete target.' }],
+        });
+        await waitFor(() => wave.observations > observed, { timeoutMs: 1000 });
+        assert.equal(wave.settled, false, 'a normal wake cannot exhaust reconciliation retries');
+        assert.equal(wave.executor.backendInvocations, 1);
+      }
+      await wave.commit('stop-running-work', {
+        stop: [{ target_id: wave.workIds[0]!, reason: 'Now stop the running work.' }],
+      });
+      const result = await withTimeout(
+        wave.reconciliation,
+        1000,
+        'the original wave must remain stoppable',
+      );
+      assert.equal(result.status, 'reconciled');
+      assert.equal(result.newActivationCount, 1);
+      assert.equal(result.dispatches.length, 1);
+      assert.equal(result.dispatches[0]!.result.status, 'cancelled');
+      assert.equal((await wave.store.listAgentGraphIntentClaims(GRAPH_ID)).length, 1);
+      assert.equal(wave.executor.backendInvocations, 1);
+      assert.equal(wave.subscriptions, 0);
+    } finally {
+      await wave.close();
+    }
+  });
+
+  test('reports a stop failure while the wave is live and retries after a later revision', async () => {
+    const wave = await gatedScheduleWave(1);
+    const failureObserved = deferred();
+    wave.beforeStop = async () => {
+      throw new Error('temporary exact-stop failure');
+    };
+    wave.onFailure = () => failureObserved.resolve();
+    try {
+      await wave.commit('failing-stop', {
+        stop: [{ target_id: wave.workIds[0]!, reason: 'First stop attempt.' }],
+      });
+      await withTimeout(
+        failureObserved.promise,
+        1000,
+        'stop failure must be visible before child completion',
+      );
+      assert.equal(wave.settled, false);
+      assert.equal(wave.failures[0]?.phase, 'stop');
+      assert.equal(wave.executor.backendInvocations, 1);
+      wave.beforeStop = undefined;
+      await wave.commit('retry-stop', {
+        stop: [{ target_id: wave.workIds[0]!, reason: 'Retry after the stop service recovered.' }],
+      });
+      const result = await withTimeout(
+        wave.reconciliation,
+        1000,
+        'later revision must retry the exact stop',
+      );
+      assert.equal(result.dispatches.length, 1);
+      assert.equal(result.dispatches[0]!.result.status, 'cancelled');
+      assert.equal(wave.failures.length, 1, 'the observer must retain the failure notification');
+      assert.equal(result.status, 'reconciled');
+      assert.equal(result.failures.length, 0, 'successful stop retry clears the obsolete failure');
+      const recovered = await wave.reconcileAgain();
+      assert.equal(recovered.status, 'reconciled');
+      assert.equal(recovered.failures.length, 0);
+      assert.equal(recovered.dispatches.length, 0);
+      assert.equal(wave.executor.backendInvocations, 1);
+    } finally {
+      await wave.close();
+    }
+  });
+
+  test('retries exact stop cleanup after the activation already has a terminal observation', async () => {
+    const wave = await gatedScheduleWave(1);
+    const failureObserved = deferred();
+    let stopAttempts = 0;
+    wave.beforeStop = async () => {
+      stopAttempts += 1;
+      if (stopAttempts === 1) {
+        wave.markTerminalWithoutCompleting();
+        throw new Error('terminal committed but exact-stop cleanup failed');
+      }
+    };
+    wave.onFailure = () => failureObserved.resolve();
+    try {
+      await wave.commit('terminal-cleanup-failure', {
+        stop: [{ target_id: wave.workIds[0]!, reason: 'Stop and settle this activation.' }],
+      });
+      await withTimeout(
+        failureObserved.promise,
+        1000,
+        'cleanup failure must be reported while the terminal activation still owns its wave',
+      );
+      assert.equal(wave.settled, false);
+      assert.equal(stopAttempts, 1);
+      assert.equal(wave.failures[0]?.phase, 'stop');
+      await wave.commit('retry-terminal-cleanup', {
+        stop: [
+          {
+            target_id: wave.workIds[0]!,
+            reason: 'Retry cleanup for the same terminal activation.',
+          },
+        ],
+      });
+      const result = await withTimeout(
+        wave.reconciliation,
+        1000,
+        'terminal observation must not bypass the exact-stop cleanup retry',
+      );
+      assert.ok(stopAttempts >= 2);
+      assert.equal(result.status, 'reconciled');
+      assert.equal(result.failures.length, 0);
+      assert.equal(result.dispatches.length, 1);
+      assert.equal(result.dispatches[0]!.result.status, 'cancelled');
+      assert.equal(result.newActivationCount, 1);
+      assert.equal((await wave.store.listAgentGraphIntentClaims(GRAPH_ID)).length, 1);
+      assert.equal(wave.executor.backendInvocations, 1);
+      assert.equal(wave.subscriptions, 0);
+    } finally {
+      await wave.close();
+    }
+  });
+
+  test('a transient control read failure leaves the wave stoppable on a later schedule wake', async () => {
+    const wave = await gatedScheduleWave(1);
+    const failureObserved = deferred();
+    wave.onFailure = () => failureObserved.resolve();
+    try {
+      wave.failNextControlRead(new Error('temporary schedule read failure'));
+      await wave.commit('read-failure-stop', {
+        stop: [{ target_id: wave.workIds[0]!, reason: 'Stop during a transient read failure.' }],
+      });
+      await withTimeout(
+        failureObserved.promise,
+        1000,
+        'control read failure must be reported while the child remains gated',
+      );
+      assert.equal(wave.settled, false);
+      assert.equal(wave.failures.length, 1);
+      assert.match(String(wave.failures[0]!.error), /temporary schedule read failure/);
+      assert.equal(wave.executor.backendInvocations, 1);
+      await wave.commit('read-recovered-stop', {
+        stop: [{ target_id: wave.workIds[0]!, reason: 'Retry after schedule reads recover.' }],
+      });
+      const result = await withTimeout(
+        wave.reconciliation,
+        1000,
+        'a recovered control read must cancel the original live wave',
+      );
+      assert.equal(result.status, 'reconciled');
+      assert.equal(result.failures.length, 0);
+      assert.equal(result.dispatches.length, 1);
+      assert.equal(result.dispatches[0]!.result.status, 'cancelled');
+      assert.equal(result.newActivationCount, 1);
+      assert.equal((await wave.store.listAgentGraphIntentClaims(GRAPH_ID)).length, 1);
+      assert.equal(wave.executor.backendInvocations, 1);
+      assert.equal(wave.subscriptions, 0);
+    } finally {
+      await wave.close();
+    }
+  });
+
+  test('mid-wave wakes do not stop an activation this reconciliation already stopped', async () => {
+    const wave = await gatedScheduleWave(2);
+    try {
+      await wave.commit('stop-first', {
+        stop: [{ target_id: wave.workIds[0]!, reason: 'Only the first branch is obsolete.' }],
+      });
+      await waitFor(() => wave.stopCalls === 1, { timeoutMs: 1000 });
+      for (let index = 0; index < 3; index += 1) {
+        const observed = wave.observations;
+        await wave.commit(`unrelated-wake-${index}`, {
+          stop: [{ target_id: `unknown-target-${index}`, reason: 'An unrelated target.' }],
+        });
+        await waitFor(() => wave.observations > observed, { timeoutMs: 1000 });
+      }
+      assert.equal(wave.stopCalls, 1, 'a settled stop target must not be stopped on every wake');
+      await wave.commit('stop-second', {
+        stop: [{ target_id: wave.workIds[1]!, reason: 'Now the second branch is obsolete.' }],
+      });
+      const result = await withTimeout(wave.reconciliation, 1000, 'second stop must settle');
+      assert.equal(wave.stopCalls, 2);
+      assert.equal(result.status, 'reconciled');
+      assert.deepEqual(
+        result.dispatches.map((dispatch) => dispatch.result.status),
+        ['cancelled', 'cancelled'],
+      );
+    } finally {
+      await wave.close();
+    }
+  });
+
+  test('retries a failed stop while its wave is parked without another schedule wake', async () => {
+    const wave = await gatedScheduleWave(1);
+    let attempts = 0;
+    wave.beforeStop = async () => {
+      attempts += 1;
+      if (attempts === 1) throw new Error('transient exact-stop failure');
+    };
+    try {
+      await wave.commit('stop-once', {
+        stop: [{ target_id: wave.workIds[0]!, reason: 'Stop despite one cleanup failure.' }],
+      });
+      const result = await withTimeout(
+        wave.reconciliation,
+        2000,
+        'a failed stop must be retried while it parks the only running wave',
+      );
+      assert.equal(attempts, 2);
+      assert.equal(result.status, 'reconciled');
+      assert.equal(result.failures.length, 0);
+      assert.equal(result.dispatches[0]!.result.status, 'cancelled');
+      assert.equal(wave.failures.length, 1);
+      assert.equal(wave.subscriptions, 0);
+    } finally {
+      await wave.close();
+    }
+  });
+
+  test('reports a persistently failing stop once as stuck and clears it when the stop succeeds', async (t) => {
+    const wave = await gatedScheduleWave(1);
+    let failing = true;
+    let attempts = 0;
+    wave.beforeStop = async () => {
+      attempts += 1;
+      if (failing) throw new Error(`exact-stop failure ${attempts}`);
+    };
+    // Real backoff reaches its 5 s cap before the eighth failure; drive it with
+    // mocked timers and real setImmediate turns for the in-memory store.
+    t.mock.timers.enable({ apis: ['setTimeout'] });
+    const flush = async () => {
+      for (let turn = 0; turn < 3; turn += 1) {
+        await new Promise<void>((resolve) => setImmediate(resolve));
+      }
+    };
+    const advanceToAttempt = async (target: number) => {
+      await flush();
+      while (attempts < target) {
+        t.mock.timers.tick(5_000);
+        await flush();
+      }
+    };
+    try {
+      await wave.commit('stuck-stop', {
+        stop: [{ target_id: wave.workIds[0]!, reason: 'Stop although cleanup keeps failing.' }],
+      });
+      await advanceToAttempt(7);
+      assert.equal(wave.failures.length, 1, 'retries before the threshold stay silent');
+      assert.doesNotMatch(String(wave.failures[0]!.error), /is stuck/);
+
+      await advanceToAttempt(8);
+      assert.equal(wave.failures.length, 2);
+      assert.equal(wave.failures[1]!.phase, 'stop');
+      assert.match(String(wave.failures[1]!.error), /is stuck after 8 consecutive failures/);
+
+      await advanceToAttempt(12);
+      assert.equal(wave.failures.length, 2, 'a stuck stop is reported once, not on every retry');
+      assert.equal(wave.settled, false, 'the stop is still retried, never dropped');
+
+      failing = false;
+      await advanceToAttempt(13);
+      await flush();
+      assert.equal(wave.settled, true);
+      const result = await wave.reconciliation;
+      assert.equal(result.status, 'reconciled');
+      assert.equal(result.failures.length, 0, 'a later success clears the stuck failure');
+      assert.equal(result.dispatches[0]!.result.status, 'cancelled');
+      assert.equal(wave.failures.length, 2);
+    } finally {
+      t.mock.timers.reset();
+      await wave.close();
+    }
+  });
+
+  test('reports a persistently failing control read once as stuck and clears it when reads recover', async (t) => {
+    const wave = await gatedScheduleWave(1);
+    t.mock.timers.enable({ apis: ['setTimeout'] });
+    const flush = async () => {
+      for (let turn = 0; turn < 3; turn += 1) {
+        await new Promise<void>((resolve) => setImmediate(resolve));
+      }
+    };
+    const base = wave.controlReads;
+    const advanceToRead = async (target: number) => {
+      await flush();
+      while (wave.controlReads - base < target) {
+        t.mock.timers.tick(5_000);
+        await flush();
+      }
+    };
+    try {
+      wave.failControlReads(new Error('schedule read failure'));
+      await wave.commit('stuck-read', {
+        stop: [
+          { target_id: wave.workIds[0]!, reason: 'Stop although control reads keep failing.' },
+        ],
+      });
+      await advanceToRead(7);
+      assert.equal(wave.failures.length, 1, 'read retries before the threshold stay silent');
+      assert.equal(wave.failures[0]!.phase, 'schedule');
+      assert.doesNotMatch(String(wave.failures[0]!.error), /is stuck/);
+
+      await advanceToRead(8);
+      assert.equal(wave.failures.length, 2);
+      assert.equal(wave.failures[1]!.phase, 'schedule');
+      assert.match(String(wave.failures[1]!.error), /is stuck after 8 consecutive failures/);
+
+      await advanceToRead(12);
+      assert.equal(wave.failures.length, 2, 'a stuck read is reported once, not on every retry');
+      assert.equal(wave.settled, false);
+
+      wave.failControlReads(undefined);
+      await advanceToRead(13);
+      await flush();
+      assert.equal(wave.settled, true);
+      const result = await wave.reconciliation;
+      assert.equal(result.status, 'reconciled');
+      assert.equal(result.failures.length, 0, 'a recovered read clears the stuck failure');
+      assert.equal(result.dispatches[0]!.result.status, 'cancelled');
+    } finally {
+      t.mock.timers.reset();
+      await wave.close();
+    }
+  });
+
+  test('a driver abort keeps retrying a failed stop until its parked wave settles', async () => {
+    const wave = await gatedScheduleWave(1);
+    let attempts = 0;
+    const failed = deferred();
+    wave.beforeStop = async () => {
+      attempts += 1;
+      if (attempts <= 2) {
+        failed.resolve();
+        throw new Error('transient exact-stop failure');
+      }
+    };
+    try {
+      await wave.commit('stop-then-abort', {
+        stop: [{ target_id: wave.workIds[0]!, reason: 'Stop while the driver is aborting.' }],
+      });
+      await withTimeout(failed.promise, 1000, 'first stop attempt must fail');
+      wave.abortDriver();
+      const result = await withTimeout(
+        wave.reconciliation,
+        2000,
+        'an aborted driver must not stop retrying the stop that parks its wave',
+      );
+      assert.equal(attempts, 3);
+      assert.equal(result.status, 'cancelled');
+      assert.equal(result.failures.length, 0);
+      assert.equal(result.dispatches[0]!.result.status, 'cancelled');
+    } finally {
+      await wave.close();
+    }
+  });
 });
 
 const GRAPH_ID = 'graph-schedule';
@@ -929,11 +1389,198 @@ function controlStoreWithProvisions(
   };
 }
 
+async function gatedScheduleWave(count: number) {
+  const store = createSqliteSessionMetadataStore(':memory:');
+  const observation = new MemoryGraphObservation();
+  const executor = new MemoryScheduleExecutor(store, observation, 'running');
+  type Identity = Parameters<AgentGraphScheduleStopController['stopAgentGraphActivation']>[0];
+  const active = new Map<
+    string,
+    { identity: Identity; gate: ReturnType<typeof deferred<void>>; cancelled: boolean }
+  >();
+  const listeners = new Set<() => void>();
+  const state = {
+    beforeStop: undefined as (() => Promise<void>) | undefined,
+    onFailure: undefined as (() => void) | undefined,
+    failures: [] as Array<{ phase: string; error: unknown }>,
+    observations: 0,
+    settled: false,
+    readFailure: undefined as Error | undefined,
+    persistentReadFailure: undefined as Error | undefined,
+    controlReads: 0,
+    stopCalls: 0,
+    // Models the runtime's retained cleanup owner after a failed exact stop.
+    retainedStops: new Set<string>(),
+    driver: new AbortController(),
+  };
+  const added = await commitSchedule(store, 'initial-wave', {
+    add_work: Array.from({ length: count }, (_, index) => ({
+      operator_id: 'writer',
+      instruction: `Gated work ${index}.`,
+      input_ids: [],
+    })),
+  });
+  const controlStore = new Proxy(store, {
+    get(target, property) {
+      if (property === 'listAgentGraphScheduleUpdates') {
+        return async (graphId: string) => {
+          state.controlReads += 1;
+          if (state.readFailure) {
+            const error = state.readFailure;
+            state.readFailure = undefined;
+            throw error;
+          }
+          if (state.persistentReadFailure) throw state.persistentReadFailure;
+          return target.listAgentGraphScheduleUpdates(graphId);
+        };
+      }
+      const value = Reflect.get(target, property, target);
+      return typeof value === 'function' ? value.bind(target) : value;
+    },
+  });
+  const input = {
+    topology: topology(),
+    controlStore,
+    executor: {
+      async runClaimedAgentGraphIntent(
+        request: Parameters<AgentGraphIntentExecutor['runClaimedAgentGraphIntent']>[0],
+      ) {
+        const result = await executor.runClaimedAgentGraphIntent(request);
+        const current = {
+          identity: {
+            sessionId: result.childSessionId,
+            runId: result.runId,
+            turnId: result.turnId,
+          },
+          gate: deferred<void>(),
+          cancelled: false,
+        };
+        active.set(result.runId, current);
+        await current.gate.promise;
+        const status = current.cancelled ? ('cancelled' as const) : ('completed' as const);
+        observation.setActivation(result.childSessionId, result.runId, status, result.turnId);
+        return { ...result, status };
+      },
+    },
+    stopController: {
+      async stopAgentGraphActivation(identity: Identity) {
+        state.stopCalls += 1;
+        try {
+          await state.beforeStop?.();
+        } catch (error) {
+          state.retainedStops.add(identity.runId);
+          throw error;
+        }
+        state.retainedStops.delete(identity.runId);
+        const current = active.get(identity.runId);
+        assert.ok(current, 'stop must address an existing exact activation');
+        assert.deepEqual(identity, current.identity);
+        observation.stopActivation(identity);
+        current.cancelled = true;
+        current.gate.resolve();
+      },
+      hasPendingAgentGraphActivationStop(identity: Identity) {
+        return state.retainedStops.has(identity.runId);
+      },
+    },
+    newId: nextId(),
+    maxNewActivations: count,
+    abortSignal: state.driver.signal,
+    observeGraph: async () => {
+      state.observations += 1;
+      return observation.read();
+    },
+    renderPrompt: ({ work }: RenderAgentGraphScheduledWorkPromptInput) => work.instruction,
+    subscribeToScheduleChanges(listener: () => void) {
+      listeners.add(listener);
+      return () => {
+        listeners.delete(listener);
+      };
+    },
+    supervisor: {
+      onReconciliationFailure(failure: { phase: string; error: unknown }) {
+        state.failures.push(failure);
+        state.onFailure?.();
+      },
+    },
+  };
+  const reconciliation = reconcileAgentGraphSchedule(input).finally(() => {
+    state.settled = true;
+  });
+  await waitFor(() => active.size === count, { timeoutMs: 1000 });
+  return {
+    store,
+    executor,
+    reconciliation,
+    workIds: added.addWork.map((work) => work.workId),
+    get beforeStop() {
+      return state.beforeStop;
+    },
+    set beforeStop(value) {
+      state.beforeStop = value;
+    },
+    get onFailure() {
+      return state.onFailure;
+    },
+    set onFailure(value) {
+      state.onFailure = value;
+    },
+    get failures() {
+      return state.failures;
+    },
+    get observations() {
+      return state.observations;
+    },
+    get settled() {
+      return state.settled;
+    },
+    get stopCalls() {
+      return state.stopCalls;
+    },
+    abortDriver() {
+      state.driver.abort();
+    },
+    get subscriptions() {
+      return listeners.size;
+    },
+    async commit(toolCallId: string, update: UpdateAgentGraphToolInput) {
+      await commitSchedule(store, toolCallId, update);
+      for (const listener of listeners) listener();
+    },
+    markTerminalWithoutCompleting() {
+      assert.equal(active.size, 1, 'this cleanup control owns one exact activation');
+      for (const current of active.values()) {
+        observation.stopActivation(current.identity);
+        current.cancelled = true;
+      }
+      // Keep the completion gate closed until a later exact-stop attempt succeeds.
+    },
+    failNextControlRead(error: Error) {
+      state.readFailure = error;
+    },
+    /** Fail every control read until called without an error. */
+    failControlReads(error: Error | undefined) {
+      state.persistentReadFailure = error;
+    },
+    get controlReads() {
+      return state.controlReads;
+    },
+    reconcileAgain: () => reconcileAgentGraphSchedule(input),
+    async close() {
+      state.beforeStop = undefined;
+      for (const current of active.values()) current.gate.resolve();
+      await reconciliation;
+      store.close();
+    },
+  };
+}
+
 class MemoryGraphObservation {
   private readonly activations = new Map<
     string,
     {
       sessionId: string;
+      turnId: string;
       status: 'running' | 'completed' | 'cancelled';
     }
   >();
@@ -942,15 +1589,20 @@ class MemoryGraphObservation {
     sessionId: string,
     activationId: string,
     status: 'running' | 'completed' | 'cancelled',
+    turnId: string,
   ): void {
-    this.activations.set(activationId, { sessionId, status });
+    this.activations.set(activationId, { sessionId, turnId, status });
   }
 
-  stopSession(sessionId: string): void {
-    for (const [activationId, activation] of this.activations) {
-      if (activation.sessionId === sessionId && activation.status === 'running') {
-        this.activations.set(activationId, { ...activation, status: 'cancelled' });
-      }
+  stopActivation(
+    identity: Parameters<AgentGraphScheduleStopController['stopAgentGraphActivation']>[0],
+  ): void {
+    const activation = this.activations.get(identity.runId);
+    if (!activation) return;
+    assert.equal(activation.sessionId, identity.sessionId);
+    assert.equal(activation.turnId, identity.turnId);
+    if (activation.status === 'running') {
+      this.activations.set(identity.runId, { ...activation, status: 'cancelled' });
     }
   }
 
@@ -999,6 +1651,39 @@ class MemoryGraphObservation {
         operators: currentTopology.operators.map((operator) => ({ ...operator })),
         ignoredPartialEvents: 0,
         records: [
+          ...[...this.activations].map(([activationId, activation]): AgentGraphRecord => {
+            const binding = currentTopology.operators.find(
+              (operator) => operator.sessionId === activation.sessionId,
+            )!;
+            return {
+              schemaVersion: 1,
+              recordId: `record-${activationId}`,
+              graphId: GRAPH_ID,
+              operatorId: binding.operatorId,
+              activationId,
+              sessionId: activation.sessionId,
+              agentRunId: activationId,
+              eventTime: 2,
+              orderKey: {
+                runCreatedAt: 1,
+                operatorId: binding.operatorId,
+                runId: activationId,
+                committedEventOrdinal: 0,
+                runtimeEventId: `event-${activationId}`,
+              },
+              type: 'agent_runtime_event',
+              facets: ['message'],
+              supervisorSignals: [],
+              source: {
+                kind: 'runtime_event',
+                runtimeEventId: `event-${activationId}`,
+                sessionId: activation.sessionId,
+                runId: activationId,
+                turnId: activation.turnId,
+                ts: 2,
+              },
+            };
+          }),
           {
             schemaVersion: 1,
             recordId: 'record-input',
@@ -1099,7 +1784,12 @@ class MemoryScheduleExecutor implements AgentGraphIntentExecutor {
     }
     this.lastPrompt = input.prompt;
     this.backendInvocations += 1;
-    this.observation.setActivation(claim.targetSessionId, claim.targetRunId, this.status);
+    this.observation.setActivation(
+      claim.targetSessionId,
+      claim.targetRunId,
+      this.status,
+      claim.targetTurnId,
+    );
     await input.onReady?.({
       claimId: claim.claimId,
       graphId: claim.graphId,
@@ -1137,13 +1827,21 @@ class MemoryScheduleExecutor implements AgentGraphIntentExecutor {
 }
 
 class MemoryStopController {
-  readonly calls: Array<{ sessionId: string; source: 'graph_supervisor' }> = [];
+  readonly calls: Array<{
+    sessionId: string;
+    runId: string;
+    turnId: string;
+    source: 'graph_supervisor';
+  }> = [];
 
   constructor(private readonly observation: MemoryGraphObservation) {}
 
-  async stopSession(sessionId: string, input: { source: 'graph_supervisor' }): Promise<void> {
-    this.calls.push({ sessionId, source: input.source });
-    this.observation.stopSession(sessionId);
+  async stopAgentGraphActivation(
+    identity: Parameters<AgentGraphScheduleStopController['stopAgentGraphActivation']>[0],
+    input: { source: 'graph_supervisor' },
+  ): Promise<void> {
+    this.calls.push({ ...identity, source: input.source });
+    this.observation.stopActivation(identity);
   }
 }
 function nextNumber(start: number): () => number {
diff --git a/packages/runtime/src/__tests__/stream-graph-schedule-stop.test.ts b/packages/runtime/src/__tests__/stream-graph-schedule-stop.test.ts
new file mode 100644
index 0000000000..9aaa1ddbde
--- /dev/null
+++ b/packages/runtime/src/__tests__/stream-graph-schedule-stop.test.ts
@@ -0,0 +1,495 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import assert from 'node:assert/strict';
+import { randomUUID } from 'node:crypto';
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { test } from 'node:test';
+import { z } from 'zod';
+import type { AgentBackend, BackendSendInput } from '@maka/core/backend-types';
+import type { SessionEvent } from '@maka/core/events';
+import { isTerminalRuntimeEvent } from '@maka/core/runtime-event';
+import {
+  deferred,
+  waitFor,
+  withTimeout,
+  type Deferred,
+} from '@maka/core/test-only/async-primitives';
+import { createSqliteAgentRunStore } from '@maka/storage/agent-run-store';
+import { OPERATIONAL_STATE_DATABASE_NAME } from '@maka/storage/operational-state-store';
+import { createWorkspaceRuntimeStore } from '@maka/storage/runtime-event-persistence';
+import { createSessionStore } from '@maka/storage/session-store';
+import { createSqliteSessionMetadataStore } from '@maka/storage/sqlite-session-metadata-store';
+import { BackendRegistry, SessionManager } from '../session-manager.js';
+import { AgentGraphCoordinator, agentGraphIdForRootSession } from '../stream-graph-coordinator.js';
+import type { UpdateAgentGraphToolInput } from '../stream-graph-supervisor-tools.js';
+import type { MakaToolContext } from '../tool-runtime.js';
+
+// This backend cannot finish a child naturally until the test opens its gate.
+// The real SessionManager still owns admission, cancellation and terminal facts.
+interface GatedSend {
+  turnId: string;
+  naturalCompletion: Deferred;
+  stopped: Deferred;
+  wasStopped: boolean;
+}
+
+class GatedGraphBackend implements AgentBackend {
+  readonly kind = 'ai-sdk' as const;
+  readonly started = deferred();
+  readonly sends: GatedSend[] = [];
+  // A non-cooperative provider: stop is delivered, but neither stop nor the
+  // send returns before the provider finishes on its own.
+  ignoreStop = false;
+  stopCalls = 0;
+  private active: GatedSend | undefined;
+
+  constructor(
+    readonly sessionId: string,
+    private readonly child: boolean,
+  ) {}
+
+  get naturalCompletion() {
+    const current = this.active ?? this.sends.at(-1);
+    assert.ok(current, 'the backend must have started before opening its completion gate');
+    return current.naturalCompletion;
+  }
+
+  async *send(input: BackendSendInput): AsyncIterable<SessionEvent> {
+    const call: GatedSend = {
+      turnId: input.turnId,
+      naturalCompletion: deferred(),
+      stopped: deferred(),
+      wasStopped: false,
+    };
+    this.sends.push(call);
+    this.active = call;
+    const event = { turnId: input.turnId, ts: Date.now() };
+    try {
+      if (this.child) {
+        this.started.resolve();
+        yield {
+          ...event,
+          id: randomUUID(),
+          type: 'text_delta',
+          messageId: input.turnId,
+          text: 'Working on the requested task.',
+        };
+        await Promise.race([call.naturalCompletion.promise, call.stopped.promise]);
+      }
+      yield {
+        ...event,
+        id: randomUUID(),
+        type: 'text_complete',
+        messageId: input.turnId,
+        text: call.wasStopped ? 'Stopped.' : 'Completed requested work.',
+      };
+      yield {
+        ...event,
+        id: randomUUID(),
+        type: 'complete',
+        stopReason: call.wasStopped ? 'user_stop' : 'end_turn',
+      };
+    } finally {
+      if (this.active === call) this.active = undefined;
+    }
+  }
+  async stop(): Promise<void> {
+    this.stopCalls += 1;
+    if (!this.active) return;
+    if (this.ignoreStop) {
+      await this.active.naturalCompletion.promise;
+      return;
+    }
+    this.active.wasStopped = true;
+    this.active.stopped.resolve();
+  }
+  async dispose(): Promise<void> {
+    await this.stop();
+  }
+  async respondToSandboxBoundary(): Promise<void> {}
+}
+
+async function graphFixture() {
+  const root = await mkdtemp(join(tmpdir(), 'maka-graph-schedule-stop-'));
+  const sessionStore = createSessionStore(root);
+  const runStore = createSqliteAgentRunStore(root);
+  const runtimeEventStore = createWorkspaceRuntimeStore(root);
+  const controlStore = createSqliteSessionMetadataStore(
+    join(root, OPERATIONAL_STATE_DATABASE_NAME),
+  );
+  const children = new Map<string, GatedGraphBackend>();
+  const backends = new BackendRegistry();
+  backends.register('ai-sdk', (context) => {
+    const backend = new GatedGraphBackend(
+      context.sessionId,
+      Boolean(context.header.subagentParent),
+    );
+    if (context.header.subagentParent) children.set(context.sessionId, backend);
+    return backend;
+  });
+  const manager = new SessionManager({
+    store: sessionStore,
+    runStore,
+    runtimeEventStore,
+    backends,
+    childTools: ['Read', 'Glob', 'Grep'].map((name) => ({
+      name,
+      description: 'In-memory fixture',
+      parameters: z.object({}),
+      categoryHint: 'read',
+      impl: async () => ({ ok: true }),
+    })),
+    newId: randomUUID,
+    now: Date.now,
+  });
+  const session = await manager.createSession({
+    cwd: root,
+    llmConnectionSlug: 'fake',
+    model: 'fake-model',
+    permissionMode: 'ask',
+    name: 'Graph supervisor',
+  });
+  const turnId = randomUUID();
+  for await (const _event of manager.sendMessage(session.id, {
+    turnId,
+    text: 'Prepare graph work.',
+  })) {
+  }
+  const [run] = await runtimeEventStore.listSessionInvocations(session.id);
+  assert.ok(run);
+  const coordinators: AgentGraphCoordinator[] = [];
+  const createCoordinator = () => {
+    const coordinator = new AgentGraphCoordinator({
+      sessionStore,
+      runtimeEventStore,
+      controlStore,
+      runtime: manager,
+      newId: randomUUID,
+      maxNewActivations: 4,
+    });
+    coordinators.push(coordinator);
+    return coordinator;
+  };
+  const coordinator = createCoordinator();
+  const update = (await coordinator.toolsForSession(session.id)).find(
+    (tool) => tool.name === 'update_agent_graph',
+  )!;
+  const context = (): MakaToolContext => ({
+    sessionId: session.id,
+    runId: run.runId,
+    turnId,
+    toolCallId: randomUUID(),
+    cwd: root,
+    abortSignal: new AbortController().signal,
+    emitOutput() {},
+  });
+  return {
+    manager,
+    coordinator,
+    createCoordinator,
+    children,
+    session,
+    runtimeEventStore,
+    controlStore,
+    update: (input: UpdateAgentGraphToolInput) => update.impl(input, context()),
+    async start(count: number) {
+      const previous = new Set(children.keys());
+      await update.impl(
+        {
+          operation: 'add_work',
+          add_work: Array.from({ length: count }, (_, i) => ({
+            target_kind: 'new_agent',
+            agent_id: 'local-read',
+            instruction: `Inspect local input ${i}.`,
+            input_ids: [],
+            replacement_mode: 'none',
+          })),
+        },
+        context(),
+      );
+      await waitFor(() => children.size === previous.size + count, { timeoutMs: 5000 });
+      const added = [...children.values()].filter((child) => !previous.has(child.sessionId));
+      await Promise.all(added.map((child) => child.started.promise));
+      const provisions = await controlStore.listAgentGraphOperatorProvisions(
+        agentGraphIdForRootSession(session.id),
+      );
+      return added.map((child) => ({
+        child,
+        workId: provisions.find((p) => p.targetSessionId === child.sessionId)!.workId,
+      }));
+    },
+    async close() {
+      for (const child of children.values()) {
+        for (const send of child.sends) send.naturalCompletion.resolve();
+      }
+      await Promise.all(coordinators.map((owned) => owned.close()));
+      controlStore.close();
+      await rm(root, { recursive: true, force: true });
+    },
+  };
+}
+
+test('a durable graph stop cancels the running child before natural completion', async () => {
+  const fixture = await graphFixture();
+  try {
+    const [{ child, workId }] = await fixture.start(1);
+    await fixture.update({
+      operation: 'stop',
+      stop: [{ target_id: workId, reason: 'This work is no longer needed.' }],
+    });
+    await withTimeout(
+      fixture.coordinator.waitForIdle(fixture.session.id),
+      1000,
+      'graph stop must settle the child while its natural completion gate remains closed',
+    );
+    assert.equal((await fixture.manager.listTurns(child.sessionId))[0]?.status, 'aborted');
+    assert.equal(
+      (await fixture.manager.listSessions()).find((s) => s.id === child.sessionId)?.status,
+      'aborted',
+    );
+    assert.deepEqual(fixture.manager.runningTurnIds(child.sessionId), []);
+    await assertDurableTerminal(fixture, child, 'aborted');
+  } finally {
+    await fixture.close();
+  }
+});
+
+type GraphFixture = Awaited<ReturnType<typeof graphFixture>>;
+
+async function assertDurableTerminal(
+  fixture: GraphFixture,
+  child: GatedGraphBackend,
+  expected: 'aborted' | 'completed' | 'either',
+  coordinator = fixture.coordinator,
+) {
+  const runs = await fixture.runtimeEventStore.listSessionInvocations(child.sessionId);
+  assert.equal(runs.length, 1, 'one child work must create exactly one durable activation');
+  const run = runs[0]!;
+  const events = await fixture.runtimeEventStore.readImmutableRuntimeEvents(
+    child.sessionId,
+    run.runId,
+  );
+  const terminal = events.filter(isTerminalRuntimeEvent);
+  assert.equal(terminal.length, 1, 'the activation must have exactly one durable terminal fact');
+  if (expected === 'either') {
+    assert.ok(terminal[0]!.status === 'completed' || terminal[0]!.status === 'aborted');
+  } else {
+    assert.equal(terminal[0]!.status, expected);
+  }
+  assert.equal(run.terminalEvent?.id, terminal[0]!.id);
+  const claims = await fixture.controlStore.listAgentGraphIntentClaims(
+    agentGraphIdForRootSession(fixture.session.id),
+  );
+  const childClaims = claims.filter((claim) => claim.targetSessionId === child.sessionId);
+  assert.equal(childClaims.length, 1);
+  assert.equal(childClaims[0]!.targetRunId, run.runId);
+  const observation = await coordinator.observe(fixture.session.id);
+  const terminalRecords = observation.projection.records.filter(
+    (record) => record.source.runtimeEventId === terminal[0]!.id,
+  );
+  assert.equal(
+    terminalRecords.length,
+    1,
+    'the durable terminal must project to one graph result record',
+  );
+  assert.equal(terminalRecords[0]!.activationId, run.runId);
+  assert.equal(child.sends.length, 1, 'reconciliation must not execute the same work again');
+  return { runId: run.runId, terminalId: terminal[0]!.id, recordId: terminalRecords[0]!.recordId };
+}
+
+test('stopping one graph child leaves its sibling running until natural completion', async () => {
+  const fixture = await graphFixture();
+  try {
+    const [target, sibling] = await fixture.start(2);
+    await fixture.update({
+      operation: 'stop',
+      stop: [{ target_id: target!.workId, reason: 'Only this branch is obsolete.' }],
+    });
+    await waitFor(
+      async () => {
+        const runs = await fixture.runtimeEventStore.listSessionInvocations(
+          target!.child.sessionId,
+        );
+        return runs[0]?.terminalEvent?.status === 'aborted';
+      },
+      { timeoutMs: 1000, message: 'selected child must stop while its sibling remains gated' },
+    );
+    await assertDurableTerminal(fixture, target!.child, 'aborted');
+    const siblingRuns = await fixture.runtimeEventStore.listSessionInvocations(
+      sibling!.child.sessionId,
+    );
+    assert.equal(siblingRuns.length, 1);
+    assert.equal(siblingRuns[0]!.terminalEvent, undefined);
+    assert.deepEqual(fixture.manager.runningTurnIds(sibling!.child.sessionId), [
+      siblingRuns[0]!.turnId,
+    ]);
+    sibling!.child.naturalCompletion.resolve();
+    await withTimeout(
+      fixture.coordinator.waitForIdle(fixture.session.id),
+      3000,
+      'sibling should finish',
+    );
+    await assertDurableTerminal(fixture, sibling!.child, 'completed');
+  } finally {
+    await fixture.close();
+  }
+});
+
+test('new graph work completes after an earlier work was stopped', async () => {
+  const fixture = await graphFixture();
+  try {
+    const [first] = await fixture.start(1);
+    await fixture.update({
+      operation: 'stop',
+      stop: [{ target_id: first!.workId, reason: 'Replace the abandoned branch.' }],
+    });
+    await withTimeout(
+      fixture.coordinator.waitForIdle(fixture.session.id),
+      1000,
+      'first work should stop',
+    );
+    const firstFacts = await assertDurableTerminal(fixture, first!.child, 'aborted');
+    const [next] = await fixture.start(1);
+    assert.notEqual(next!.child.sessionId, first!.child.sessionId);
+    assert.equal(fixture.manager.runningTurnIds(next!.child.sessionId).length, 1);
+    next!.child.naturalCompletion.resolve();
+    await withTimeout(
+      fixture.coordinator.waitForIdle(fixture.session.id),
+      3000,
+      'new work should finish',
+    );
+    await assertDurableTerminal(fixture, next!.child, 'completed');
+    assert.deepEqual(await assertDurableTerminal(fixture, first!.child, 'aborted'), firstFacts);
+  } finally {
+    await fixture.close();
+  }
+});
+
+test('concurrent graph stop and natural completion preserve one durable activation and result', async () => {
+  const fixture = await graphFixture();
+  try {
+    const [{ child, workId }] = await fixture.start(1);
+    const stop = fixture.update({
+      operation: 'stop',
+      stop: [{ target_id: workId, reason: 'Stop races with the final child output.' }],
+    });
+    child.naturalCompletion.resolve();
+    await stop;
+    await withTimeout(
+      fixture.coordinator.waitForIdle(fixture.session.id),
+      3000,
+      'racing operations should settle',
+    );
+    const facts = await assertDurableTerminal(fixture, child, 'either');
+    await fixture.coordinator.reconcile(fixture.session.id);
+    assert.deepEqual(await assertDurableTerminal(fixture, child, 'either'), facts);
+    assert.deepEqual(fixture.manager.runningTurnIds(child.sessionId), []);
+  } finally {
+    await fixture.close();
+  }
+});
+
+test('a fresh graph coordinator recovers stopped and completed work without replaying children', async () => {
+  const fixture = await graphFixture();
+  try {
+    const [stopped, completed] = await fixture.start(2);
+    await fixture.update({
+      operation: 'stop',
+      stop: [{ target_id: stopped!.workId, reason: 'Persist this cancellation before recovery.' }],
+    });
+    await waitFor(
+      async () => {
+        const runs = await fixture.runtimeEventStore.listSessionInvocations(
+          stopped!.child.sessionId,
+        );
+        return runs[0]?.terminalEvent?.status === 'aborted';
+      },
+      { timeoutMs: 1000, message: 'stop must settle before the sibling is allowed to complete' },
+    );
+    completed!.child.naturalCompletion.resolve();
+    await withTimeout(
+      fixture.coordinator.waitForIdle(fixture.session.id),
+      3000,
+      'work should settle before recovery',
+    );
+    const stoppedFacts = await assertDurableTerminal(fixture, stopped!.child, 'aborted');
+    const completedFacts = await assertDurableTerminal(fixture, completed!.child, 'completed');
+    await fixture.coordinator.close();
+    const recovered = fixture.createCoordinator();
+    assert.deepEqual(await recovered.recover(), [fixture.session.id]);
+    assert.deepEqual(
+      await assertDurableTerminal(fixture, stopped!.child, 'aborted', recovered),
+      stoppedFacts,
+    );
+    assert.deepEqual(
+      await assertDurableTerminal(fixture, completed!.child, 'completed', recovered),
+      completedFacts,
+    );
+    assert.equal(fixture.children.size, 2);
+  } finally {
+    await fixture.close();
+  }
+});
+
+test('a stop that one backend ignores does not hold back its sibling stop in the same pass', async () => {
+  const fixture = await graphFixture();
+  try {
+    const [ignored, cooperative] = await fixture.start(2);
+    ignored!.child.ignoreStop = true;
+    await fixture.update({
+      operation: 'stop',
+      stop: [
+        { target_id: ignored!.workId, reason: 'This provider ignores the stop.' },
+        { target_id: cooperative!.workId, reason: 'This provider honors the stop.' },
+      ],
+    });
+    await waitFor(
+      async () => {
+        const runs = await fixture.runtimeEventStore.listSessionInvocations(
+          cooperative!.child.sessionId,
+        );
+        return runs[0]?.terminalEvent?.status === 'aborted';
+      },
+      {
+        timeoutMs: 1000,
+        message: 'the cooperative stop must settle while its sibling still ignores its own',
+      },
+    );
+    assert.ok(ignored!.child.stopCalls >= 1, 'both stops are delivered in the same pass');
+    const ignoredRuns = await fixture.runtimeEventStore.listSessionInvocations(
+      ignored!.child.sessionId,
+    );
+    assert.equal(ignoredRuns[0]!.terminalEvent, undefined);
+    await assertDurableTerminal(fixture, cooperative!.child, 'aborted');
+
+    ignored!.child.naturalCompletion.resolve();
+    await withTimeout(
+      fixture.coordinator.waitForIdle(fixture.session.id),
+      3000,
+      'the ignored stop settles once its provider finishes',
+    );
+    await assertDurableTerminal(fixture, ignored!.child, 'either');
+    await assertDurableTerminal(fixture, cooperative!.child, 'aborted');
+  } finally {
+    await fixture.close();
+  }
+});
diff --git a/packages/runtime/src/agent-run.ts b/packages/runtime/src/agent-run.ts
index 3ba9c42ed8..ba2f0a1e7f 100644
--- a/packages/runtime/src/agent-run.ts
+++ b/packages/runtime/src/agent-run.ts
@@ -553,6 +553,15 @@ export class AgentRun {
     // method exists to prevent.
     const runStore = this.input.runStore;
     if (!this.input.runtimeEventStore || !runStore) return;
+    // A stream may have attempted finalization after an earlier stop failed.
+    // Join that write, but let this explicit retry reuse its reserved fact if
+    // the write failed; a rejected promise is not durable terminal evidence.
+    const previousWrite = this.terminalClaim.write;
+    try {
+      await previousWrite;
+    } catch {
+      if (this.terminalClaim.write === previousWrite) this.terminalClaim.write = undefined;
+    }
     await this.flushRuntimePartialBuffer(true);
     // The claim only fences writers inside this Run. Another owner — a Host
     // recovery, a resumed continuation — may have sealed the ledger already,
diff --git a/packages/runtime/src/message-authority.ts b/packages/runtime/src/message-authority.ts
index bb9b5e403c..06349ba108 100644
--- a/packages/runtime/src/message-authority.ts
+++ b/packages/runtime/src/message-authority.ts
@@ -63,7 +63,14 @@ export interface RuntimeHostedRootExecutionInput extends RuntimeMessageRunIdenti
 /** Host-only root lifecycle capability. Embedded compositions must omit it. */
 export interface RuntimeHostedRootAuthority extends RuntimeMessageAuthority {
   executeRoot(input: RuntimeHostedRootExecutionInput): Promise<void>;
+  /** Stops the Session's live execution, including a claim whose Run has not attached. */
   stopRoot(identity: RuntimeMessageRunIdentity, input?: StopSessionInput): Promise<void>;
+  /**
+   * Stops only this exact Run's Runtime owner, for an activation-scoped graph
+   * stop. Kept separate from `stopRoot` so an authority cannot satisfy it by
+   * ignoring an option and silently widening to a Session stop.
+   */
+  stopRootRun(identity: RuntimeMessageRunIdentity, input?: StopSessionInput): Promise<void>;
   stopSession(sessionId: string, input?: StopSessionInput): Promise<void>;
 }
 
diff --git a/packages/runtime/src/runtime-kernel.ts b/packages/runtime/src/runtime-kernel.ts
index 4bada91831..7f5b67bca6 100644
--- a/packages/runtime/src/runtime-kernel.ts
+++ b/packages/runtime/src/runtime-kernel.ts
@@ -183,6 +183,10 @@ export interface RuntimeKernelLike {
   compactSession(sessionId: string, input?: CompactSessionInput): AsyncIterable<SessionEvent>;
   preflightContextCompaction(sessionId: string): Promise<void>;
   stopSession(sessionId: string, input?: StopSessionInput): Promise<void>;
+  stopExecution?(claim: RuntimeExecutionClaim, input?: StopSessionInput): Promise<void>;
+  waitForExecutionStop?(claim: RuntimeExecutionClaim): Promise<void>;
+  stopRun?(identity: RuntimeMessageRunIdentity, input?: StopSessionInput): Promise<void>;
+  hasPendingRunStop?(identity: RuntimeMessageRunIdentity): boolean;
   respondToSandboxBoundary(sessionId: string, response: SandboxBoundaryResponse): Promise<void>;
   listActiveInteractions?(sessionId: string): ActiveInteractionRequestEvent[];
   respondToUserQuestion?(sessionId: string, response: UserQuestionResponse): Promise<void>;
@@ -359,6 +363,9 @@ interface StopOperation {
   statusProjected: boolean;
   targets: Map<number, StopTarget>;
   queue: Promise<void>;
+  completion: Promise<void>;
+  resolveCompletion(): void;
+  rejectCompletion(error: unknown): void;
 }
 
 interface SessionStopIntent {
@@ -383,6 +390,7 @@ interface PendingExecutionClaim {
   backendHeaderSnapshot?: { invalidated: boolean };
   backendPreparation?: PreparedBackendActivation;
   stopIntent?: SessionStopIntent;
+  scopedStopAttempt?: Promise<void>;
   finalization?: ExecutionClaimOutcome;
 }
 
@@ -1933,6 +1941,130 @@ export class RuntimeKernel implements RuntimeKernelLike {
     ];
   }
 
+  /** Stop one captured owner without cancelling other queued executions in its Session. */
+  stopExecution(claim: RuntimeExecutionClaim, input: StopSessionInput = {}): Promise<void> {
+    normalizeStopSessionSource(input.source, input.workHubActionId);
+    const execution = this.executionClaimStates.get(claim);
+    if (!execution) throw new Error('Runtime execution claim belongs to another owner');
+    if (execution.scopedStopAttempt) return execution.scopedStopAttempt;
+    const run = execution.run;
+    const active =
+      run &&
+      this.backendGenerationsFor(execution.sessionId).find(
+        (candidate) => candidate.activeRuns.get(run.runId) === run,
+      );
+    if (!active && (execution.phase === 'released' || execution.phase === 'failed')) {
+      return run
+        ? this.retryStoppedRun(
+            {
+              sessionId: execution.sessionId,
+              runId: run.runId,
+              turnId: run.turnId,
+            },
+            input,
+          )
+        : Promise.resolve();
+    }
+    if (active && run) this.assertScopedStopGeneration(active, run);
+    // Validate first, then capture the capability synchronously. A pending
+    // owner carries the stop through attach/reserve without a Session fence.
+    execution.stopIntent ??= { input, claims: new Set([execution]) };
+    if (active && run) this.captureScopedRunStop(active, run, input);
+    else run?.stop(input.source, input.workHubActionId);
+    execution.abortController.abort(execution.cancellation);
+    const attempt = this.stopCapturedExecution(execution, input).finally(() => {
+      if (execution.scopedStopAttempt === attempt) execution.scopedStopAttempt = undefined;
+    });
+    execution.scopedStopAttempt = attempt;
+    return attempt;
+  }
+
+  /** A failed cleanup attempt keeps its captured owner parked until a successful retry. */
+  waitForExecutionStop(claim: RuntimeExecutionClaim): Promise<void> {
+    const run = this.executionClaimStates.get(claim)?.run;
+    const operation =
+      run &&
+      this.retainedRunStopOperation({
+        sessionId: claim.sessionId,
+        runId: run.runId,
+        turnId: run.turnId,
+      });
+    return operation ? operation.completion : Promise.resolve();
+  }
+
+  /** Whether a retained stop operation still owns cleanup for this exact Run. */
+  hasPendingRunStop(identity: RuntimeMessageRunIdentity): boolean {
+    return this.retainedRunStopOperation(identity) !== undefined;
+  }
+
+  private retainedRunStopOperation(identity: RuntimeMessageRunIdentity): StopOperation | undefined {
+    const operation = this.stopOperations.get(identity.sessionId);
+    return operation &&
+      [...operation.targets.values()].some(
+        (target) => target.runs.get(identity.runId)?.turnId === identity.turnId,
+      )
+      ? operation
+      : undefined;
+  }
+
+  /** Exact identity lookup never redirects a completed Run's stop to its successor. */
+  stopRun(identity: RuntimeMessageRunIdentity, input: StopSessionInput = {}): Promise<void> {
+    normalizeStopSessionSource(input.source, input.workHubActionId);
+    const execution = [...(this.executionClaims.get(identity.sessionId) ?? [])].find(
+      (candidate) =>
+        candidate.run?.runId === identity.runId && candidate.run.turnId === identity.turnId,
+    );
+    if (execution) return this.stopExecution(execution.handle, input);
+    for (const active of this.backendGenerationsFor(identity.sessionId)) {
+      const run = active.activeRuns.get(identity.runId);
+      if (!run || run.turnId !== identity.turnId) continue;
+      this.assertScopedStopGeneration(active, run);
+      this.captureScopedRunStop(active, run, input);
+      return this.retryStoppedRun(identity, input);
+    }
+    return this.retryStoppedRun(identity, input);
+  }
+
+  private assertScopedStopGeneration(active: BackendGeneration, run: AgentRun): void {
+    if ([...active.activeRuns.values()].some((other) => other !== run && !other.isStopped())) {
+      throw new Error('Cannot stop one Run while its backend generation owns another active Run');
+    }
+  }
+
+  private captureScopedRunStop(
+    active: BackendGeneration,
+    run: AgentRun,
+    input: StopSessionInput,
+  ): void {
+    const operation = this.claimRunForStop(active.sessionId, input, active, run);
+    if (operation && active.phase === 'active') active.phase = 'stopping';
+  }
+
+  private async stopCapturedExecution(
+    execution: PendingExecutionClaim,
+    input: StopSessionInput,
+  ): Promise<void> {
+    await execution.settled;
+    if (execution.run) {
+      await this.retryStoppedRun(
+        {
+          sessionId: execution.sessionId,
+          runId: execution.run.runId,
+          turnId: execution.run.turnId,
+        },
+        input,
+      );
+    }
+  }
+
+  private async retryStoppedRun(
+    identity: RuntimeMessageRunIdentity,
+    input: StopSessionInput,
+  ): Promise<void> {
+    const operation = this.retainedRunStopOperation(identity);
+    if (operation) await this.enqueueStopOperation(identity.sessionId, operation, input, true);
+  }
+
   stopSession(sessionId: string, input: StopSessionInput = {}): Promise<void> {
     normalizeStopSessionSource(input.source, input.workHubActionId);
     const existing = this.stopAttempts.get(sessionId);
@@ -2037,12 +2169,22 @@ export class RuntimeKernel implements RuntimeKernelLike {
   private buildStopOperation(input: StopSessionInput): StopOperation {
     const abortSource = normalizeStopSessionSource(input.source, input.workHubActionId);
     const ts = this.deps.now();
+    let resolveCompletion!: () => void;
+    let rejectCompletion!: (error: unknown) => void;
+    const completion = new Promise<void>((resolve, reject) => {
+      resolveCompletion = resolve;
+      rejectCompletion = reject;
+    });
+    void completion.catch(() => undefined);
     return {
       abortSource,
       ts,
       statusProjected: false,
       targets: new Map(),
       queue: Promise.resolve(),
+      completion,
+      resolveCompletion,
+      rejectCompletion,
     };
   }
 
@@ -2162,7 +2304,13 @@ export class RuntimeKernel implements RuntimeKernelLike {
     for (const target of operation.targets.values()) {
       if (target.delivery.kind === 'failed') failures.add(target.delivery.error);
     }
-    failures.throwIfAny(`Stop cleanup failed for session ${sessionId}`);
+    try {
+      failures.throwIfAny(`Stop cleanup failed for session ${sessionId}`);
+      if (completed) operation.resolveCompletion();
+    } catch (error) {
+      if (completed) operation.rejectCompletion(error);
+      throw error;
+    }
   }
 
   async respondToSandboxBoundary(
@@ -2836,7 +2984,9 @@ export class RuntimeKernel implements RuntimeKernelLike {
   private releaseStoppedRunReferences(operation: StopOperation, run: AgentRun): void {
     for (const target of operation.targets.values()) {
       const stoppedRun = target.runs.get(run.runId);
-      if (stoppedRun?.run === run) stoppedRun.run = undefined;
+      // A failed terminal append still needs this captured Run on retry, even
+      // after its stream exits. Drop it only once stop settlement succeeded.
+      if (stoppedRun?.run === run && stoppedRun.stopCompleted) stoppedRun.run = undefined;
       if (
         target.active &&
         target.delivery.kind !== 'pending' &&
@@ -2878,7 +3028,7 @@ export class RuntimeKernel implements RuntimeKernelLike {
     const ownsCurrentStop =
       execution.phase === 'attached' &&
       execution.stopIntent !== undefined &&
-      this.stopIntents.get(sessionId) === execution.stopIntent;
+      execution.stopIntent.claims.has(execution);
     if (this.stopOperations.has(sessionId) && !ownsCurrentStop) {
       throw new Error(`Session ${sessionId} is quarantined by a retained stop operation`);
     }
diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts
index bd416d90cf..74877ecbff 100644
--- a/packages/runtime/src/session-manager.ts
+++ b/packages/runtime/src/session-manager.ts
@@ -409,6 +409,7 @@ type ResolvedClaimedAgentGraphIntentInput = Omit<
 > & {
   claim: AgentGraphIntentClaim;
   hostedGraphExecution?: RuntimeHostedAgentGraphExecutionCapability;
+  onHostAdmission?: () => void;
 };
 
 const CHILD_AGENT_SUMMARY_MAX_CHARS = 4_000;
@@ -903,7 +904,16 @@ export class SessionManager {
   >();
   private readonly claimedAgentGraphIntentRuns = new Map<
     string,
-    { requestFingerprint: string; promise: Promise<ClaimedAgentGraphIntentResult> }
+    {
+      requestFingerprint: string;
+      owner: {
+        identity: RuntimeMessageRunIdentity;
+        execution: RuntimeExecutionClaim;
+        hostAdmitted: boolean;
+        stopTask?: Promise<void>;
+      };
+      promise: Promise<ClaimedAgentGraphIntentResult>;
+    }
   >();
   private readonly claimedAgentGraphSessionTails = new Map<string, Promise<void>>();
 
@@ -2948,11 +2958,33 @@ export class SessionManager {
       return await inFlight.promise;
     }
     const runtimeExecution = this.runtimeKernel.claimExecution(claim.targetSessionId);
-    const promise = this.enqueueClaimedAgentGraphIntent(resolved, runtimeExecution).finally(() =>
-      runtimeExecution.release(),
-    );
+    const owner: {
+      identity: RuntimeMessageRunIdentity;
+      execution: RuntimeExecutionClaim;
+      hostAdmitted: boolean;
+      stopTask?: Promise<void>;
+    } = {
+      identity: {
+        sessionId: claim.targetSessionId,
+        runId: claim.targetRunId,
+        turnId: claim.targetTurnId,
+      },
+      execution: runtimeExecution,
+      hostAdmitted: false,
+    };
+    resolved.onHostAdmission = () => {
+      owner.hostAdmitted = true;
+    };
+    const promise = this.enqueueClaimedAgentGraphIntent(resolved, runtimeExecution, async () => {
+      runtimeExecution.release();
+      // Keep the existing Session slot until scoped backend cleanup completes.
+      // Other queued claims retain their own uncancelled runtime capabilities.
+      await owner.stopTask?.catch(() => undefined);
+      await this.runtimeKernel.waitForExecutionStop?.(runtimeExecution);
+    });
     this.claimedAgentGraphIntentRuns.set(claim.claimId, {
       requestFingerprint,
+      owner,
       promise,
     });
     try {
@@ -2967,6 +2999,7 @@ export class SessionManager {
   private enqueueClaimedAgentGraphIntent(
     input: ResolvedClaimedAgentGraphIntentInput,
     runtimeExecution: RuntimeExecutionClaim,
+    release: () => Promise<void>,
   ): Promise<ClaimedAgentGraphIntentResult> {
     const sessionId = input.claim.targetSessionId;
     const previous = this.claimedAgentGraphSessionTails.get(sessionId) ?? Promise.resolve();
@@ -2984,7 +3017,8 @@ export class SessionManager {
         }
         enteredQueue = true;
         return this.runClaimedAgentGraphIntentOnce(input, runtimeExecution);
-      });
+      })
+      .finally(release);
     const tail = queuedExecution.then(
       () => {},
       () => {},
@@ -3001,7 +3035,10 @@ export class SessionManager {
       rejectStopped = reject;
     });
     const onRuntimeStop = (): void => {
-      if (!enteredQueue) rejectStopped(runtimeExecution.stopSignal.reason);
+      if (!enteredQueue) {
+        runtimeExecution.release();
+        rejectStopped(runtimeExecution.stopSignal.reason);
+      }
     };
     runtimeExecution.stopSignal.addEventListener('abort', onRuntimeStop, { once: true });
     if (runtimeExecution.stopSignal.aborted) onRuntimeStop();
@@ -3074,6 +3111,9 @@ export class SessionManager {
           claim,
           input.hostedGraphExecution,
         );
+        if (input.hostedGraphExecution && runtimeExecution.stopSignal.aborted)
+          throw runtimeExecution.stopSignal.reason;
+        input.onHostAdmission?.();
         await this.consumeLinkedRootExecution({
           sessionId: child.id,
           turnId: claim.targetTurnId,
@@ -3123,17 +3163,24 @@ export class SessionManager {
 
     // An invocation is what makes a Turn exist on the ledger, so the run listing
     // is the whole occupancy check: a Turn with durable content has one.
-    const turnOwner = (await this.listInvocations(child.id)).find(
-      (candidate) => candidate.turnId === claim.targetTurnId,
-    );
+    const invocations = await this.listInvocations(child.id);
+    const turnOwner = invocations.find((candidate) => candidate.turnId === claim.targetTurnId);
     if (turnOwner) {
       throw new Error(
         `Claimed graph turn ${claim.targetTurnId} is already owned by run ${turnOwner.runId}`,
       );
     }
-    if (child.isArchived || child.status === 'aborted') {
+    if (child.isArchived) {
       throw new Error('Claimed graph execution target child session is terminated');
     }
+    if (child.status === 'aborted') {
+      const latest = [...invocations].sort((left, right) => right.openedAt - left.openedAt)[0];
+      // A graph stop ends one activation, not the reusable operator. Only a
+      // durable graph-supervisor stop permits a new claim on an aborted child.
+      if (latest?.terminalEvent?.actions?.stateDelta?.abortSource !== 'graph.supervisor') {
+        throw new Error('Claimed graph execution target child session is terminated');
+      }
+    }
     if (input.abortSignal?.aborted) {
       throw new Error('Claimed graph execution was cancelled before runtime admission');
     }
@@ -3149,11 +3196,26 @@ export class SessionManager {
     let aborted = false;
     let stopPromise: Promise<void> | undefined;
     const userMessageId = await this.claimedGraphUserMessageId(claim, input.hostedGraphExecution);
+    if (input.hostedGraphExecution && runtimeExecution.stopSignal.aborted)
+      throw runtimeExecution.stopSignal.reason;
+    // Host runs this gate in its Session admission lease immediately before
+    // the durable admission. Until it passes, a graph stop only cancels the
+    // local capability and the gate refuses the Turn; after it passes, the
+    // stop must go through the Host fence.
+    const hostedAdmitExecution = input.hostedGraphExecution
+      ? async (): Promise<'executing' | 'cancelled'> => {
+          if (runtimeExecution.stopSignal.aborted) return 'cancelled';
+          const admission = admitExecution ? await admitExecution() : 'executing';
+          if (admission === 'cancelled' || runtimeExecution.stopSignal.aborted) return 'cancelled';
+          input.onHostAdmission?.();
+          return 'executing';
+        }
+      : admitExecution;
     const execution = this.consumeLinkedRootExecution({
       ...identity,
       userMessageId,
       execution: rootExecution,
-      ...(admitExecution ? { admitExecution } : {}),
+      ...(hostedAdmitExecution ? { admitExecution: hostedAdmitExecution } : {}),
       content: { text: input.prompt },
       start: ({ runId, userMessageId, onRunStarted }) =>
         this.sendMessage(
@@ -3895,6 +3957,75 @@ export class SessionManager {
     };
   }
 
+  /** Stop the captured graph activation, never a newer Run or a queued sibling. */
+  async stopAgentGraphActivation(
+    identity: RuntimeMessageRunIdentity,
+    input: StopSessionInput = { source: 'graph_supervisor' },
+  ): Promise<void> {
+    const authority = isRuntimeHostedRootAuthority(this.deps.messageAuthority)
+      ? this.deps.messageAuthority
+      : undefined;
+    // Until the Host admission gate passes, the capability is only local
+    // ownership: cancel it synchronously, and both the pre-submission check
+    // and that gate refuse the Turn. Once admitted, Host must commit its
+    // durable fence before Runtime stop.
+    if (authority) {
+      const admission = await this.deps.hostedAgentGraphExecution?.readRootTurnAdmissionIdentity(
+        identity.sessionId,
+        identity.turnId,
+      );
+      if (admission && admission.runId !== identity.runId) {
+        throw new Error('Graph stop identity does not match its durable admission');
+      }
+      // A recovered local claim can be new while the Host already owns its Run.
+      // Recheck the gate after the durable read closes that lookup race.
+      if (admission || this.findGraphRuntimeActivation(identity)?.hostAdmitted) {
+        await authority.stopRootRun(identity, input);
+        return;
+      }
+    }
+    await this.stopGraphRuntimeActivation(identity, input);
+  }
+
+  /** Whether a failed stop left cleanup retained for this exact graph activation. */
+  hasPendingAgentGraphActivationStop(identity: RuntimeMessageRunIdentity): boolean {
+    return this.runtimeKernel.hasPendingRunStop?.(identity) ?? false;
+  }
+
+  private findGraphRuntimeActivation(identity: RuntimeMessageRunIdentity) {
+    return [...this.claimedAgentGraphIntentRuns.values()]
+      .map((entry) => entry.owner)
+      .find(
+        (candidate) =>
+          candidate.identity.sessionId === identity.sessionId &&
+          candidate.identity.runId === identity.runId &&
+          candidate.identity.turnId === identity.turnId,
+      );
+  }
+
+  private stopGraphRuntimeActivation(
+    identity: RuntimeMessageRunIdentity,
+    input: StopSessionInput,
+  ): Promise<void> {
+    const owner = this.findGraphRuntimeActivation(identity);
+    if (owner) {
+      if (owner.stopTask) return owner.stopTask;
+      if (!this.runtimeKernel.stopExecution) {
+        throw new Error('Runtime kernel does not support scoped graph activation stops');
+      }
+      const task = this.runtimeKernel.stopExecution(owner.execution, input);
+      owner.stopTask = task;
+      void task.catch(() => {
+        if (owner.stopTask === task) owner.stopTask = undefined;
+      });
+      return task;
+    }
+    if (!this.runtimeKernel.stopRun) {
+      throw new Error('Runtime kernel does not support exact Run stops');
+    }
+    return this.runtimeKernel.stopRun(identity, input);
+  }
+
   async stopSession(sessionId: string, input: StopSessionInput = {}): Promise<void> {
     const hostedAuthority = isRuntimeHostedRootAuthority(this.deps.messageAuthority)
       ? this.deps.messageAuthority
@@ -3911,17 +4042,27 @@ export class SessionManager {
     );
   }
 
-  async deliverHostedRootStop(sessionId: string, input: StopSessionInput = {}): Promise<void> {
+  async deliverHostedRootStop(
+    sessionId: string,
+    input: StopSessionInput = {},
+    identity?: RuntimeMessageRunIdentity,
+  ): Promise<void> {
+    if (identity && identity.sessionId !== sessionId) {
+      throw new Error('Hosted stop identity belongs to another Session');
+    }
     const authority = isRuntimeHostedRootAuthority(this.deps.messageAuthority)
       ? this.deps.messageAuthority
       : undefined;
     await this.#stopSessionTree(
       sessionId,
-      this.runtimeKernel.stopSession(sessionId, input),
+      identity
+        ? this.stopGraphRuntimeActivation(identity, input)
+        : this.runtimeKernel.stopSession(sessionId, input),
       (childSessionId) =>
         authority
           ? authority.stopSession(childSessionId, input)
           : this.runtimeKernel.stopSession(childSessionId, input),
+      identity,
     );
   }
 
@@ -3929,6 +4070,7 @@ export class SessionManager {
     sessionId: string,
     ownStop: Promise<void>,
     stopChild: (childSessionId: string) => Promise<void>,
+    parentIdentity?: RuntimeMessageRunIdentity,
   ): Promise<void> {
     // Observe immediately while child lookup runs; await below still propagates the original error.
     void ownStop.catch(() => undefined);
@@ -3938,7 +4080,12 @@ export class SessionManager {
       const children = await this.listChildSessions(sessionId);
       childStops = await Promise.allSettled(
         children
-          .filter((child) => child.subagentParent?.lifecycle === 'foreground')
+          .filter(
+            (child) =>
+              child.subagentParent?.lifecycle === 'foreground' &&
+              (!parentIdentity ||
+                child.subagentParent.spawnedBy.parentTurnId === parentIdentity.turnId),
+          )
           .map((child) => stopChild(child.id)),
       );
     } catch (error) {
diff --git a/packages/runtime/src/stream-graph-coordinator.ts b/packages/runtime/src/stream-graph-coordinator.ts
index 6383024202..485c2113d9 100644
--- a/packages/runtime/src/stream-graph-coordinator.ts
+++ b/packages/runtime/src/stream-graph-coordinator.ts
@@ -106,6 +106,8 @@ export interface AgentGraphCoordinatorRuntime {
   provisionAgentGraphOperator: SessionManager['provisionAgentGraphOperator'];
   runClaimedAgentGraphIntent: SessionManager['runClaimedAgentGraphIntent'];
   stopSession: SessionManager['stopSession'];
+  stopAgentGraphActivation: SessionManager['stopAgentGraphActivation'];
+  hasPendingAgentGraphActivationStop?: SessionManager['hasPendingAgentGraphActivationStop'];
 }
 
 export interface AgentGraphCoordinatorInput {
@@ -185,6 +187,7 @@ interface GraphDriver {
   lastResult?: AgentGraphScheduleReconciliationResult;
   lastError?: unknown;
   yieldWaiters: Set<GraphYieldWaiter>;
+  scheduleChangeListeners: Set<() => void>;
   activityOperators: Map<
     string,
     Pick<AgentGraphClientOperator, 'status' | 'childSessionId'> & {
@@ -761,6 +764,7 @@ export class AgentGraphCoordinator {
     try {
       driver.lastError = undefined;
       driver.paused = false;
+      for (const listener of driver.scheduleChangeListeners) listener();
       this.#requestDrive(driver);
       // An epoch handover must not redirect this caller to the next driver.
       while (driver.task) await driver.task;
@@ -1064,6 +1068,10 @@ export class AgentGraphCoordinator {
         }),
       renderPrompt: this.#input.renderPrompt ?? renderAgentGraphScheduledWorkPrompt,
       abortSignal,
+      subscribeToScheduleChanges: (listener) => {
+        driver.scheduleChangeListeners.add(listener);
+        return () => driver.scheduleChangeListeners.delete(listener);
+      },
       supervisor: {
         onObservation: (observation) => {
           this.#queueClientProjectionUpdate(
@@ -1126,6 +1134,9 @@ export class AgentGraphCoordinator {
           void notify(this.#input.supervisor?.onRuntimeEvent, event);
         },
         onReconciliationFailure: (failure) => {
+          if (!failure.work && !failure.targetId) {
+            void notify(this.#input.onError, driver.rootSessionId, failure.error);
+          }
           this.#queueClientProjectionUpdate(driver, async () => {
             await this.#mergeReconciliationFailure(driver, failure);
             await notify(this.#input.onCheckpoint, driver.rootSessionId);
@@ -1285,17 +1296,19 @@ export class AgentGraphCoordinator {
     const successfulWorkIds = new Set(
       result.dispatches.map((dispatch) => dispatch.intent.readinessId),
     );
+    const stoppedTargetIds = new Set(result.stops.map((stop) => stop.targetId));
     const failuresByWorkId = new Map(
       existingFailures
-        .filter(
-          (failure) =>
-            requestedWorkIds.has(failure.workId) && !successfulWorkIds.has(failure.workId),
+        .filter((failure) =>
+          failure.phase === 'stop'
+            ? !stoppedTargetIds.has(failure.workId)
+            : requestedWorkIds.has(failure.workId) && !successfulWorkIds.has(failure.workId),
         )
         .map((failure) => [failure.workId, failure]),
     );
     for (const failure of result.failures) {
       const projected = durableReconciliationFailure(failure);
-      if (projected && requestedWorkIds.has(projected.workId)) {
+      if (projected && (projected.phase === 'stop' || requestedWorkIds.has(projected.workId))) {
         failuresByWorkId.set(projected.workId, projected);
       }
     }
@@ -1955,6 +1968,7 @@ export class AgentGraphCoordinator {
       pendingOutputDeltas: new Map(),
       runtimeFailureRunIds: new Set(),
       yieldWaiters: new Set(),
+      scheduleChangeListeners: new Set(),
       activityOperators: new Map(),
       activityRequestedIntentIds: new Set(),
       activityClaims: new Map(),
@@ -1976,6 +1990,7 @@ export class AgentGraphCoordinator {
       const driver = await this.#driver(rootSessionId);
       if (driver.stopping) return;
       driver.paused = false;
+      for (const listener of driver.scheduleChangeListeners) listener();
       this.#requestDrive(driver);
     } catch (error) {
       await notify(this.#input.onError, rootSessionId, error);
@@ -1993,6 +2008,7 @@ export class AgentGraphCoordinator {
       return;
     }
     driver.paused = false;
+    for (const listener of driver.scheduleChangeListeners) listener();
     this.#requestDrive(driver);
   }
 
diff --git a/packages/runtime/src/stream-graph-schedule-reconcile.ts b/packages/runtime/src/stream-graph-schedule-reconcile.ts
index 4d36c60aeb..939da331d0 100644
--- a/packages/runtime/src/stream-graph-schedule-reconcile.ts
+++ b/packages/runtime/src/stream-graph-schedule-reconcile.ts
@@ -33,6 +33,7 @@ import type {
   AgentGraphProvisionedEdge,
 } from '@maka/core/agent-graph-topology';
 import type { SessionEvent } from '@maka/core/events';
+import type { RuntimeMessageRunIdentity } from './message-authority.js';
 import { claimAgentGraphRunnableIntent } from './stream-graph-admission.js';
 import type {
   AgentGraphDispatchedActivation,
@@ -57,9 +58,26 @@ import { stableHash } from './request-shape.js';
 
 const MAX_RECONCILIATION_ATTEMPTS = 8;
 const SCHEDULE_INTENT_SCHEMA_VERSION = 1 as const;
+// A failed control pass can park the wave it is stopping, and nothing else may
+// wake it. Retry with a capped backoff while that wave is still running.
+const CONTROL_RETRY_INITIAL_DELAY_MS = 100;
+const CONTROL_RETRY_MAX_DELAY_MS = 5_000;
+// The eighth consecutive failure of a stop or a control read lands about 11 s
+// after the first, once the backoff has reached its cap: a fault that clears
+// during the ramp is never reported as stuck, while a persistent one is
+// surfaced instead of leaving the graph parked behind a single failure record.
+const CONTROL_STUCK_FAILURES = 8;
 
 export interface AgentGraphScheduleStopController {
-  stopSession(sessionId: string, input: { source: 'graph_supervisor' }): Promise<void>;
+  stopAgentGraphActivation(
+    identity: RuntimeMessageRunIdentity,
+    input: { source: 'graph_supervisor' },
+  ): Promise<void>;
+  /**
+   * Whether a failed stop left cleanup retained for this exact activation. A
+   * terminal activation without it resolves without another runtime stop.
+   */
+  hasPendingAgentGraphActivationStop?(identity: RuntimeMessageRunIdentity): boolean;
 }
 
 export interface RenderAgentGraphScheduledWorkPromptInput {
@@ -87,6 +105,7 @@ export interface ReconcileAgentGraphScheduleInput {
   ): Promise<readonly AgentGraphRecord[]>;
   renderPrompt(input: RenderAgentGraphScheduledWorkPromptInput): string | Promise<string>;
   abortSignal?: AbortSignal;
+  subscribeToScheduleChanges?(listener: () => void): () => void;
   supervisor?: AgentGraphSupervisorObserver;
 }
 
@@ -171,6 +190,42 @@ export async function reconcileAgentGraphSchedule(
     throw new Error('Agent graph maxNewActivations must be a non-negative safe integer');
   }
 
+  const wake = createReconciliationWake();
+  // Install before reading the durable schedule: a commit during any await
+  // remains pending until this reconciliation consumes it.
+  const unsubscribe = input.subscribeToScheduleChanges?.(wake.notify);
+  try {
+    return await reconcileSchedule(input, wake);
+  } finally {
+    unsubscribe?.();
+  }
+}
+
+function createReconciliationWake() {
+  let pending = false;
+  let resolve!: () => void;
+  let next = new Promise<void>((done) => {
+    resolve = done;
+  });
+  return {
+    notify() {
+      pending = true;
+      resolve();
+    },
+    async wait() {
+      if (!pending) await next;
+      pending = false;
+      next = new Promise<void>((done) => {
+        resolve = done;
+      });
+    },
+  };
+}
+
+async function reconcileSchedule(
+  input: ReconcileAgentGraphScheduleInput,
+  wake: ReturnType<typeof createReconciliationWake>,
+): Promise<AgentGraphScheduleReconciliationResult> {
   const processedIntentIds = new Set<string>();
   const dispatches: AgentGraphDispatchedActivation[] = [];
   const stops: AgentGraphScheduleStopResult[] = [];
@@ -179,6 +234,11 @@ export async function reconcileAgentGraphSchedule(
   // reconciliation; cache them per source graph so repeated snapshot reads do
   // not replay the full committed projection of a closed epoch.
   const selectedResultCache: SelectedResultCache = new Map();
+  // Exact stops that already succeeded in this reconciliation; later passes
+  // and wakes do not stop the same activation again.
+  const resolvedStops = new Set<string>();
+  // Consecutive stop failures per target, reset when the target resolves.
+  const stopFailureStreaks = new Map<string, number>();
   let newActivationCount = 0;
   let observedExistingActivationCount = 0;
   let snapshot = await readScheduleSnapshot(input, selectedResultCache);
@@ -197,9 +257,8 @@ export async function reconcileAgentGraphSchedule(
       );
     }
 
-    const stopWave = await applyScheduleStops(input, snapshot);
-    stops.push(...stopWave.stops);
-    for (const failure of stopWave.failures) recordReconciliationFailure(input, failures, failure);
+    const stopWave = await applyScheduleStops(input, snapshot, resolvedStops);
+    recordScheduleStops(input, stops, failures, stopFailureStreaks, stopWave);
     if (failures.length > 0) {
       snapshot = await readScheduleSnapshot(input, selectedResultCache);
       return reconciliationResult(
@@ -415,7 +474,8 @@ export async function reconcileAgentGraphSchedule(
       );
     }
 
-    const outcomes = await Promise.all(
+    let waveSettled = false;
+    const wave = Promise.all(
       prepared.map(async (work) => {
         const outcome = await dispatchScheduledWork(input, work, snapshot.schedule.revision);
         if (outcome.status === 'rejected') {
@@ -423,7 +483,74 @@ export async function reconcileAgentGraphSchedule(
         }
         return outcome;
       }),
-    );
+    ).finally(() => {
+      waveSettled = true;
+      wake.notify();
+    });
+    let controlReadFailure: AgentGraphScheduleReconciliationFailure | undefined;
+    let controlReadFailureStreak = 0;
+    let retryTimer: ReturnType<typeof setTimeout> | undefined;
+    let retryDelay = CONTROL_RETRY_INITIAL_DELAY_MS;
+    try {
+      // A driver abort does not end this loop: the wave still owns its
+      // dispatches, and a child parked behind a failed stop settles only
+      // after a retry succeeds.
+      while (!waveSettled) {
+        await wake.wait();
+        clearTimeout(retryTimer);
+        retryTimer = undefined;
+        if (waveSettled) break;
+        let retry: boolean;
+        try {
+          const latest = await readScheduleSnapshot(input, selectedResultCache);
+          const controls = await applyScheduleStops(input, latest, resolvedStops);
+          recordScheduleStops(input, stops, failures, stopFailureStreaks, controls);
+          if (controlReadFailure) {
+            failures.splice(failures.indexOf(controlReadFailure), 1);
+            controlReadFailure = undefined;
+          }
+          controlReadFailureStreak = 0;
+          retry = controls.failures.length > 0;
+        } catch (error) {
+          if (controlReadFailure) failures.splice(failures.indexOf(controlReadFailure), 1);
+          controlReadFailureStreak += 1;
+          // Like a stop, a read failure notifies on its first occurrence and,
+          // once, when it has persisted long enough to be stuck.
+          controlReadFailure =
+            controlReadFailureStreak < CONTROL_STUCK_FAILURES
+              ? { phase: 'schedule', error }
+              : {
+                  phase: 'schedule',
+                  error: new Error(
+                    `Reading graph schedule controls is stuck after ${controlReadFailureStreak} consecutive failures and is still retrying`,
+                    { cause: error },
+                  ),
+                };
+          failures.push(controlReadFailure);
+          if (
+            controlReadFailureStreak === 1 ||
+            controlReadFailureStreak === CONTROL_STUCK_FAILURES
+          ) {
+            notifySupervisor(input.supervisor?.onReconciliationFailure, controlReadFailure);
+          }
+          retry = true;
+        }
+        // A failed stop can keep its child, and so this wave, parked until a
+        // retry succeeds. Back off instead of waiting for another commit.
+        if (retry) {
+          retryTimer = setTimeout(wake.notify, retryDelay);
+          retryDelay = Math.min(retryDelay * 2, CONTROL_RETRY_MAX_DELAY_MS);
+        } else {
+          retryDelay = CONTROL_RETRY_INITIAL_DELAY_MS;
+        }
+      }
+    } finally {
+      clearTimeout(retryTimer);
+      // This wave remains the sole owner of all dispatched executions, even
+      // if a control-store read fails while processing a schedule wake.
+      await wave;
+    }
+    const outcomes = await wave;
     let stale = false;
     for (const outcome of outcomes) {
       if (outcome.status === 'stale') {
@@ -631,9 +758,59 @@ function selectedResultKey(selected: AgentGraphSelectedResultInput): string {
   return `${selected.sourceGraphId}\u0000${selected.resultId}`;
 }
 
+function recordScheduleStops(
+  input: ReconcileAgentGraphScheduleInput,
+  stops: AgentGraphScheduleStopResult[],
+  failures: AgentGraphScheduleReconciliationFailure[],
+  streaks: Map<string, number>,
+  controls: {
+    stops: AgentGraphScheduleStopResult[];
+    failures: AgentGraphScheduleReconciliationFailure[];
+  },
+): void {
+  stops.push(...controls.stops);
+  const resolvedTargets = new Set(controls.stops.map((stop) => stop.targetId));
+  for (const targetId of resolvedTargets) streaks.delete(targetId);
+  for (let index = failures.length - 1; index >= 0; index -= 1) {
+    const failure = failures[index]!;
+    if (failure.phase === 'stop' && failure.targetId && resolvedTargets.has(failure.targetId)) {
+      failures.splice(index, 1);
+    }
+  }
+  for (const failure of controls.failures) {
+    const targetId = failure.targetId ?? '';
+    const streak = (streaks.get(targetId) ?? 0) + 1;
+    streaks.set(targetId, streak);
+    // The stuck state travels in the message: supervisor notifications are
+    // structured clones and the durable client failure keeps only the reason.
+    const recorded: AgentGraphScheduleReconciliationFailure =
+      streak < CONTROL_STUCK_FAILURES
+        ? failure
+        : {
+            ...failure,
+            error: new Error(
+              `Stop for graph target ${targetId} is stuck after ${streak} consecutive failures and is still retrying`,
+              { cause: failure.error },
+            ),
+          };
+    // A retried stop that fails again replaces its earlier failure. The
+    // supervisor hears about the first failure and, once, about a stuck stop,
+    // not about every backoff attempt.
+    const previous = failures.findIndex(
+      (existing) => existing.phase === 'stop' && existing.targetId === failure.targetId,
+    );
+    if (previous >= 0) failures[previous] = recorded;
+    else failures.push(recorded);
+    if (streak === 1 || streak === CONTROL_STUCK_FAILURES) {
+      notifySupervisor(input.supervisor?.onReconciliationFailure, recorded);
+    }
+  }
+}
+
 async function applyScheduleStops(
   input: ReconcileAgentGraphScheduleInput,
   snapshot: ScheduleSnapshot,
+  resolvedStops: Set<string>,
 ): Promise<{
   stops: AgentGraphScheduleStopResult[];
   failures: AgentGraphScheduleReconciliationFailure[];
@@ -653,10 +830,12 @@ async function applyScheduleStops(
   const claimsByIntent = new Map(snapshot.claims.map((claim) => [claim.intentId, claim]));
   const activationTargets = graphActivationTargets(snapshot.observation);
   const immediate: AgentGraphScheduleStopResult[] = [];
-  const targetsBySession = new Map<
-    string,
-    Array<{ targetId: string; reason: string; activationId: string }>
-  >();
+  const targets: Array<{
+    targetId: string;
+    reason: string;
+    identity: RuntimeMessageRunIdentity;
+    status: 'stopped' | 'already_terminal';
+  }> = [];
   const failures: AgentGraphScheduleReconciliationFailure[] = [];
 
   for (const [targetId, reason] of requests) {
@@ -668,13 +847,28 @@ async function applyScheduleStops(
       ? activationTargets.get(claim.targetRunId)
       : activationTargets.get(targetId);
     if (activation && isTerminalActivationStatus(activation.status)) {
-      immediate.push({
-        targetId,
-        reason,
-        status: 'already_terminal',
+      const identity = {
         sessionId: activation.sessionId,
-        activationId: activation.activationId,
-      });
+        runId: activation.activationId,
+        turnId: activation.turnId,
+      };
+      // A durable terminal does not prove that backend cleanup succeeded.
+      // Retry the exact stop only while a failed stop retains cleanup for this
+      // Run; a settled historical target resolves without a runtime call.
+      if (
+        resolvedStops.has(stopTargetKey(targetId, identity)) ||
+        !input.stopController.hasPendingAgentGraphActivationStop?.(identity)
+      ) {
+        immediate.push({
+          targetId,
+          reason,
+          status: 'already_terminal',
+          sessionId: activation.sessionId,
+          activationId: activation.activationId,
+        });
+        continue;
+      }
+      targets.push({ targetId, reason, status: 'already_terminal', identity });
       continue;
     }
     if (work && claim) {
@@ -707,44 +901,43 @@ async function applyScheduleStops(
     }
     const sessionId = activation?.sessionId ?? claim!.targetSessionId;
     const activationId = activation?.activationId ?? claim!.targetRunId;
-    const sessionTargets = targetsBySession.get(sessionId) ?? [];
-    sessionTargets.push({
-      targetId,
-      reason,
-      activationId,
-    });
-    targetsBySession.set(sessionId, sessionTargets);
+    const turnId = activation?.turnId ?? claim!.targetTurnId;
+    const identity = { sessionId, runId: activationId, turnId };
+    if (resolvedStops.has(stopTargetKey(targetId, identity))) {
+      immediate.push({ targetId, reason, status: 'stopped', sessionId, activationId });
+      continue;
+    }
+    targets.push({ targetId, reason, status: 'stopped', identity });
   }
 
   const settled = await Promise.allSettled(
-    [...targetsBySession].map(async ([sessionId, targets]) => {
-      await input.stopController.stopSession(sessionId, { source: 'graph_supervisor' });
-      return targets.map(
-        (target): AgentGraphScheduleStopResult => ({
-          targetId: target.targetId,
-          reason: target.reason,
-          status: 'stopped',
-          sessionId,
-          activationId: target.activationId,
-        }),
-      );
+    targets.map(async (target): Promise<AgentGraphScheduleStopResult> => {
+      await input.stopController.stopAgentGraphActivation(target.identity, {
+        source: 'graph_supervisor',
+      });
+      return {
+        targetId: target.targetId,
+        reason: target.reason,
+        status: target.status,
+        sessionId: target.identity.sessionId,
+        activationId: target.identity.runId,
+      };
     }),
   );
   settled.forEach((result, index) => {
+    const target = targets[index]!;
     if (result.status === 'fulfilled') {
-      immediate.push(...result.value);
+      resolvedStops.add(stopTargetKey(target.targetId, target.identity));
+      immediate.push(result.value);
       return;
     }
-    const [sessionId, targets] = [...targetsBySession][index]!;
-    targets.forEach((target) => {
-      failures.push({
-        phase: 'stop',
-        targetId: target.targetId,
-        error: new Error(
-          `Failed to stop graph supervisor target ${target.targetId} in ${sessionId}`,
-          { cause: result.reason },
-        ),
-      });
+    failures.push({
+      phase: 'stop',
+      targetId: target.targetId,
+      error: new Error(
+        `Failed to stop graph supervisor target ${target.targetId} in ${target.identity.sessionId}`,
+        { cause: result.reason },
+      ),
     });
   });
   return {
@@ -753,6 +946,10 @@ async function applyScheduleStops(
   };
 }
 
+function stopTargetKey(targetId: string, identity: RuntimeMessageRunIdentity): string {
+  return [targetId, identity.sessionId, identity.runId, identity.turnId].join('\u0000');
+}
+
 async function dispatchScheduledWork(
   input: ReconcileAgentGraphScheduleInput,
   prepared: PreparedWork,
@@ -1062,6 +1259,7 @@ function graphActivationTargets(observation: AgentGraphSupervisorObservation): M
   {
     sessionId: string;
     activationId: string;
+    turnId: string;
     status: string;
   }
 > {
@@ -1070,9 +1268,13 @@ function graphActivationTargets(observation: AgentGraphSupervisorObservation): M
     {
       sessionId: string;
       activationId: string;
+      turnId: string;
       status: string;
     }
   >();
+  const sourceByActivation = new Map(
+    observation.projection.records.map((record) => [record.activationId, record.source]),
+  );
   for (const binding of observation.projection.operators) {
     const state = observation.projection.state.operators[binding.operatorId];
     for (const activation of Object.values(state?.activations ?? {})) {
@@ -1081,9 +1283,12 @@ function graphActivationTargets(observation: AgentGraphSupervisorObservation): M
           `Graph activation ${activation.activationId} belongs to multiple operators`,
         );
       }
+      const source = sourceByActivation.get(activation.activationId);
+      if (!source) throw new Error(`Graph activation ${activation.activationId} has no source`);
       targets.set(activation.activationId, {
         sessionId: binding.sessionId,
         activationId: activation.activationId,
+        turnId: source.turnId,
         status: activation.status,
       });
     }