Plugin: add monitor channels for cross-thread activity

Author: huntharo

README.md

@@ -40,7 +40,7 @@ Pre-release packages are published on matching npm dist-tags instead of `latest`
 
 - Uses your existing local Codex CLI setup instead of a separate hosted bridge.
 - Feels natural in chat: bind once with `/cas_resume`, then just talk.
-- Keeps useful controls close at hand with `/cas_status`, `/cas_plan`, `/cas_review`, `/cas_model`, and more.
+- Keeps useful controls close at hand with `/cas_status`, `/cas_monitor`, `/cas_plan`, `/cas_review`, `/cas_model`, and more.
 - Works well for Telegram and Discord conversations that you want tied to a real Codex thread.
 
 ## Typical Workflow
@@ -62,6 +62,9 @@ Pre-release packages are published on matching npm dist-tags instead of `latest`
 | `/cas_resume --sync` | Resume and try to sync the chat/topic name to the Codex thread. | You can combine this with other flags. |
 | `/cas_resume release-fix` | Resume a matching thread by title or id. | If more than one thread matches, you get buttons to choose. |
 | `/cas_status` | Show the current binding and thread state. | Includes thread id, model, workspace, sandbox, and permissions when available. |
+| `/cas_monitor` | Bind this conversation as a cross-thread monitor surface. | Shows pending approvals, pending questions, and unread thread activity across sessions. |
+| `/cas_monitor --cwd ~/github/openclaw` | Limit monitor mode to one workspace. | Uses the same `--cwd` path handling as `/cas_resume`. |
+| `/cas_monitor status|off` | Inspect or disable monitor mode. | `off` is also compatible with `/cas_detach`. |
 | `/cas_detach` | Unbind this conversation from Codex. | Stops routing plain text from this conversation into the bound thread. |
 | `/cas_stop` | Interrupt the active Codex run. | Only applies when a turn is currently in progress. |
 | `/cas_steer <message>` | Send follow-up steer text to an active run. | Example: `/cas_steer focus on the failing tests first` |

index.test.ts

