feat(delegation): skip the wake for a completion the parent already read

apps/server/src/mcp/toolkits/delegation/handlers.test.ts

@@ -680,6 +680,45 @@ describe("delegated_thread_status", () => {
       });
     }),
   );
+
+  it.effect(
+    "marks delegation observation consumed on a completed child and skips when timed out",
+    () =>
+      Effect.gen(function* () {
+        const completedChild = makeShell(CHILD_ID, {
+          latestTurn: completedTurn({ turnId: TurnId.make("turn-2") }),
+        });
+        const completedHarness = yield* makeHarness({
+          shells: [makeShell(PARENT_ID), completedChild],
+        });
+        yield* completedHarness.call("delegated_thread_status", { delegationKey: "k1" });
+        const completedCommands = yield* Ref.get(completedHarness.commands);
+        expect(completedCommands).toHaveLength(1);
+        expect(completedCommands[0]).toMatchObject({
+          type: "thread.activity.append",
+          threadId: PARENT_ID,
+          activity: {
+            kind: "delegation.child-state",
+            payload: {
+              baseline: true,
+              childThreadId: CHILD_ID,
+            },
+          },
+        });
+
+        const running = makeShell(CHILD_ID, { latestTurn: turn() });
+        const runningHarness = yield* makeHarness({
+          shells: [makeShell(PARENT_ID), running],
+        });
+        const runningFiber = yield* Effect.forkChild(
+          runningHarness.call("delegated_thread_status", { delegationKey: "k1", waitSeconds: 5 }),
+        );
+        yield* TestClock.adjust("5 seconds");
+        yield* Fiber.join(runningFiber);
+        const runningCommands = yield* Ref.get(runningHarness.commands);
+        expect(runningCommands).toHaveLength(0);
+      }),
+  );
 });
 
 describe("delegated_thread_result", () => {

apps/server/src/mcp/toolkits/delegation/handlers.ts

@@ -34,6 +34,7 @@ import { ServerConfig } from "../../../config.ts";
 import * as GitWorkflowService from "../../../git/GitWorkflowService.ts";
 import type { OrchestrationDispatchError } from "../../../orchestration/Errors.ts";
 import * as OrchestrationEngine from "../../../orchestration/Services/OrchestrationEngine.ts";
+import { markDelegationObservationConsumed } from "../../../orchestration/delegationObservationConsumed.ts";
 import * as ProjectionSnapshotQuery from "../../../orchestration/Services/ProjectionSnapshotQuery.ts";
 import * as ThreadDeletionReactor from "../../../orchestration/Services/ThreadDeletionReactor.ts";
 import * as ProviderRegistry from "../../../provider/Services/ProviderRegistry.ts";
@@ -321,7 +322,7 @@ const make = Effect.gen(function* () {
     readonly waitSeconds?: number | undefined;
   }) =>
     Effect.gen(function* () {
-      const { childId, shell } = yield* lookupChild(input.delegationKey);
+      const { scope, childId, shell } = yield* lookupChild(input.delegationKey);
       const initial = statusOf(shell);
       // A settled child or an actionable blocker needs attention now, not a
       // state transition. In particular, do not spend the entire wait budget
@@ -331,6 +332,7 @@ const make = Effect.gen(function* () {
         initial.hasPendingApprovals ||
         initial.hasPendingUserInput
       ) {
+        yield* markDelegationObservationConsumed({ parentId: scope.threadId, child: shell });
         return statusPayload(input.delegationKey, shell, 0, false);
       }
       const startedAt = yield* Clock.currentTimeMillis;
@@ -354,9 +356,11 @@ const make = Effect.gen(function* () {
           latest.hasPendingApprovals !== initial.hasPendingApprovals ||
           latest.hasPendingUserInput !== initial.hasPendingUserInput
         ) {
+          yield* markDelegationObservationConsumed({ parentId: scope.threadId, child: current });
           return statusPayload(input.delegationKey, current, elapsedSeconds(now), true);
         }
       }
+      yield* markDelegationObservationConsumed({ parentId: scope.threadId, child: current });
       return statusPayload(input.delegationKey, current, elapsedSeconds(now), false);
     });
 
@@ -365,8 +369,9 @@ const make = Effect.gen(function* () {
     readonly maxChars?: number | undefined;
   }) =>
     Effect.gen(function* () {
-      const { childId, shell } = yield* lookupChild(input.delegationKey);
+      const { scope, childId, shell } = yield* lookupChild(input.delegationKey);
       const state = deriveDelegatedThreadState(shell);
+      yield* markDelegationObservationConsumed({ parentId: scope.threadId, child: shell });
       // The detail query serves active threads only, so an archived child has no body here.
       const detail =
         state === "archived"

apps/server/src/mcp/toolkits/pair/handlers.test.ts

@@ -694,6 +694,41 @@ describe("pair_handoff", () => {
 });
 
 describe("pair_await", () => {
+  it.effect(
+    "marks delegation observation consumed on a completed executor and skips when timed out",
+    () =>
+      Effect.gen(function* () {
+        const completedHarness = yield* makeHarness({
+          shells: [makeShell(LEAD_ID), finished],
+          details: [finishedDetail],
+        });
+        yield* completedHarness.call("pair_await", { maxSeconds: 10 });
+        const completedCommands = yield* Ref.get(completedHarness.commands);
+        expect(completedCommands).toHaveLength(1);
+        expect(completedCommands[0]).toMatchObject({
+          type: "thread.activity.append",
+          threadId: LEAD_ID,
+          activity: {
+            kind: "delegation.child-state",
+            payload: {
+              baseline: true,
+              childThreadId: EXECUTOR_ID,
+            },
+          },
+        });
+
+        const runningHarness = yield* makeHarness({
+          shells: [makeShell(LEAD_ID), runningExecutor()],
+        });
+        const runningFiber = yield* Effect.forkChild(
+          runningHarness.call("pair_await", { maxSeconds: 5 }),
+        );
+        yield* TestClock.adjust("5 seconds");
+        yield* Fiber.join(runningFiber);
+        const runningCommands = yield* Ref.get(runningHarness.commands);
+        expect(runningCommands).toHaveLength(0);
+      }),
+  );
   const finished = makeExecutor({
     latestTurn: completedTurn({ assistantMessageId: MessageId.make("a-1") }),
   });
@@ -1054,7 +1089,7 @@ describe("protected paths", () => {
       expect(yield* harness.call("pair_await", {})).toMatchObject({
         protectedPaths: { checked: 2, changed: ["src/a.test.ts"] },
       });
-      expect(yield* harness.commandTypes).toEqual(["thread.turn.start"]);
+      expect(yield* harness.commandTypes).toEqual(["thread.turn.start", "thread.activity.append"]);
     }),
   );
 

apps/server/src/mcp/toolkits/pair/handlers.ts

@@ -30,6 +30,7 @@ import * as Semaphore from "effect/Semaphore";
 
 import type { OrchestrationDispatchError } from "../../../orchestration/Errors.ts";
 import * as OrchestrationEngine from "../../../orchestration/Services/OrchestrationEngine.ts";
+import { markDelegationObservationConsumed } from "../../../orchestration/delegationObservationConsumed.ts";
 import * as ProjectionSnapshotQuery from "../../../orchestration/Services/ProjectionSnapshotQuery.ts";
 import * as ProviderRegistry from "../../../provider/Services/ProviderRegistry.ts";
 import { PAIR_LEAD_PROTOCOL } from "../../../provider/RuntimeInstructions.ts";
@@ -575,6 +576,13 @@ const make = Effect.gen(function* () {
         }
       }
 
+      if (state !== "running") {
+        yield* markDelegationObservationConsumed({
+          parentId: scope.threadId,
+          child: current,
+        });
+      }
+
       const result: PairAwaitResult = {
         threadId: executorId,
         state,

apps/server/src/orchestration/DelegationFollowThroughReactor.test.ts

@@ -372,6 +372,34 @@ describe("DelegationFollowThroughReactor", () => {
       }),
     ),
   );
+  it.effect("does not wake a parent for a completion a tool already handed it in-turn", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        // The child finishes while the parent is mid-turn, so the reactor records
+        // the observation without delivering it.
+        const busy = yield* makeHarness([shell(PARENT, true), shell(CHILD, true)]);
+        busy.replace(shell(CHILD));
+        yield* busy.emit(CHILD);
+        assert.strictEqual(busy.wakes().length, 0);
+        // pair_await or delegated_thread_status then returns that completion to the
+        // parent and rewrites the same receipt with baseline: true.
+        const consumed = {
+          ...busy.persisted,
+          activities: busy.persisted.activities.map((activity) =>
+            activity.kind === "delegation.child-state"
+              ? { ...activity, payload: { ...Object(activity.payload), baseline: true } }
+              : activity,
+          ),
+        };
+        assert.isTrue(
+          consumed.activities.some((activity) => activity.kind === "delegation.child-state"),
+        );
+        const idle = yield* makeHarness([shell(PARENT), shell(CHILD)], true, consumed);
+        yield* idle.emit(PARENT);
+        assert.strictEqual(idle.wakes().length, 0);
+      }),
+    ),
+  );
   it.effect(
     "recovers a persisted pending observation and respects a persisted delivery receipt",
     () =>

apps/server/src/orchestration/DelegationFollowThroughReactor.ts

@@ -24,13 +24,13 @@ import { isChildOfParent } from "../mcp/toolkits/delegation/logic.ts";
 import * as OrchestrationEngine from "./Services/OrchestrationEngine.ts";
 import * as ProjectionSnapshotQuery from "./Services/ProjectionSnapshotQuery.ts";
 import {
+  DELEGATION_OBSERVED_ACTIVITY_KIND,
+  delegationObservationReceipt,
   observeDelegatedChild,
   isActionableDelegationObservation,
   isDelegationParentEligible,
   formatDelegationFollowThroughPrompt,
 } from "./delegationFollowThrough.logic.ts";
-
-const OBSERVED = "delegation.child-state";
 const DELIVERED = "delegation.follow-through.delivered";
 const PAUSED = "delegation.follow-through.paused";
 const MESSAGE_PREFIX = "delegation-follow-through:";
@@ -124,7 +124,7 @@ export const make = Effect.gen(function* () {
     const children = snapshot.threads.filter((child) => isChildOfParent(child.id, parent.id));
     if (children.length === 0) return;
     const detailOption = yield* snapshots.getThreadDetailById(parent.id, {
-      activityKinds: [OBSERVED, DELIVERED, PAUSED],
+      activityKinds: [DELEGATION_OBSERVED_ACTIVITY_KIND, DELIVERED, PAUSED],
     });
     if (Option.isNone(detailOption)) return;
     const detail = detailOption.value;
@@ -135,7 +135,7 @@ export const make = Effect.gen(function* () {
       childThreadIds: children.map((child) => child.id),
     });
     for (const activity of [...detail.activities, ...observations]) {
-      if (activity.kind === OBSERVED) {
+      if (activity.kind === DELEGATION_OBSERVED_ACTIVITY_KIND) {
         const decoded = decodeObservation(activity.payload);
         if (Option.isSome(decoded)) previous.set(decoded.value.childThreadId, decoded.value);
       } else if (activity.kind === DELIVERED) {
@@ -177,9 +177,13 @@ export const make = Effect.gen(function* () {
           : (startupPass || !old) &&
             work.liveChildId !== child.id &&
             isActionableDelegationObservation(observation);
-      const notificationId = EventId.make(
-        `delegation-notice:${yield* digest(observation.noticeKey)}`,
-      );
+      const noticeDigest = yield* digest(observation.noticeKey);
+      const receipt = delegationObservationReceipt({
+        observation,
+        noticeDigest,
+        baseline,
+      });
+      const notificationId = EventId.make(receipt.payload.notificationId);
       if (old?.noticeKey !== observation.noticeKey) {
         yield* engine.dispatch({
           type: "thread.activity.append",
@@ -188,16 +192,11 @@ export const make = Effect.gen(function* () {
           ),
           threadId: parent.id,
           activity: {
-            id: EventId.make(`delegation-observation:${child.id}`),
-            kind: OBSERVED,
+            id: EventId.make(receipt.activityId),
+            kind: DELEGATION_OBSERVED_ACTIVITY_KIND,
             tone: "info",
             summary: `Pylon child ${observation.phase}`,
-            payload: {
-              childThreadId: child.id,
-              noticeKey: observation.noticeKey,
-              notificationId,
-              baseline,
-            },
+            payload: receipt.payload,
             turnId: null,
             createdAt: now,
           },

apps/server/src/orchestration/delegationFollowThrough.logic.test.ts

@@ -9,6 +9,9 @@ import {
 import { describe, expect, it } from "vite-plus/test";
 import { deriveDelegatedThreadState } from "../mcp/toolkits/delegation/logic.ts";
 import {
+  DELEGATION_OBSERVED_ACTIVITY_KIND,
+  consumedDelegationObservation,
+  delegationObservationReceipt,
   observeDelegatedChild,
   isActionableDelegationObservation,
   isDelegationParentEligible,
@@ -217,3 +220,39 @@ describe("parent follow-through eligibility", () => {
       );
   });
 });
+
+describe("an observation a parent read in-turn", () => {
+  it("counts only a finished attempt, so a child that needs the user still wakes its parent", () => {
+    expect(consumedDelegationObservation(shell())).toEqual(observeDelegatedChild(shell()));
+    for (const state of ["error", "interrupted"] as const) {
+      const finished = shell({ latestTurn: { ...turn, state } });
+      expect(consumedDelegationObservation(finished)).toEqual(observeDelegatedChild(finished));
+    }
+    expect(consumedDelegationObservation(shell({ hasPendingApprovals: true }))).toBeNull();
+    expect(consumedDelegationObservation(shell({ hasPendingUserInput: true }))).toBeNull();
+    expect(
+      consumedDelegationObservation(
+        shell({ session: { ...session, status: "running", activeTurnId: turn.turnId } }),
+      ),
+    ).toBeNull();
+    expect(consumedDelegationObservation(shell({ latestTurn: null }))).toBeNull();
+    expect(consumedDelegationObservation(shell({ archivedAt: timestamp }))).toBeNull();
+  });
+
+  it("writes the receipt the reactor reads, under the id that replaces the last one", () => {
+    const observation = observeDelegatedChild(shell());
+    if (observation === null) throw new Error("expected an observation");
+    expect(DELEGATION_OBSERVED_ACTIVITY_KIND).toBe("delegation.child-state");
+    expect(
+      delegationObservationReceipt({ observation, noticeDigest: "abc123", baseline: true }),
+    ).toEqual({
+      activityId: `delegation-observation:${id}`,
+      payload: {
+        childThreadId: id,
+        noticeKey: observation.noticeKey,
+        notificationId: "delegation-notice:abc123",
+        baseline: true,
+      },
+    });
+  });
+});

apps/server/src/orchestration/delegationFollowThrough.logic.ts

@@ -68,6 +68,59 @@ export function isActionableDelegationObservation(observation: DelegationObserva
   return observation.phase !== "running";
 }
 
+/** The activity kind that records what the follow-through reactor last saw of a child. */
+export const DELEGATION_OBSERVED_ACTIVITY_KIND = "delegation.child-state";
+
+/**
+ * What a parent has already read when a tool hands it a child's state in-turn,
+ * or null when waking it later is still right. Only a finished attempt counts:
+ * a child that needs the user keeps its wake, and so does one still running.
+ */
+export function consumedDelegationObservation(
+  shell: OrchestrationThreadShell,
+): DelegationObservation | null {
+  const observation = observeDelegatedChild(shell);
+  if (observation === null) return null;
+  if (
+    observation.phase === "completed" ||
+    observation.phase === "error" ||
+    observation.phase === "interrupted"
+  ) {
+    return observation;
+  }
+  return null;
+}
+
+/**
+ * The receipt both the reactor and the tools write for a child, keyed so the
+ * newest one replaces the last. `baseline: true` tells the reactor the parent
+ * needs no wake for this notice. `noticeDigest` is the hex SHA-256 of the
+ * observation's `noticeKey`.
+ */
+export function delegationObservationReceipt(input: {
+  readonly observation: DelegationObservation;
+  readonly noticeDigest: string;
+  readonly baseline: boolean;
+}): {
+  readonly activityId: string;
+  readonly payload: {
+    readonly childThreadId: string;
+    readonly noticeKey: string;
+    readonly notificationId: string;
+    readonly baseline: boolean;
+  };
+} {
+  return {
+    activityId: `delegation-observation:${input.observation.childThreadId}`,
+    payload: {
+      childThreadId: input.observation.childThreadId,
+      noticeKey: input.observation.noticeKey,
+      notificationId: `delegation-notice:${input.noticeDigest}`,
+      baseline: input.baseline,
+    },
+  };
+}
+
 /** No new turn should bypass deliberate stops, user blockers, or owned transitions. */
 export function isDelegationParentEligible(shell: OrchestrationThreadShell, nowMs: number) {
   const session = shell.session;