Plugin: add monitor channels for cross-thread activity

Author: huntharo

index.ts

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

src/client.test.ts

@@ -218,6 +218,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,
@@ -1188,13 +1190,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") {
@@ -2500,6 +2540,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;
@@ -3558,6 +3602,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";
 
 function makeStateDir(): string {
   return fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-app-server-test-"));
@@ -102,13 +103,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 () => [
@@ -591,6 +595,138 @@ describe("Discord controller flows", () => {
     })).toBeNull();
   });
 
+  it("enables codex_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(
+      "codex_monitor",
+      buildDiscordCommandContext({
+        commandBody: "/codex_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 codex_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(
+      "codex_detach",
+      buildDiscordCommandContext({
+        commandBody: "/codex_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 codex_status when the bound conversation has an active plan run", async () => {
     const { controller } = await createControllerHarness();
     await (controller as any).store.upsertBinding({