@@ -38,6 +38,7 @@ describe("plugin registration", () => {
     expect(api.registerCommand).toHaveBeenCalled();
     expect(api.registerCommand.mock.calls.map(([params]) => params.name)).toEqual([
       "cas_resume",
+      "cas_monitor",
       "cas_detach",
       "cas_status",
       "cas_stop",

index.ts

@@ -4,6 +4,7 @@ import { INTERACTIVE_NAMESPACE } from "./src/types.js";
 
 const COMMANDS = [
   ["cas_resume", "Resume or bind an existing Codex thread."],
+  ["cas_monitor", "Monitor cross-thread approvals, prompts, and unread activity."],
   ["cas_detach", "Detach this conversation from the current Codex thread."],
   ["cas_status", "Show the current Codex binding and thread state."],
   ["cas_stop", "Stop the active Codex turn."],

src/client.test.ts

@@ -261,6 +261,39 @@ describe("extractThreadTokenUsageSnapshot", () => {
       remainingPercent: 80,
     });
   });
+
+  it("extracts thread status and active flags from thread list payloads", () => {
+    expect(
+      __testing.extractThreadsFromValue({
+        threads: [
+          {
+            id: "thread-1",
+            name: "Needs approval",
+            cwd: "/repo/openclaw",
+            updatedAt: 1_710_000_000,
+            status: {
+              type: "active",
+              activeFlags: ["waitingOnApproval", "waitingOnUserInput"],
+            },
+          },
+        ],
+      }),
+    ).toEqual([
+      {
+        threadId: "thread-1",
+        title: "Needs approval",
+        summary: "",
+        projectKey: "/repo/openclaw",
+        createdAt: undefined,
+        updatedAt: 1_710_000_000_000,
+        gitBranch: undefined,
+        status: {
+          type: "active",
+          activeFlags: ["waitingOnApproval", "waitingOnUserInput"],
+        },
+      },
+    ]);
+  });
 });
 
 describe("extractFileChangePathsFromReadResult", () => {

src/client.ts

@@ -21,8 +21,10 @@ import type {
   ReviewResult,
   ReviewTarget,
   SkillSummary,
+  ThreadActiveFlag,
   ThreadReplay,
   ThreadState,
+  ThreadStatusSummary,
   ThreadSummary,
   TurnTerminalError,
   TurnResult,
@@ -1217,13 +1219,51 @@ function extractThreadsFromValue(value: unknown): ThreadSummary[] {
         pickString(asRecord(record.git_info) ?? {}, ["branch"]) ??
         pickString(asRecord(sessionRecord?.gitInfo) ?? {}, ["branch"]) ??
         pickString(asRecord(sessionRecord?.git_info) ?? {}, ["branch"]),
+      status: extractThreadStatus(record) ?? extractThreadStatus(sessionRecord),
     });
   }
   return [...summaries.values()].sort(
     (left, right) => (right.updatedAt ?? 0) - (left.updatedAt ?? 0),
   );
 }
 
+function normalizeThreadActiveFlag(value: unknown): ThreadActiveFlag | undefined {
+  if (typeof value !== "string") {
+    return undefined;
+  }
+  const normalized = value.trim();
+  return normalized === "waitingOnApproval" || normalized === "waitingOnUserInput"
+    ? normalized
+    : undefined;
+}
+
+function extractThreadStatus(value: unknown): ThreadStatusSummary | undefined {
+  const record = asRecord(asRecord(value)?.status ?? value);
+  if (!record) {
+    const statusType = typeof value === "string" ? value.trim() : "";
+    if (
+      statusType === "notLoaded" ||
+      statusType === "idle" ||
+      statusType === "systemError"
+    ) {
+      return { type: statusType };
+    }
+    return undefined;
+  }
+  const type = pickString(record, ["type", "status", "kind"]);
+  if (type === "notLoaded" || type === "idle" || type === "systemError") {
+    return { type };
+  }
+  if (type === "active") {
+    const rawFlags = findFirstArrayByKeys(record, ["activeFlags", "active_flags"]) ?? [];
+    const activeFlags = rawFlags
+      .map((entry) => normalizeThreadActiveFlag(entry))
+      .filter((entry): entry is ThreadActiveFlag => Boolean(entry));
+    return { type, activeFlags };
+  }
+  return undefined;
+}
+
 function normalizeConversationRole(value: string | undefined): "user" | "assistant" | undefined {
   const normalized = value?.trim().toLowerCase();
   if (normalized === "user" || normalized === "usermessage") {
@@ -2529,6 +2569,10 @@ export class CodexAppServerClient {
     await connection?.client.close().catch(() => undefined);
   }
 
+  onNotification(listener: (method: string, params: unknown) => Promise<void> | void): () => void {
+    return this.addNotificationListener(listener);
+  }
+
   async listThreads(params: {
     sessionKey?: string;
     workspaceDir?: string;
@@ -3618,6 +3662,7 @@ export const __testing = {
   extractStartupProbeInfo,
   formatFileEditNotice,
   extractThreadTokenUsageSnapshot,
+  extractThreadsFromValue,
   extractRateLimitSummaries,
   formatStdioProcessLog,
   resolveTurnStoppedReason,

src/controller.test.ts

@@ -5,6 +5,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import type { OpenClawPluginApi, PluginCommandContext } from "openclaw/plugin-sdk";
 import { CodexAppServerClient } from "./client.js";
 import { CodexPluginController } from "./controller.js";
+import type { ThreadSummary } from "./types.js";
 
 const discordSdkState = vi.hoisted(() => ({
   buildDiscordComponentMessage: vi.fn((params: { spec: { text?: string; blocks?: unknown[] } }) => ({
@@ -123,13 +124,16 @@ async function createControllerHarness() {
   const controller = new CodexPluginController(api);
   await controller.start();
   const clientMock = {
-    listThreads: vi.fn(async () => [
+    listThreads: vi.fn(async (): Promise<ThreadSummary[]> => [
       {
         threadId: "thread-1",
         title: "Discord Thread",
         projectKey: "/repo/openclaw",
         createdAt: Date.now() - 60_000,
         updatedAt: Date.now() - 30_000,
+        status: {
+          type: "idle",
+        },
       },
     ]),
     listModels: vi.fn(async () => [
@@ -899,6 +903,138 @@ describe("Discord controller flows", () => {
     })).toBeNull();
   });
 
+  it("enables cas_monitor and reports cross-thread activity", async () => {
+    const { controller, clientMock } = await createControllerHarness();
+    clientMock.listThreads.mockResolvedValue([
+      {
+        threadId: "thread-approval",
+        title: "Approve deploy",
+        projectKey: "/repo/openclaw",
+        updatedAt: Date.now() - 60_000,
+        status: {
+          type: "active",
+          activeFlags: ["waitingOnApproval"],
+        },
+      },
+      {
+        threadId: "thread-unread",
+        title: "Fresh output",
+        projectKey: "/repo/openclaw",
+        updatedAt: Date.now() - 30_000,
+        status: {
+          type: "idle",
+        },
+      },
+    ]);
+
+    const reply = await controller.handleCommand(
+      "cas_monitor",
+      buildDiscordCommandContext({
+        commandBody: "/cas_monitor",
+      }),
+    );
+
+    expect(reply.text).toContain("Monitor: active");
+    expect(reply.text).toContain("Pending approvals:");
+    expect(reply.text).toContain("Unread activity:");
+    expect((controller as any).store.getMonitorBinding({
+      channel: "discord",
+      accountId: "default",
+      conversationId: "channel:chan-1",
+    })).toEqual(
+      expect.objectContaining({
+        workspaceDir: undefined,
+      }),
+    );
+  });
+
+  it("detaches monitor bindings with cas_detach", async () => {
+    const { controller } = await createControllerHarness();
+    await (controller as any).store.upsertMonitorBinding({
+      conversation: {
+        channel: "discord",
+        accountId: "default",
+        conversationId: "channel:chan-1",
+      },
+      updatedAt: Date.now(),
+    });
+
+    const reply = await controller.handleCommand(
+      "cas_detach",
+      buildDiscordCommandContext({
+        commandBody: "/cas_detach",
+      }),
+    );
+
+    expect(reply).toEqual({
+      text: "Detached this conversation from Codex.",
+    });
+    expect((controller as any).store.getMonitorBinding({
+      channel: "discord",
+      accountId: "default",
+      conversationId: "channel:chan-1",
+    })).toBeNull();
+  });
+
+  it("does not claim inbound messages for monitor-only conversations", async () => {
+    const { controller } = await createControllerHarness();
+    await (controller as any).store.upsertMonitorBinding({
+      conversation: {
+        channel: "discord",
+        accountId: "default",
+        conversationId: "channel:chan-1",
+      },
+      updatedAt: Date.now(),
+    });
+
+    const result = await controller.handleInboundClaim({
+      content: "hello",
+      channel: "discord",
+      accountId: "default",
+      conversationId: "discord:channel:chan-1",
+      metadata: { guildId: "guild-1" },
+    });
+
+    expect(result).toEqual({ handled: false });
+  });
+
+  it("dedupes unchanged monitor refresh summaries", async () => {
+    const { controller, clientMock, sendMessageDiscord } = await createControllerHarness();
+    await (controller as any).store.upsertMonitorBinding({
+      conversation: {
+        channel: "discord",
+        accountId: "default",
+        conversationId: "channel:chan-1",
+      },
+      updatedAt: Date.now(),
+    });
+    clientMock.listThreads.mockResolvedValue([
+      {
+        threadId: "thread-unread",
+        title: "Fresh output",
+        projectKey: "/repo/openclaw",
+        updatedAt: Date.now() - 30_000,
+        status: {
+          type: "idle",
+        },
+      },
+    ]);
+
+    await (controller as any).refreshMonitorBindings({ force: true });
+    await (controller as any).refreshMonitorBindings();
+
+    expect(sendMessageDiscord).toHaveBeenCalledTimes(1);
+    expect((controller as any).store.getMonitorBinding({
+      channel: "discord",
+      accountId: "default",
+      conversationId: "channel:chan-1",
+    })).toEqual(
+      expect.objectContaining({
+        lastSummarySignature: expect.stringContaining("Unread activity:"),
+      }),
+    );
+  });
+
   it("shows plan mode on in cas_status when the bound conversation has an active plan run", async () => {
     const { controller } = await createControllerHarness();
     await (controller as any).store.upsertBinding({