feat(ledger): add waits history portfolio and read-only panel

apps/server/src/auth/RpcAuthorization.test.ts

@@ -69,7 +69,11 @@ describe("RPC authorization scopes", () => {
   });
 
   it("lets thread readers observe and refresh checkout branches", () => {
-    for (const method of [WS_METHODS.subscribeVcsBranch, WS_METHODS.vcsRefreshBranch]) {
+    for (const method of [
+      WS_METHODS.subscribeVcsBranch,
+      WS_METHODS.vcsRefreshBranch,
+      WS_METHODS.serverQueryWorkLedger,
+    ]) {
       expect(requiredScopeForRpcMethod(method)).toBe(AuthOrchestrationReadScope);
     }
   });

apps/server/src/auth/RpcAuthorization.ts

@@ -90,6 +90,7 @@ export const RPC_REQUIRED_SCOPES = {
   [WS_METHODS.serverRemoveKeybinding]: AuthSettingsWriteScope,
   [WS_METHODS.serverGetStorageCleanupReport]: AuthOrchestrationReadScope,
   [WS_METHODS.serverGetDatabaseStorage]: AuthOrchestrationReadScope,
+  [WS_METHODS.serverQueryWorkLedger]: AuthOrchestrationReadScope,
   [WS_METHODS.peerLinksList]: AuthAccessReadScope,
   [WS_METHODS.serverUpdateSettings]: AuthSettingsWriteScope,
   [WS_METHODS.serverSearchAcpRegistry]: AuthOrchestrationReadScope,

apps/server/src/mcp/OrchestratorMcpService.ts

@@ -2521,6 +2521,9 @@ const make = Effect.gen(function* () {
       const result = yield* threadManagement
         .dispatch({
           type: "delegated_task.request",
+          ...(input.clientRequestId === undefined
+            ? {}
+            : { clientRequestId: input.clientRequestId }),
           createdBy: "agent",
           creationSource: "mcp",
           commandId,
@@ -2899,6 +2902,9 @@ const make = Effect.gen(function* () {
                 yield* threadManagement
                   .dispatch({
                     type: "message.dispatch",
+                    ...(input.clientRequestId === undefined
+                      ? {}
+                      : { clientRequestId: input.clientRequestId }),
                     createdBy: "agent",
                     creationSource: "mcp",
                     commandId: stableCommandId({
@@ -3362,6 +3368,9 @@ const make = Effect.gen(function* () {
       }
       const result = yield* threadManagement
         .sendToThread({
+          ...(input.clientRequestId === undefined
+            ? {}
+            : { clientRequestId: input.clientRequestId }),
           projectId: target.thread.projectId,
           commandId: stableCommandId({
             scope,

apps/server/src/mcp/ThreadLaunchMcpService.ts

@@ -315,6 +315,9 @@ const launch = Effect.fn("ThreadLaunchMcpService.launch")(
           ? {}
           : {
               initialMessage: {
+                ...(input.clientRequestId === undefined
+                  ? {}
+                  : { clientRequestId: input.clientRequestId }),
                 messageId,
                 ...(caller === undefined ? {} : { senderThreadId: caller.id }),
                 ...McpInvocationContext.messageSenderFor(scope),

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

@@ -17,6 +17,18 @@ export const layer = Access.toLayer(LedgerToolkit, {
       return yield* (yield* Ledger.WorkLedger).upsert(input, updatedBy);
     }),
   ),
+  ledger_history: Access.reads((input) =>
+    Effect.gen(function* () {
+      yield* readCaller();
+      return yield* (yield* Ledger.WorkLedger).history(input);
+    }),
+  ),
+  portfolio_status: Access.reads((input) =>
+    Effect.gen(function* () {
+      yield* readCaller();
+      return yield* (yield* Ledger.WorkLedger).portfolio(input);
+    }),
+  ),
   ledger_query: Access.reads((input) =>
     Effect.gen(function* () {
       yield* readCaller();

apps/server/src/mcp/toolkits/ledger/tools.ts

@@ -1,6 +1,10 @@
 import {
   WorkLedgerUpsertInput,
-  WorkLedgerItem,
+  WorkLedgerUpsertResult,
+  WorkLedgerHistoryInput,
+  WorkLedgerHistoryResult,
+  PortfolioStatusInput,
+  PortfolioStatusResult,
   WorkLedgerQueryInput,
   WorkLedgerQueryResult,
   OrchestratorMcpFailure,
@@ -19,17 +23,35 @@ const common = {
   ],
 };
 export const LedgerToolkit = Toolkit.make(
+  Tool.make("ledger_history", {
+    ...common,
+    description:
+      "Read who changed a ledger key, when, and the before/after values. Ordered oldest first; continue with nextCursor. Default 25 rows.",
+    parameters: WorkLedgerHistoryInput,
+    success: WorkLedgerHistoryResult,
+  })
+    .annotate(Tool.Readonly, true)
+    .annotate(Tool.Destructive, false),
+  Tool.make("portfolio_status", {
+    ...common,
+    description:
+      "Open ledger rows with owner thread latest run, pending request count, result excerpt and staleWorker (no run for over 24 hours). Default 10 rows; continue with nextCursor.",
+    parameters: PortfolioStatusInput,
+    success: PortfolioStatusResult,
+  })
+    .annotate(Tool.Readonly, true)
+    .annotate(Tool.Destructive, false),
   Tool.make("ledger_upsert", {
     ...common,
     description:
       "Create or partially update work by a stable `key`. New items need title, ownerKind and ownerName. Omitted fields stay; null clears a nullable field. Requires operate access. approvalNote records context and grants no approval.",
     parameters: WorkLedgerUpsertInput,
-    success: WorkLedgerItem,
+    success: WorkLedgerUpsertResult,
   }).annotate(Tool.Destructive, true),
   Tool.make("ledger_query", {
     ...common,
     description:
-      "Read compact work ledger rows by status, ownerKind, ownerName or updatedSince (inclusive). Ordered by key; continue with nextCursor. Default 25 rows.",
+      "Read compact work ledger rows by status, ownerKind, ownerName, waitReason, due or updatedSince (inclusive). Ordered by key; continue with nextCursor. Default 25 rows.",
     parameters: WorkLedgerQueryInput,
     success: WorkLedgerQueryResult,
   })

apps/server/src/observability/RpcInstrumentation.ts

@@ -91,6 +91,7 @@ const RPC_AGGREGATES = {
   [WS_METHODS.serverListTextGeneratorModels]: "server",
   [WS_METHODS.serverPrepareTriage]: "server",
   [WS_METHODS.serverGetUsageSummary]: "server",
+  [WS_METHODS.serverQueryWorkLedger]: "server",
   [WS_METHODS.serverRefreshUsageRates]: "server",
   [WS_METHODS.serverSignalProcess]: "server",
   [WS_METHODS.serverReportClientActivity]: "server",

apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts

@@ -2155,6 +2155,8 @@ export const CLAUDE_READ_ONLY_T3_MCP_ALLOWED_TOOLS: ReadonlyArray<string> = [
   "mcp__t3-code__t3_approval_log",
   "mcp__t3-code__t3_thread_result",
   "mcp__t3-code__ledger_query",
+  "mcp__t3-code__ledger_history",
+  "mcp__t3-code__portfolio_status",
   "mcp__t3-code__t3_thread_wait",
   "mcp__t3-code__t3_worktree_status",
   "mcp__t3-code__t3_worktree_list",

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -7130,6 +7130,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         command.creationSource === "provider" &&
         !(yield* stopReachedRun(command, command.threadId, activeRun.id));
       const run: OrchestrationV2Run = {
+        ...(command.clientRequestId === undefined
+          ? {}
+          : { clientRequestId: command.clientRequestId }),
         id: runId,
         threadId: command.threadId,
         ordinal,
@@ -7533,6 +7536,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
                 mapDispatchError(command),
               );
       const run: OrchestrationV2Run = {
+        ...(command.clientRequestId === undefined
+          ? {}
+          : { clientRequestId: command.clientRequestId }),
         id: runId,
         threadId: command.threadId,
         ordinal,
@@ -8319,6 +8325,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         ),
       );
     const run: OrchestrationV2Run = {
+      ...(command.clientRequestId === undefined
+        ? {}
+        : { clientRequestId: command.clientRequestId }),
       id: runId,
       threadId: command.threadId,
       ordinal,
@@ -9030,6 +9039,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
 
       const childMessageCommand = {
         type: "message.dispatch",
+        ...(command.clientRequestId === undefined
+          ? {}
+          : { clientRequestId: command.clientRequestId }),
         createdBy: command.createdBy,
         creationSource: command.creationSource,
         commandId: command.commandId,

apps/server/src/orchestration-v2/ThreadLaunchService.ts

@@ -70,6 +70,7 @@ export type ThreadLaunchWorkspaceStrategy =
     };
 
 export interface ThreadLaunchInitialMessage {
+  readonly clientRequestId?: string;
   readonly messageId?: MessageId;
   readonly scheduledTaskId?: ScheduledTaskId;
   readonly senderThreadId?: ThreadId;
@@ -1052,6 +1053,9 @@ const make = Effect.gen(function* () {
           const dispatched = yield* threads
             .dispatch({
               type: "message.dispatch",
+              ...(input.initialMessage.clientRequestId === undefined
+                ? {}
+                : { clientRequestId: input.initialMessage.clientRequestId }),
               commandId: messageCommandId,
               threadId,
               messageId,

apps/server/src/orchestration-v2/ThreadManagementService.ts

@@ -140,6 +140,7 @@ export type ThreadManagementTerminalRunStatus = Extract<
 >;
 
 export interface ThreadManagementSendInput {
+  readonly clientRequestId?: string;
   readonly projectId: ProjectId;
   readonly commandId: CommandId;
   readonly threadId: ThreadId;
@@ -868,6 +869,7 @@ const make = Effect.gen(function* () {
 
     const dispatched = yield* dispatch({
       type: "message.dispatch",
+      ...(input.clientRequestId === undefined ? {} : { clientRequestId: input.clientRequestId }),
       commandId: input.commandId,
       threadId: input.threadId,
       messageId: input.messageId,

apps/server/src/orchestration-v2/runtimeLayer.test.ts

@@ -1082,13 +1082,17 @@ it.layer(layerTest)("OrchestrationV2LayerLive", (it) => {
       expect(projection.messages).toHaveLength(1);
       expect(projection.messages[0]).toMatchObject({
         text: command.text,
+        postOnly: true,
+        posted: true,
         runId: null,
         createdBy: "agent",
         creationSource: "mcp",
         senderThreadId: "sender",
         senderClient: command.senderClient,
       });
-      expect(projection.turnItems.filter((item) => item.type === "user_message")).toHaveLength(1);
+      expect(projection.turnItems.filter((item) => item.type === "user_message")).toMatchObject([
+        { messageId: command.messageId, postOnly: true, inputIntent: "post" },
+      ]);
     }),
   );
 

apps/server/src/persistence/ForkMigrations.ts

@@ -11,6 +11,7 @@ import ForkMigration0006 from "./ForkMigrations/006_McpSubscriptionRenewal.ts";
 import ForkMigration0007 from "./ForkMigrations/007_McpRequestLog.ts";
 import ForkMigration0008 from "./ForkMigrations/008_McpApprovalAnswers.ts";
 import ForkMigration0009 from "./ForkMigrations/009_McpInboxRelay.ts";
+import ForkMigration0010 from "./ForkMigrations/010_WorkLedgerPortfolio.ts";
 // Fork-only migrations, recorded in their own table so upstream keeps every id in
 // effect_sql_migrations. They run after all upstream migrations. Add new ones here, never in
 // Migrations.ts, and never rename one: the repair below finds old rows by name. See FORK.md.
@@ -24,6 +25,7 @@ const forkMigrationEntries = [
   [7, "McpRequestLog", ForkMigration0007],
   [8, "McpApprovalAnswers", ForkMigration0008],
   [9, "McpInboxRelay", ForkMigration0009],
+  [10, "WorkLedgerPortfolio", ForkMigration0010],
 ] as const;
 
 export const FORK_MIGRATIONS_TABLE = "t3_fork_migrations";

apps/server/src/persistence/ForkMigrations/010_WorkLedgerPortfolio.ts

@@ -0,0 +1,58 @@
+import { formatThreadLink, parseThreadLinkHref } from "@t3tools/shared/threadLinks";
+import * as Schema from "effect/Schema";
+import * as Effect from "effect/Effect";
+import * as SqlClient from "effect/sql/SqlClient";
+
+const linksCodec = Schema.fromJsonString(Schema.Array(Schema.String));
+const decodeLinks = Schema.decodeUnknownEffect(linksCodec);
+const encodeLinks = Schema.encodeEffect(linksCodec);
+
+export default Effect.gen(function* () {
+  const sql = yield* SqlClient.SqlClient;
+  yield* sql`ALTER TABLE work_ledger ADD COLUMN wait_reason TEXT`;
+  yield* sql`ALTER TABLE work_ledger ADD COLUMN next_check_at TEXT`;
+  yield* sql`ALTER TABLE work_ledger ADD COLUMN last_run_id TEXT`;
+  yield* sql`ALTER TABLE work_ledger ADD COLUMN last_run_status TEXT`;
+  yield* sql`ALTER TABLE work_ledger ADD COLUMN last_run_at TEXT`;
+  yield* sql`UPDATE work_ledger SET wait_reason = CASE
+    WHEN blocker GLOB 'WAIT_EXTERNAL*' THEN 'external'
+    WHEN blocker GLOB 'WAIT_NATURAL*' THEN 'natural'
+    WHEN blocker GLOB 'NEEDS_DECISION*' THEN 'decision' ELSE NULL END`;
+  const owners = yield* sql<{ work_key: string; owner_thread_id: string; links_json: string }>`
+    SELECT work_key, owner_thread_id, links_json FROM work_ledger WHERE owner_thread_id IS NOT NULL
+  `;
+  for (const owner of owners) {
+    const ref = parseThreadLinkHref(owner.owner_thread_id);
+    if (ref === null) continue;
+    if (ref.environmentId === null) {
+      yield* sql`UPDATE work_ledger SET owner_thread_id = ${ref.threadId} WHERE work_key = ${owner.work_key}`;
+    } else {
+      const links = yield* decodeLinks(owner.links_json);
+      const href = formatThreadLink(ref).slice(1, -1);
+      const encodedLinks = yield* encodeLinks([...new Set([...links, href])]);
+      yield* sql`UPDATE work_ledger SET owner_thread_id = NULL,
+        links_json = ${encodedLinks} WHERE work_key = ${owner.work_key}`;
+    }
+  }
+  yield* sql`CREATE INDEX work_ledger_due_idx ON work_ledger(next_check_at) WHERE next_check_at IS NOT NULL`;
+  yield* sql`CREATE TABLE work_ledger_history (
+    id INTEGER PRIMARY KEY AUTOINCREMENT, work_key TEXT NOT NULL,
+    changed_at TEXT NOT NULL, changed_by TEXT NOT NULL, changes_json TEXT NOT NULL
+  )`;
+  yield* sql`CREATE INDEX work_ledger_history_key_idx ON work_ledger_history(work_key, id)`;
+  // Keep terminal-run metadata in the projection transaction, including recovery and replay.
+  for (const operation of ["INSERT", "UPDATE"] as const) {
+    yield* sql.unsafe(`CREATE TRIGGER work_ledger_run_${operation.toLowerCase()}
+      AFTER ${operation} ON orchestration_v2_projection_runs
+      WHEN NEW.status IN ('completed', 'failed', 'cancelled', 'interrupted', 'rolled_back')
+        AND CASE WHEN json_valid(NEW.payload_json)
+          THEN json_extract(NEW.payload_json, '$.clientRequestId') IS NOT NULL ELSE 0 END
+      BEGIN
+        UPDATE work_ledger SET last_run_id = NEW.run_id, last_run_status = NEW.status,
+          last_run_at = COALESCE(NEW.completed_at, NEW.requested_at),
+          updated_at = MAX(updated_at, COALESCE(NEW.completed_at, NEW.requested_at))
+        WHERE work_key = json_extract(NEW.payload_json, '$.clientRequestId')
+          AND (last_run_at IS NULL OR last_run_at <= COALESCE(NEW.completed_at, NEW.requested_at));
+      END`);
+  }
+});