Merge pull request #137 from pylon-code/feat/plan-progress-preview

feat(progress): show honest waits and delegated work

Author: rynfar

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts

@@ -2380,6 +2380,12 @@ projectionSnapshotLayer("ProjectionSnapshotQuery windowed thread detail", (it) =
           activity_id, thread_id, turn_id, tone, kind, summary, payload_json, sequence, created_at
         )
         VALUES
+          (
+            'plan-old', 'thread-w', 'turn-5', 'info', 'turn.plan.updated',
+            'Waiting on external system',
+            '{"plan":[{"step":"Await CI","status":"waiting","waitingOn":"external"}]}', NULL,
+            '2026-03-01T00:00:00.500Z'
+          ),
           (
             'approval-old', 'thread-w', NULL, 'approval', 'approval.requested',
             'Approve old command', '{"requestId":"approval-1"}', NULL,
@@ -2429,25 +2435,31 @@ projectionSnapshotLayer("ProjectionSnapshotQuery windowed thread detail", (it) =
       const detailWithPinnedRequests = yield* snapshotQuery.getThreadDetailById(threadW);
       assert.equal(detailWithPinnedRequests._tag, "Some");
       if (detailWithPinnedRequests._tag === "Some") {
-        const ids = detailWithPinnedRequests.value.activities.map((activity) => activity.id);
-        assert.equal(detailWithPinnedRequests.value.activities.length, 503);
-        assert.equal(ids.includes(asEventId("approval-old")), true);
-        assert.equal(ids.includes(asEventId("user-input-old")), true);
-        assert.equal(ids.includes(asEventId("user-input-closed")), false);
-        assert.equal(ids.includes(asEventId("user-input-tied-z-request")), true);
+        const ids = new Set(
+          detailWithPinnedRequests.value.activities.map((activity) => activity.id),
+        );
+        assert.equal(detailWithPinnedRequests.value.activities.length, 504);
+        assert.equal(ids.has(asEventId("plan-old")), true);
+        assert.equal(ids.has(asEventId("approval-old")), true);
+        assert.equal(ids.has(asEventId("user-input-old")), true);
+        assert.equal(ids.has(asEventId("user-input-closed")), false);
+        assert.equal(ids.has(asEventId("user-input-tied-z-request")), true);
       }
 
       const windowWithPinnedRequests = yield* snapshotQuery.getThreadDetailSnapshot(threadW, {
         turnLimit: 2,
       });
       assert.equal(windowWithPinnedRequests._tag, "Some");
       if (windowWithPinnedRequests._tag === "Some") {
-        const ids = windowWithPinnedRequests.value.thread.activities.map((activity) => activity.id);
-        assert.equal(windowWithPinnedRequests.value.thread.activities.length, 503);
-        assert.equal(ids.includes(asEventId("approval-old")), true);
-        assert.equal(ids.includes(asEventId("user-input-old")), true);
-        assert.equal(ids.includes(asEventId("user-input-closed")), false);
-        assert.equal(ids.includes(asEventId("user-input-tied-z-request")), true);
+        const ids = new Set(
+          windowWithPinnedRequests.value.thread.activities.map((activity) => activity.id),
+        );
+        assert.equal(windowWithPinnedRequests.value.thread.activities.length, 504);
+        assert.equal(ids.has(asEventId("plan-old")), true);
+        assert.equal(ids.has(asEventId("approval-old")), true);
+        assert.equal(ids.has(asEventId("user-input-old")), true);
+        assert.equal(ids.has(asEventId("user-input-closed")), false);
+        assert.equal(ids.has(asEventId("user-input-tied-z-request")), true);
       }
     }),
   );

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts

