orchestrate: eight open fixes in one branch (kill-tree, pagination, handoff parsing, measurements, redaction, comment-queue lock, watchdog stream)

orchestrate/skills/orchestrate/scripts/__tests__/andon-root-cache.test.ts

@@ -124,6 +124,104 @@ describe("SlackReactionAndonSource", () => {
     expect(calls).toEqual([{ limit: 20, latest: expect.any(String) }]);
   });
 
+  test("Finds the newest reason beyond the first reply page", async () => {
+    const cursors: (string | undefined)[] = [];
+    const latestBounds: (string | undefined)[] = [];
+    const source = new SlackReactionAndonSource(
+      {
+        ...slackWithReactions([{ name: "rotating_light", users: ["U123"] }]),
+        async getThreadReplies(args) {
+          cursors.push(args.cursor);
+          latestBounds.push(args.latest);
+          return args.cursor === undefined
+            ? {
+                messages: [
+                  {
+                    ts: "111.222",
+                    text: "🚨 ANDON RAISED by parent: ignore parent",
+                  },
+                  {
+                    ts: "111.300",
+                    text: "🚨 ANDON RAISED by operator: old reason",
+                  },
+                ],
+                nextCursor: "page-two",
+              }
+            : {
+                messages: [
+                  {
+                    ts: "111.400",
+                    text: ":rotating_light: ANDON RAISED by operator: newest reason",
+                  },
+                  { ts: "111.500", text: "unrelated newest reply" },
+                ],
+              };
+        },
+      },
+      { channel: "C123", ts: "111.222" }
+    );
+    await expect(source.snapshot()).resolves.toMatchObject({
+      active: true,
+      reason: "newest reason",
+    });
+    expect(cursors).toEqual([undefined, "page-two"]);
+    expect(latestBounds[0]).toBe(latestBounds[1]);
+  });
+
+  test("Applies the recent reply window across pages", async () => {
+    const source = new SlackReactionAndonSource(
+      {
+        ...slackWithReactions([{ name: "rotating_light", users: ["U123"] }]),
+        async getThreadReplies(args) {
+          return args.cursor === undefined
+            ? {
+                messages: [
+                  {
+                    ts: "111.300",
+                    text: "🚨 ANDON RAISED by operator: outside recent window",
+                  },
+                ],
+                nextCursor: "page-two",
+              }
+            : {
+                messages: Array.from({ length: 20 }, (_, i) => ({
+                  ts: `112.${100 + i}`,
+                  text: "unrelated",
+                })),
+              };
+        },
+      },
+      { channel: "C123", ts: "111.222" }
+    );
+    const state = await source.snapshot();
+    expect(state.active).toBe(true);
+    expect(state).not.toHaveProperty("reason", "outside recent window");
+  });
+
+  test("Keeps Andon active when a later reply page fails", async () => {
+    const source = new SlackReactionAndonSource(
+      {
+        ...slackWithReactions([{ name: "rotating_light", users: ["U123"] }]),
+        async getThreadReplies(args) {
+          if (args.cursor) throw new Error("slack_replies_unavailable");
+          return {
+            messages: [
+              {
+                ts: "111.300",
+                text: "🚨 ANDON RAISED by operator: incomplete history",
+              },
+            ],
+            nextCursor: "page-two",
+          };
+        },
+      },
+      { channel: "C123", ts: "111.222" }
+    );
+    const state = await source.snapshot();
+    expect(state.active).toBe(true);
+    expect(state).not.toHaveProperty("reason", "incomplete history");
+  });
+
   test("Keeps Andon active when reason reply fetch fails", async () => {
     const source = new SlackReactionAndonSource(
       {

orchestrate/skills/orchestrate/scripts/__tests__/comment-retry-queue-lock.test.ts

@@ -0,0 +1,78 @@
+import { describe, expect, test } from "bun:test";
+import { mkdtempSync, readFileSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+import type { SlackAdapter, SlackMessageRef } from "../adapters/types.ts";
+import {
+  commentRetryQueuePath,
+  postOrQueueComment,
+} from "../core/comment-retry-queue.ts";
+
+const TEST_SLACK_CHANNEL = "C123TEST";
+
+// Slack is down for everyone. The first post hangs for a moment so that a
+// second caller reads and writes the queue file while the first is waiting.
+function slowThenFailingSlack(): SlackAdapter {
+  let calls = 0;
+  const notUsed = async (): Promise<never> => {
+    throw new Error("not used");
+  };
+  return {
+    postRunKickoff: notUsed,
+    lookupFirstNameByEmail: async () => undefined,
+    postInThread: notUsed,
+    editThreadMessage: notUsed,
+    uploadFileToThread: notUsed,
+    getReactions: notUsed,
+    getThreadReplies: notUsed,
+    async postCommentInThread(): Promise<SlackMessageRef> {
+      calls++;
+      if (calls === 1) await new Promise(resolve => setTimeout(resolve, 60));
+      throw new Error("slack_unavailable");
+    },
+    addReaction: notUsed,
+    removeReaction: notUsed,
+  };
+}
+
+describe("required comment retry queue under concurrent callers", () => {
+  // Two required comments posted at the same time, as two local workers (or the
+  // orchestrator and a worker) do at a handoff.
+  test("Neither entry is lost and neither caller reports a comment it never delivered", async () => {
+    const workspace = mkdtempSync(join(tmpdir(), "orch-comment-queue-lock-"));
+    try {
+      const slack = slowThenFailingSlack();
+      const destination = `slack:${TEST_SLACK_CHANNEL}:111.222`;
+      const first = postOrQueueComment({
+        destinations: { slack },
+        workspace,
+        destination,
+        body: "first worker is done",
+        sender: "worker-1",
+        criticality: "required",
+      });
+      await new Promise(resolve => setTimeout(resolve, 10));
+      const second = postOrQueueComment({
+        destinations: { slack },
+        workspace,
+        destination,
+        body: "second worker is done",
+        sender: "worker-2",
+        criticality: "required",
+      });
+      const results = await Promise.all([first, second]);
+
+      const queue = JSON.parse(
+        readFileSync(commentRetryQueuePath(workspace), "utf8")
+      ) as { body: string }[];
+      expect(queue.map(entry => entry.body).sort()).toEqual([
+        "first worker is done",
+        "second worker is done",
+      ]);
+      expect(results).toEqual(["queued", "queued"]);
+    } finally {
+      rmSync(workspace, { recursive: true, force: true });
+    }
+  });
+});

orchestrate/skills/orchestrate/scripts/__tests__/comment-retry-queue.test.ts

@@ -1,5 +1,11 @@
 import { describe, expect, test } from "bun:test";
-import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
+import {
+  mkdirSync,
+  mkdtempSync,
+  readFileSync,
+  rmSync,
+  writeFileSync,
+} from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 
@@ -392,3 +398,97 @@ describe("required comment retry queue", () => {
     }
   });
 });
+
+describe("invalid persisted required comment queues", () => {
+  const valid = {
+    id: "persisted-required-comment",
+    destination: `slack:${TEST_SLACK_CHANNEL}:111.222`,
+    body: "keep this required comment",
+    sender: "agent",
+    attempts: 0,
+    createdAt: "1970-01-01T00:00:00.000Z",
+    nextAttemptAt: "1970-01-01T00:00:00.000Z",
+  };
+  for (const [name, content] of [
+    ["malformed JSON", '[{"body":"private fixture content"}'],
+    ["non-array JSON", JSON.stringify({ pending: [valid] })],
+    [
+      "invalid entry beside a valid comment",
+      JSON.stringify([valid, { id: "broken" }]),
+    ],
+  ]) {
+    for (const operation of ["enqueue", "drain"] as const) {
+      test(`${operation} preserves ${name} without posting`, async () => {
+        const workspace = mkdtempSync(
+          join(tmpdir(), "orch-comment-corruption-")
+        );
+        const posts: SlackPost[] = [];
+        const queuePath = commentRetryQueuePath(workspace);
+        try {
+          writeFileSync(queuePath, content);
+          const destinations = { slack: slackAdapter({ posts }) };
+          const run =
+            operation === "drain"
+              ? drainCommentRetryQueue({ workspace, destinations })
+              : postOrQueueComment({
+                  workspace,
+                  destinations,
+                  destination: valid.destination,
+                  body: "new required comment",
+                  sender: "agent",
+                  criticality: "required",
+                });
+          await expect(run).rejects.toThrow(/comment queue/);
+          expect(readFileSync(queuePath, "utf8")).toBe(content);
+          expect(posts).toEqual([]);
+        } finally {
+          rmSync(workspace, { recursive: true, force: true });
+        }
+      });
+    }
+  }
+
+  test("read failures preserve the queue path without posting", async () => {
+    const workspace = mkdtempSync(join(tmpdir(), "orch-comment-read-error-"));
+    const posts: SlackPost[] = [];
+    try {
+      mkdirSync(commentRetryQueuePath(workspace));
+      await expect(
+        postOrQueueComment({
+          workspace,
+          destinations: { slack: slackAdapter({ posts }) },
+          destination: valid.destination,
+          body: "new comment",
+          sender: "agent",
+          criticality: "required",
+        })
+      ).rejects.toThrow(/comment queue/);
+      expect(posts).toEqual([]);
+    } finally {
+      rmSync(workspace, { recursive: true, force: true });
+    }
+  });
+
+  test("a valid empty queue can enqueue and deliver normally", async () => {
+    const workspace = mkdtempSync(join(tmpdir(), "orch-comment-empty-"));
+    const posts: SlackPost[] = [];
+    try {
+      writeFileSync(commentRetryQueuePath(workspace), "[]");
+      const result = await postOrQueueComment({
+        workspace,
+        destinations: { slack: slackAdapter({ posts }) },
+        destination: valid.destination,
+        body: "new comment",
+        sender: "agent",
+        criticality: "required",
+      });
+      expect(result).toBe("posted");
+      expect(posts.map(post => post.text)).toEqual(["new comment"]);
+      expect(
+        JSON.parse(readFileSync(commentRetryQueuePath(workspace), "utf8"))
+      ).toEqual([]);
+    } finally {
+      rmSync(workspace, { recursive: true, force: true });
+    }
+  });
+});

orchestrate/skills/orchestrate/scripts/__tests__/handoff-branch-parser.test.ts

@@ -3,6 +3,13 @@ import { describe, expect, test } from "bun:test";
 import { parseHandoffBranch, resolveRunBranch } from "../core/handoff.ts";
 
 describe("parseHandoffBranch / resolveRunBranch", () => {
+  test("A branch written as a one-item list is read", () => {
+    const body = ["## Branch", "- `agent/listed-abc1`", "## What I did"].join(
+      "\n"
+    );
+    expect(parseHandoffBranch(body)).toBe("agent/listed-abc1");
+  });
+
   test("Worker template wraps branch value in single backticks", () => {
     const body = [
       "## Status",

orchestrate/skills/orchestrate/scripts/__tests__/handoff-empty-canonical.test.ts

@@ -0,0 +1,27 @@
+import { describe, expect, test } from "bun:test";
+import { parseHandoffVerification } from "../core/handoff.ts";
+
+describe("canonical verification remains authoritative when empty", () => {
+  test("empty canonical verdict cannot inherit legacy pass", () => {
+    expect(
+      parseHandoffVerification("## Verification\n\n## Verdict\npass\n")
+    ).toBeNull();
+  });
+  test("empty terminal canonical section cannot inherit earlier legacy pass", () => {
+    expect(
+      parseHandoffVerification("## Verdict\npass\n\n## Verification\n")
+    ).toBeNull();
+  });
+  test("legacy-only verdict remains supported", () => {
+    expect(parseHandoffVerification("## Verdict\npass\n")).toBe(
+      "type-check-only"
+    );
+  });
+  test("recognized canonical verdict retains precedence", () => {
+    expect(
+      parseHandoffVerification(
+        "## Verification\nverifier-blocked\n\n## Verdict\npass\n"
+      )
+    ).toBe("verifier-blocked");
+  });
+});