@@ -1287,9 +1287,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
       `,
   });
 
-  // Blocking request payloads must remain available even if they predate the
-  // recent activity window. Each CTE returns at most one unresolved row per
-  // request, so the merge below stays bounded by actionable work.
+  // Blocking request payloads and the latest turn's authoritative plan must
+  // remain available even if they predate the recent activity window. Each
+  // CTE returns bounded state, so the merge below cannot grow with history.
   const listPinnedThreadActivityRowsByThread = SqlSchema.findAll({
     Request: ThreadIdLookupInput,
     Result: ProjectionThreadActivityDbRowSchema,
@@ -1349,6 +1349,20 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
             )
             AND json_extract(activity.payload_json, '$.requestId') IS NOT NULL
         ),
+        latest_plan_activity AS (
+          SELECT activity.activity_id
+          FROM projection_thread_activities AS activity
+          INNER JOIN projection_threads AS thread
+            ON thread.thread_id = activity.thread_id
+          WHERE activity.thread_id = ${threadId}
+            AND activity.turn_id = thread.latest_turn_id
+            AND activity.kind = 'turn.plan.updated'
+          ORDER BY
+            activity.sequence DESC,
+            activity.created_at DESC,
+            activity.activity_id DESC
+          LIMIT 1
+        ),
         pinned_activity_ids AS (
           SELECT activity_id
           FROM pending_approval_activities
@@ -1358,6 +1372,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
           FROM user_input_lifecycle
           WHERE request_order = 1
             AND kind = 'user-input.requested'
+          UNION ALL
+          SELECT activity_id
+          FROM latest_plan_activity
         )
         SELECT
           activity.activity_id AS "activityId",

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

@@ -24,7 +24,7 @@ describe("ThreadPlanProgress", () => {
     expect(progress.getThreadPlanProgress(threadId)).toBeNull();
   });
 
-  it("falls back to the first non-completed step when nothing is in progress", () => {
+  it("falls back to the first pending step when nothing is active", () => {
     const progress = ThreadPlanProgress.make();
     const threadId = "t-plan-2";
     progress.recordPlanProgress(threadId, [
@@ -34,6 +34,20 @@ describe("ThreadPlanProgress", () => {
     expect(progress.getThreadPlanProgress(threadId)?.step).toBe("First");
   });
 
+  it("surfaces an explicit wait before pending work", () => {
+    const progress = ThreadPlanProgress.make();
+    const threadId = "t-plan-wait";
+    progress.recordPlanProgress(threadId, [
+      { step: "Pending setup", status: "pending" },
+      { step: "Await review", status: "waiting" },
+    ]);
+    expect(progress.getThreadPlanProgress(threadId)).toEqual({
+      step: "Await review",
+      completedSteps: 0,
+      totalSteps: 2,
+    });
+  });
+
   it("clearThreadPlanProgress removes the entry (turn settled / session died)", () => {
     const progress = ThreadPlanProgress.make();
     const threadId = "t-plan-3";

apps/server/src/orchestration/ThreadPlanProgress.ts

@@ -49,11 +49,12 @@ export function make(): ThreadPlanProgressService["Service"] {
     recordPlanProgress: (threadId, plan) => {
       const totalSteps = plan.length;
       const completedSteps = plan.filter((step) => step.status === "completed").length;
-      // Current step: the in-progress one, else the first pending one (a
-      // plan that was just written has no in-progress step yet).
+      // Current step: active work or an explicit wait, else the first
+      // pending outcome (a plan that was just written has no active step yet).
       const current =
         plan.find((step) => step.status === "inProgress") ??
-        plan.find((step) => step.status !== "completed");
+        plan.find((step) => step.status === "waiting") ??
+        plan.find((step) => step.status === "pending");
       if (totalSteps === 0 || completedSteps === totalSteps || current === undefined) {
         progressByThreadId.delete(threadId);
         return;

apps/server/src/provider/prime/PrimeAgentDaemonEvents.test.ts

@@ -255,7 +255,8 @@ describe("PrimeAgentDaemonEvents", () => {
         explanation: "  Safe task parity  ",
         plan: [
           { step: "  Inspect behavior  ", status: "completed" },
-          { step: "Project plan", status: "inProgress" },
+          { step: "Project plan", status: "pending" },
+          { step: "Await delegated review", status: "waiting", waitingOn: "delegates" },
         ],
       },
       isError: false,
@@ -272,7 +273,8 @@ describe("PrimeAgentDaemonEvents", () => {
           explanation: "Safe task parity",
           plan: [
             { step: "Inspect behavior", status: "completed" },
-            { step: "Project plan", status: "inProgress" },
+            { step: "Project plan", status: "pending" },
+            { step: "Await delegated review", status: "waiting", waitingOn: "delegates" },
           ],
         },
       },
@@ -335,6 +337,33 @@ describe("PrimeAgentDaemonEvents", () => {
           plan: [{ step: "   ", status: "pending" }],
         },
       },
+      {
+        toolName: "pylon_update_plan",
+        isError: false,
+        details: {
+          protocol: "pylon-plan-v1",
+          plan: [{ step: "Wait", status: "waiting" }],
+        },
+      },
+      {
+        toolName: "pylon_update_plan",
+        isError: false,
+        details: {
+          protocol: "pylon-plan-v1",
+          plan: [{ step: "Work", status: "inProgress", waitingOn: "user" }],
+        },
+      },
+      {
+        toolName: "pylon_update_plan",
+        isError: false,
+        details: {
+          protocol: "pylon-plan-v1",
+          plan: [
+            { step: "Work", status: "inProgress" },
+            { step: "Wait", status: "waiting", waitingOn: "external" },
+          ],
+        },
+      },
     ];
     for (const [index, input] of invalidResults.entries()) {
       const decoded = decodePrimeAgentDaemonEvent(

apps/server/src/provider/prime/PrimeAgentDaemonEvents.ts

@@ -101,10 +101,17 @@ export const PrimeAgentDaemonMessage = Schema.Union([
 ]);
 export type PrimeAgentDaemonMessage = typeof PrimeAgentDaemonMessage.Type;
 
-const managedPlanStepDetails = Schema.Struct({
-  step: Schema.String.check(Schema.isMaxLength(PRIME_AGENT_PLAN_MAX_STEP_CHARS)),
-  status: Schema.Literals(["pending", "inProgress", "completed"]),
-}).annotate({ parseOptions: { onExcessProperty: "error" } });
+const managedPlanStepDetails = Schema.Union([
+  Schema.Struct({
+    step: Schema.String.check(Schema.isMaxLength(PRIME_AGENT_PLAN_MAX_STEP_CHARS)),
+    status: Schema.Literals(["pending", "inProgress", "completed"]),
+  }).annotate({ parseOptions: { onExcessProperty: "error" } }),
+  Schema.Struct({
+    step: Schema.String.check(Schema.isMaxLength(PRIME_AGENT_PLAN_MAX_STEP_CHARS)),
+    status: Schema.Literal("waiting"),
+    waitingOn: Schema.Literals(["user", "delegates", "external"]),
+  }).annotate({ parseOptions: { onExcessProperty: "error" } }),
+]);
 const managedPlanDetails = Schema.Struct({
   protocol: Schema.Literal(PRIME_AGENT_PLAN_PROTOCOL),
   explanation: Schema.optional(
@@ -660,10 +667,17 @@ export interface PrimeDaemonToolCall {
 export interface PrimeDaemonPlanUpdate {
   readonly toolCallId: string;
   readonly explanation?: string | undefined;
-  readonly plan: ReadonlyArray<{
-    readonly step: string;
-    readonly status: "pending" | "inProgress" | "completed";
-  }>;
+  readonly plan: ReadonlyArray<
+    | {
+        readonly step: string;
+        readonly status: "pending" | "inProgress" | "completed";
+      }
+    | {
+        readonly step: string;
+        readonly status: "waiting";
+        readonly waitingOn: "user" | "delegates" | "external";
+      }
+  >;
 }
 
 export type PrimeDaemonMessage =
@@ -991,11 +1005,17 @@ export function projectPrimeAgentManagedPlanUpdate(input: {
   if (toolCallId.length === 0 || toolCallId.length > MAX_PREVIEW_LENGTH) return undefined;
   const decoded = decodeManagedPlanDetails(input.details);
   if (Option.isNone(decoded)) return undefined;
-  const plan = decoded.value.plan.map((item) => ({
-    step: item.step.trim(),
-    status: item.status,
-  }));
-  if (plan.some((item) => item.step.length === 0)) return undefined;
+  const plan: PrimeDaemonPlanUpdate["plan"] = decoded.value.plan.map((item) =>
+    item.status === "waiting"
+      ? { step: item.step.trim(), status: item.status, waitingOn: item.waitingOn }
+      : { step: item.step.trim(), status: item.status },
+  );
+  if (
+    plan.some((item) => item.step.length === 0) ||
+    plan.filter((item) => item.status === "inProgress" || item.status === "waiting").length > 1
+  ) {
+    return undefined;
+  }
   const explanation = decoded.value.explanation?.trim();
   return {
     toolCallId,

apps/server/src/provider/prime/PrimeAgentDaemonRuntimeEvents.test.ts

@@ -235,7 +235,7 @@ describe("mapPrimeAgentDaemonRuntimeEventDrafts", () => {
         explanation: "Prime tasks",
         plan: [
           { step: "Inspect", status: "completed" as const },
-          { step: "Implement", status: "inProgress" as const },
+          { step: "Await review", status: "waiting" as const, waitingOn: "user" as const },
         ],
       },
     };
@@ -255,7 +255,7 @@ describe("mapPrimeAgentDaemonRuntimeEventDrafts", () => {
           explanation: "Prime tasks",
           plan: [
             { step: "Inspect", status: "completed" },
-            { step: "Implement", status: "inProgress" },
+            { step: "Await review", status: "waiting", waitingOn: "user" },
           ],
         },
       },

apps/server/src/provider/prime/PrimeAgentManagedExtension.test.ts

@@ -2,6 +2,7 @@ import { describe, expect, it } from "@effect/vitest";
 
 import {
   PRIME_AGENT_PLAN_PROTOCOL,
+  PRIME_AGENT_PLAN_TOOL_DEFINITION,
   PRIME_AGENT_PLAN_TOOL_NAME,
   makePrimeAgentManagedExtensionSource,
   projectPrimeAgentManagedPermissionRequest,
@@ -14,6 +15,19 @@ const project = (input: {
   readonly message?: string;
 }) => projectPrimeAgentManagedPermissionRequest(input, token);
 
+describe("PRIME_AGENT_PLAN_TOOL_DEFINITION", () => {
+  it("advertises the same discriminated waiting contract enforced at runtime", () => {
+    const [nonWaiting, waiting] =
+      PRIME_AGENT_PLAN_TOOL_DEFINITION.parameters.properties.plan.items.oneOf;
+
+    expect(nonWaiting.properties.status.enum).toEqual(["pending", "inProgress", "completed"]);
+    expect(nonWaiting.properties).not.toHaveProperty("waitingOn");
+    expect(waiting.properties.status.const).toBe("waiting");
+    expect(waiting.properties.waitingOn.enum).toEqual(["user", "delegates", "external"]);
+    expect(waiting.required).toContain("waitingOn");
+  });
+});
+
 describe("projectPrimeAgentManagedPermissionRequest", () => {
   it("projects only the versioned, session-authorized managed confirmation format", () => {
     expect(
@@ -189,6 +203,70 @@ describe("makePrimeAgentManagedExtensionSource", () => {
     ).rejects.toThrow("root Pylon session");
   });
 
+  it("preserves honest waits and rejects ambiguous managed plans", async () => {
+    const extension = await loadExtension(extensionSource());
+    const execute = extension.tools.get(PRIME_AGENT_PLAN_TOOL_NAME)?.execute;
+    if (!execute) throw new Error("expected managed plan tool");
+
+    for (const waitingOn of ["user", "delegates", "external"] as const) {
+      await expect(
+        execute(
+          `call-wait-${waitingOn}`,
+          { plan: [{ step: "Await dependency", status: "waiting", waitingOn }] },
+          undefined,
+          undefined,
+          extensionContext(true),
+        ),
+      ).resolves.toMatchObject({
+        details: {
+          plan: [{ step: "Await dependency", status: "waiting", waitingOn }],
+        },
+      });
+    }
+
+    await expect(
+      execute(
+        "call-wait-missing-owner",
+        { plan: [{ step: "Await dependency", status: "waiting" }] },
+        undefined,
+        undefined,
+        extensionContext(true),
+      ),
+    ).rejects.toThrow("Waiting plan steps require waitingOn");
+    await expect(
+      execute(
+        "call-owner-without-wait",
+        { plan: [{ step: "Keep working", status: "inProgress", waitingOn: "user" }] },
+        undefined,
+        undefined,
+        extensionContext(true),
+      ),
+    ).rejects.toThrow("waitingOn is valid only for waiting plan steps");
+    await expect(
+      execute(
+        "call-ambiguous-active",
+        {
+          plan: [
+            { step: "Implement", status: "inProgress" },
+            { step: "Await agents", status: "waiting", waitingOn: "delegates" },
+          ],
+        },
+        undefined,
+        undefined,
+        extensionContext(true),
+      ),
+    ).rejects.toThrow("at most one in-progress or waiting step");
+    await expect(
+      execute(
+        "call-all-complete",
+        { plan: [{ step: "Done", status: "completed" }] },
+        undefined,
+        undefined,
+        extensionContext(true),
+      ),
+    ).resolves.toMatchObject({ details: { plan: [{ step: "Done", status: "completed" }] } });
+  });
+
   it("combines plan updates with the supervised gate and allows only that side-effect-free tool", async () => {
     const source = extensionSource(token);
     expect(source).toContain('pi.on("tool_call"');