From 5d62cdf574b95b0f00731c9c8c7de1b3462953d2 Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Wed, 11 Nov 2026 23:22:49 -0700 Subject: [PATCH 01/10] feat(ledger): track waits history due checks and portfolio --- apps/server/src/mcp/OrchestratorMcpService.ts | 9 + apps/server/src/mcp/ThreadLaunchMcpService.ts | 3 + .../src/mcp/toolkits/ledger/handlers.ts | 12 + apps/server/src/mcp/toolkits/ledger/tools.ts | 28 ++- .../src/orchestration-v2/Orchestrator.ts | 12 + .../orchestration-v2/ThreadLaunchService.ts | 4 + .../ThreadManagementService.ts | 2 + .../src/persistence/ForkMigrations.test.ts | 1 + apps/server/src/persistence/ForkMigrations.ts | 2 + .../ForkMigrations/007_WorkLedgerPortfolio.ts | 34 +++ .../server/src/persistence/WorkLedger.test.ts | 157 ++++++++++++- apps/server/src/persistence/WorkLedger.ts | 207 +++++++++++++++--- packages/contracts/src/orchestrationV2.ts | 3 + packages/contracts/src/workLedger.ts | 68 +++++- 14 files changed, 509 insertions(+), 33 deletions(-) create mode 100644 apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index b6d37d4b666f6..838b84d8145fd 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -2505,6 +2505,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, @@ -2883,6 +2886,9 @@ const make = Effect.gen(function* () { yield* threadManagement .dispatch({ type: "message.dispatch", + ...(input.clientRequestId === undefined + ? {} + : { clientRequestId: input.clientRequestId }), createdBy: "agent", creationSource: "mcp", commandId: stableCommandId({ @@ -3245,6 +3251,9 @@ const make = Effect.gen(function* () { ); const result = yield* threadManagement .sendToThread({ + ...(input.clientRequestId === undefined + ? {} + : { clientRequestId: input.clientRequestId }), projectId: target.thread.projectId, commandId: stableCommandId({ scope, diff --git a/apps/server/src/mcp/ThreadLaunchMcpService.ts b/apps/server/src/mcp/ThreadLaunchMcpService.ts index 0aadb506dc0ce..0f8f9de45031d 100644 --- a/apps/server/src/mcp/ThreadLaunchMcpService.ts +++ b/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 }), text: input.message ?? "", diff --git a/apps/server/src/mcp/toolkits/ledger/handlers.ts b/apps/server/src/mcp/toolkits/ledger/handlers.ts index 02c0d396b4f2f..effd615be4256 100644 --- a/apps/server/src/mcp/toolkits/ledger/handlers.ts +++ b/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(); diff --git a/apps/server/src/mcp/toolkits/ledger/tools.ts b/apps/server/src/mcp/toolkits/ledger/tools.ts index 6abfc9e9b1d03..924afd4f9477d 100644 --- a/apps/server/src/mcp/toolkits/ledger/tools.ts +++ b/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, }) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index d59cdab879c43..556419f2c5fbc 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -7022,6 +7022,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, @@ -7424,6 +7427,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, @@ -8208,6 +8214,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), ); const run: OrchestrationV2Run = { + ...(command.clientRequestId === undefined + ? {} + : { clientRequestId: command.clientRequestId }), id: runId, threadId: command.threadId, ordinal, @@ -8917,6 +8926,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, diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index 65d2c3b8d8e9c..00b6d45dda5cb 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -69,6 +69,7 @@ export type ThreadLaunchWorkspaceStrategy = }; export interface ThreadLaunchInitialMessage { + readonly clientRequestId?: string; readonly messageId?: MessageId; readonly scheduledTaskId?: ScheduledTaskId; readonly senderThreadId?: ThreadId; @@ -1050,6 +1051,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, diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 57e2ac2b771d9..123244f3c8e9c 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -137,6 +137,7 @@ export type ThreadManagementTerminalRunStatus = Extract< >; export interface ThreadManagementSendInput { + readonly clientRequestId?: string; readonly projectId: ProjectId; readonly commandId: CommandId; readonly threadId: ThreadId; @@ -864,6 +865,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, diff --git a/apps/server/src/persistence/ForkMigrations.test.ts b/apps/server/src/persistence/ForkMigrations.test.ts index 11acf92adc0f5..e55551914fcfc 100644 --- a/apps/server/src/persistence/ForkMigrations.test.ts +++ b/apps/server/src/persistence/ForkMigrations.test.ts @@ -113,6 +113,7 @@ describe("fork migrations", () => { [3, "McpClientThreads"], [4, "McpBridgeStatus"], [5, "McpOAuthRenewal"], + [7, "WorkLedgerPortfolio"], ], ); assert.equal((yield* forkHistory)[0]?.[2], "2026-11-12 04:10:00"); diff --git a/apps/server/src/persistence/ForkMigrations.ts b/apps/server/src/persistence/ForkMigrations.ts index dc07fd32149a7..f1c89d5f99f43 100644 --- a/apps/server/src/persistence/ForkMigrations.ts +++ b/apps/server/src/persistence/ForkMigrations.ts @@ -5,6 +5,7 @@ import * as SqlClient from "effect/sql/SqlClient"; import ForkMigration0001 from "./ForkMigrations/001_McpEvents.ts"; import ForkMigration0002 from "./ForkMigrations/002_WorkLedger.ts"; import ForkMigration0003 from "./ForkMigrations/003_McpClientThreads.ts"; +import ForkMigration0007 from "./ForkMigrations/007_WorkLedgerPortfolio.ts"; import ForkMigration0004 from "./ForkMigrations/004_McpBridgeStatus.ts"; import ForkMigration0005 from "./ForkMigrations/005_McpOAuthRenewal.ts"; @@ -17,6 +18,7 @@ const forkMigrationEntries = [ [3, "McpClientThreads", ForkMigration0003], [4, "McpBridgeStatus", ForkMigration0004], [5, "McpOAuthRenewal", ForkMigration0005], + [7, "WorkLedgerPortfolio", ForkMigration0007], ] as const; export const FORK_MIGRATIONS_TABLE = "t3_fork_migrations"; diff --git a/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts b/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts new file mode 100644 index 0000000000000..de7d3875646fa --- /dev/null +++ b/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts @@ -0,0 +1,34 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +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 LIKE 'WAIT_EXTERNAL%' THEN 'external' + WHEN blocker LIKE 'WAIT_NATURAL%' THEN 'natural' + WHEN blocker LIKE 'NEEDS_DECISION%' THEN 'decision' ELSE NULL END`; + 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 json_extract(NEW.payload_json, '$.clientRequestId') IS NOT NULL + 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) + 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`); + } +}); diff --git a/apps/server/src/persistence/WorkLedger.test.ts b/apps/server/src/persistence/WorkLedger.test.ts index 006b0b4532565..12edcf477dbf5 100644 --- a/apps/server/src/persistence/WorkLedger.test.ts +++ b/apps/server/src/persistence/WorkLedger.test.ts @@ -1,7 +1,12 @@ +import { ThreadId } from "@t3tools/contracts"; import { expect, it } from "@effect/vitest"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import LedgerMigration from "./ForkMigrations/007_WorkLedgerPortfolio.ts"; +import BaseLedgerMigration from "./ForkMigrations/002_WorkLedger.ts"; import * as TestClock from "effect/testing/TestClock"; import * as Sqlite from "./Sqlite.ts"; import * as WorkLedger from "./WorkLedger.ts"; @@ -27,7 +32,11 @@ it.effect( "session-a", ); yield* TestClock.setTime(1000); - const updated = yield* ledger.upsert({ key: "a", status: "done", blocker: null }, "thread-b"); + const { warnings, ...updated } = yield* ledger.upsert( + { key: "a", status: "done", blocker: null }, + "thread-b", + ); + expect(warnings).toEqual([]); expect(updated).toMatchObject({ title: first.title, ownerKind: first.ownerKind, @@ -71,5 +80,149 @@ it.effect( // Thread ids from create_threads can be longer than a display name. const longCaller = `thread:${"x".repeat(300)}`; expect((yield* ledger.upsert({ key: "a" }, longCaller)).updatedBy).toBe(longCaller); - }).pipe(Effect.provide(WorkLedger.layer.pipe(Layer.provide(Sqlite.layerMemory)))), + }).pipe(Effect.provide(WorkLedger.layer.pipe(Layer.provideMerge(Sqlite.layerMemory)))), +); + +it.effect( + "filters waits and due checks, normalizes links and records every upsert atomically", + () => + Effect.gen(function* () { + const ledger = yield* WorkLedger.WorkLedger; + yield* TestClock.setTime(1000); + const first = yield* ledger.upsert( + { + key: "wait", + title: "Wait", + ownerKind: "t3", + ownerName: "worker", + ownerThreadId: ThreadId.make("t3-thread://v1/missing"), + links: ["https://app.t3.codes/threads/env-1/also-missing"], + waitReason: "external", + nextCheckAt: "1970-01-01T01:00:00+01:00", + }, + "caller-a", + ); + expect(first.ownerThreadId).toBe("missing"); + expect(first.links).toEqual(["t3-thread://v2/env-1/also-missing"]); + expect(first.warnings).toHaveLength(2); + expect( + (yield* ledger.query({ due: true, waitReason: "external" })).items.map((item) => item.key), + ).toEqual(["wait"]); + expect((yield* ledger.query({ due: false })).items).toEqual([]); + const change = { key: "wait", waitReason: null, nextCheckAt: null } as const; + yield* ledger.upsert(change, "caller-b"); + yield* ledger.upsert({ key: "wait" }, "caller-b"); + const page = yield* ledger.history({ key: "wait", limit: 1 }); + expect(page.items[0]).toMatchObject({ + changedBy: "caller-a", + changes: { title: { before: null, after: "Wait" } }, + }); + const rest = yield* ledger.history({ key: "wait", cursor: page.nextCursor! }); + expect(rest.items[0]).toMatchObject({ + changedBy: "caller-b", + changes: { + waitReason: { before: "external", after: null }, + nextCheckAt: { before: "1970-01-01T00:00:00.000Z", after: null }, + }, + }); + expect(rest.items[1]!.changes).toEqual({}); + expect((yield* ledger.query({ waitReason: null })).items).toHaveLength(1); + expect((yield* ledger.query({ due: true })).items).toEqual([]); + const before = yield* ledger.history({ key: "wait" }); + yield* ledger.upsert({ key: "wait", nextCheckAt: "invalid" }, "caller-c").pipe(Effect.flip); + expect(yield* ledger.history({ key: "wait" })).toEqual(before); + }).pipe(Effect.provide(WorkLedger.layer.pipe(Layer.provideMerge(Sqlite.layerMemory)))), +); + +it.effect( + "joins bounded portfolio results and tracks terminal runs without changing ledger status", + () => + Effect.gen(function* () { + const ledger = yield* WorkLedger.WorkLedger; + const sql = yield* SqlClient.SqlClient; + yield* TestClock.setTime(0); + yield* ledger.upsert( + { + key: "run-key", + title: "Run", + ownerKind: "t3", + ownerName: "worker", + ownerThreadId: ThreadId.make("worker-thread"), + status: "blocked", + }, + "caller", + ); + yield* ledger.upsert( + { key: "done", title: "Done", ownerKind: "human", ownerName: "Chris", status: "done" }, + "caller", + ); + yield* sql`INSERT INTO orchestration_v2_projection_runs + (run_id, thread_id, ordinal, provider, provider_instance_id, status, requested_at, payload_json) + VALUES ('run-1', 'worker-thread', 1, 'codex', 'codex', 'running', '1970-01-01T00:00:00.000Z', '{"clientRequestId":"run-key"}')`; + expect( + (yield* ledger.query({})).items.find((item) => item.key === "run-key")!.lastRunId, + ).toBeNull(); + yield* sql`UPDATE orchestration_v2_projection_runs SET status = 'completed', completed_at = '1970-01-01T00:00:01.000Z' WHERE run_id = 'run-1'`; + yield* sql`INSERT INTO orchestration_v2_projection_messages + (message_id, thread_id, run_id, role, streaming, created_at, updated_at, payload_json) + VALUES ('result-1', 'worker-thread', 'run-1', 'assistant', 0, '1970-01-01T00:00:01.000Z', '1970-01-01T00:00:01.000Z', '{"text":"Completed the batch"}')`; + yield* sql`INSERT INTO orchestration_v2_projection_runtime_requests + (runtime_request_id, thread_id, node_id, kind, status, created_at, payload_json) + VALUES ('request-1', 'worker-thread', 'node-1', 'question', 'pending', '1970-01-01T00:00:01.000Z', '{}')`; + yield* TestClock.setTime(25 * 60 * 60 * 1000); + const portfolio = yield* ledger.portfolio({ limit: 1 }); + expect(portfolio.items).toHaveLength(1); + expect(portfolio.items[0]).toMatchObject({ + item: { + key: "run-key", + status: "blocked", + lastRunId: "run-1", + lastRunStatus: "completed", + lastRunAt: "1970-01-01T00:00:01.000Z", + }, + latestRunStatus: "completed", + pendingRequestCount: 1, + resultExcerpt: "Completed the batch", + staleWorker: true, + }); + yield* ledger.upsert( + { key: "z", title: "No thread", ownerKind: "human", ownerName: "Chris" }, + "caller", + ); + const page = yield* ledger.portfolio({ limit: 1 }); + expect(page.nextCursor).toBe("run-key"); + expect((yield* ledger.portfolio({ cursor: page.nextCursor! })).items[0]).toMatchObject({ + item: { key: "z" }, + staleWorker: false, + }); + yield* sql`UPDATE orchestration_v2_projection_runs SET completed_at = '1969-12-31T23:59:59.000Z', status = 'failed' WHERE run_id = 'run-1'`; + expect( + (yield* ledger.query({})).items.find((item) => item.key === "run-key")!.lastRunStatus, + ).toBe("completed"); + }).pipe(Effect.provide(WorkLedger.layer.pipe(Layer.provideMerge(Sqlite.layerMemory)))), +); + +it.effect("backfills only recognized blocker prefixes during migration", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* BaseLedgerMigration; + yield* sql`CREATE TABLE orchestration_v2_projection_runs (run_id TEXT, status TEXT, completed_at TEXT, requested_at TEXT, payload_json TEXT)`; + for (const [key, blocker] of [ + ["a", "WAIT_EXTERNAL: vendor"], + ["b", "WAIT_NATURAL event"], + ["c", "NEEDS_DECISION: approve"], + ["d", "Other"], + ]) { + yield* sql`INSERT INTO work_ledger (work_key, title, owner_kind, owner_name, status, blocker, links_json, created_at, updated_at, updated_by) + VALUES (${key!}, 'Title', 't3', 'worker', 'blocked', ${blocker!}, '[]', '1970-01-01', '1970-01-01', 'caller')`; + } + yield* LedgerMigration; + expect(yield* sql`SELECT wait_reason FROM work_ledger ORDER BY work_key`).toEqual([ + { wait_reason: "external" }, + { wait_reason: "natural" }, + { wait_reason: "decision" }, + { wait_reason: null }, + ]); + expect(yield* sql`SELECT * FROM work_ledger_history`).toEqual([]); + }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/apps/server/src/persistence/WorkLedger.ts b/apps/server/src/persistence/WorkLedger.ts index ba754f6257ea6..05a8d48445728 100644 --- a/apps/server/src/persistence/WorkLedger.ts +++ b/apps/server/src/persistence/WorkLedger.ts @@ -1,10 +1,18 @@ import { WorkLedgerItem, + RunId, + type WorkLedgerHistoryResult, + WorkLedgerHistoryEntry, + type WorkLedgerUpsertResult, + type WorkLedgerHistoryInput, + type PortfolioStatusInput, + type PortfolioStatusResult, type WorkLedgerUpsertInput, type WorkLedgerQueryInput, type WorkLedgerQueryResult, OrchestratorMcpFailure, } from "@t3tools/contracts"; +import { parseThreadLinkHref, formatThreadLink } from "@t3tools/shared/threadLinks"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -20,7 +28,13 @@ export class WorkLedger extends Context.Service< readonly upsert: ( input: WorkLedgerUpsertInput, updatedBy: string, - ) => Effect.Effect; + ) => Effect.Effect; + readonly history: ( + input: WorkLedgerHistoryInput, + ) => Effect.Effect; + readonly portfolio: ( + input: PortfolioStatusInput, + ) => Effect.Effect; readonly query: ( input: WorkLedgerQueryInput, ) => Effect.Effect; @@ -44,13 +58,41 @@ const storage = () => }); const make = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; - const select = sql`SELECT work_key AS key, title, owner_kind AS "ownerKind", owner_name AS "ownerName", owner_thread_id AS "ownerThreadId", status, blocker, next_action AS "nextAction", approval_note AS "approvalNote", links_json AS links, created_at AS "createdAt", updated_at AS "updatedAt", updated_by AS "updatedBy" FROM work_ledger`; + const select = sql`SELECT work_key AS key, title, owner_kind AS "ownerKind", owner_name AS "ownerName", owner_thread_id AS "ownerThreadId", status, wait_reason AS "waitReason", next_check_at AS "nextCheckAt", last_run_id AS "lastRunId", last_run_status AS "lastRunStatus", last_run_at AS "lastRunAt", blocker, next_action AS "nextAction", approval_note AS "approvalNote", links_json AS links, created_at AS "createdAt", updated_at AS "updatedAt", updated_by AS "updatedBy" FROM work_ledger`; const read = (key: string) => sql`${select} WHERE work_key = ${key}`.pipe( Effect.flatMap(decode), Effect.map((rows) => rows[0]), Effect.mapError(storage), ); + const query = Effect.fn("WorkLedger.query")(function* (input: WorkLedgerQueryInput) { + const limit = input.limit ?? 25; + const now = DateTime.formatIso(yield* DateTime.now); + // Stored times are UTC ISO strings, so an offset timestamp must be normalized to compare. + const since = + input.updatedSince === undefined + ? undefined + : Option.map(DateTime.make(input.updatedSince), DateTime.formatIso); + if (since !== undefined && Option.isNone(since)) + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "updatedSince must be an ISO-8601 time.", + }); + const rows = yield* sql`${select} WHERE 1 = 1 + ${input.waitReason === undefined ? sql`` : input.waitReason === null ? sql`AND wait_reason IS NULL` : sql`AND wait_reason = ${input.waitReason}`} + ${input.due === undefined ? sql`` : input.due ? sql`AND next_check_at <= ${now}` : sql`AND (next_check_at IS NULL OR next_check_at > ${now})`} + ${input.status === undefined ? sql`` : sql`AND status = ${input.status}`} + ${input.ownerKind === undefined ? sql`` : sql`AND owner_kind = ${input.ownerKind}`} + ${input.ownerName === undefined ? sql`` : sql`AND owner_name = ${input.ownerName}`} + ${since === undefined ? sql`` : sql`AND updated_at >= ${Option.getOrThrow(since)}`} + ${input.cursor === undefined ? sql`` : sql`AND work_key > ${input.cursor}`} + ORDER BY work_key ASC LIMIT ${limit + 1}`.pipe( + Effect.flatMap(decode), + Effect.mapError(storage), + ); + const items = rows.slice(0, limit); + return { items, nextCursor: rows.length > limit ? items.at(-1)!.key : null }; + }); return WorkLedger.of({ upsert: Effect.fn("WorkLedger.upsert")((input, updatedBy) => sql @@ -63,6 +105,11 @@ const make = Effect.gen(function* () { ); const candidate = { ownerThreadId: null, + waitReason: null, + nextCheckAt: null, + lastRunId: null, + lastRunStatus: null, + lastRunAt: null, status: "open", blocker: null, nextAction: null, @@ -74,6 +121,35 @@ const make = Effect.gen(function* () { updatedAt: now, updatedBy, }; + if (candidate.nextCheckAt !== null) { + const parsed = DateTime.make(candidate.nextCheckAt); + if (Option.isNone(parsed)) + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "nextCheckAt must be an ISO-8601 time.", + }); + candidate.nextCheckAt = DateTime.formatIso(parsed.value); + } + const warnings: string[] = []; + const threadIds = new Set(); + if (candidate.ownerThreadId !== null) { + const ref = parseThreadLinkHref(candidate.ownerThreadId); + if (ref !== null) candidate.ownerThreadId = ref.threadId; + threadIds.add(candidate.ownerThreadId); + } + candidate.links = candidate.links.map((link) => { + const href = link.match(/^\[[^\]]*\]\((.*)\)$/)?.[1] ?? link.replace(/^<|>$/g, ""); + const ref = parseThreadLinkHref(href); + if (ref === null) return link; + threadIds.add(ref.threadId); + return formatThreadLink(ref).slice(1, -1); + }); + for (const threadId of threadIds) { + const exists = + yield* sql`SELECT 1 FROM orchestration_v2_projection_threads WHERE thread_id = ${threadId} AND deleted_at IS NULL`; + if (exists.length === 0) + warnings.push(`Thread ${threadId} does not exist in this environment.`); + } const item = yield* decodeItem(candidate).pipe( Effect.mapError( () => @@ -84,40 +160,119 @@ const make = Effect.gen(function* () { ), ); const linksJson = yield* encodeLinks(item.links).pipe(Effect.mapError(storage)); - yield* sql`INSERT INTO work_ledger (work_key, title, owner_kind, owner_name, owner_thread_id, status, blocker, next_action, approval_note, links_json, created_at, updated_at, updated_by) - VALUES (${item.key}, ${item.title}, ${item.ownerKind}, ${item.ownerName}, ${item.ownerThreadId}, ${item.status}, ${item.blocker}, ${item.nextAction}, ${item.approvalNote}, ${linksJson}, ${item.createdAt}, ${item.updatedAt}, ${item.updatedBy}) - ON CONFLICT(work_key) DO UPDATE SET title = excluded.title, owner_kind = excluded.owner_kind, owner_name = excluded.owner_name, owner_thread_id = excluded.owner_thread_id, status = excluded.status, blocker = excluded.blocker, next_action = excluded.next_action, approval_note = excluded.approval_note, links_json = excluded.links_json, updated_at = excluded.updated_at, updated_by = excluded.updated_by`.pipe( - Effect.mapError(storage), + const row = { + work_key: item.key, + title: item.title, + owner_kind: item.ownerKind, + owner_name: item.ownerName, + owner_thread_id: item.ownerThreadId, + status: item.status, + blocker: item.blocker, + wait_reason: item.waitReason, + next_check_at: item.nextCheckAt, + next_action: item.nextAction, + approval_note: item.approvalNote, + links_json: linksJson, + created_at: item.createdAt, + updated_at: item.updatedAt, + updated_by: item.updatedBy, + }; + yield* sql`INSERT INTO work_ledger ${sql.insert(row)} ON CONFLICT(work_key) DO UPDATE SET + title=excluded.title, owner_kind=excluded.owner_kind, owner_name=excluded.owner_name, + owner_thread_id=excluded.owner_thread_id, status=excluded.status, blocker=excluded.blocker, + wait_reason=excluded.wait_reason, next_check_at=excluded.next_check_at, + next_action=excluded.next_action, approval_note=excluded.approval_note, links_json=excluded.links_json, + updated_at=excluded.updated_at, updated_by=excluded.updated_by`; + const changes = Object.fromEntries( + Object.entries(item) + .filter( + ([field, value]) => + !["createdAt", "updatedAt", "updatedBy"].includes(field) && + JSON.stringify(current?.[field as keyof WorkLedgerItem]) !== + JSON.stringify(value), + ) + .map(([field, value]) => [ + field, + { before: current?.[field as keyof WorkLedgerItem] ?? null, after: value }, + ]), ); - return item; + const changesJson = yield* Schema.encodeEffect( + Schema.fromJsonString(WorkLedgerHistoryEntry.fields.changes), + )(changes); + yield* sql`INSERT INTO work_ledger_history (work_key, changed_at, changed_by, changes_json) + VALUES (${item.key}, ${now}, ${updatedBy}, ${changesJson})`; + return { ...item, warnings }; }), ) .pipe(Effect.mapError((error) => (isFailure(error) ? error : storage()))), ), - query: Effect.fn("WorkLedger.query")(function* (input) { + query, + history: Effect.fn("WorkLedger.history")(function* (input) { const limit = input.limit ?? 25; - // Stored times are UTC ISO strings, so an offset timestamp must be normalized to compare. - const since = - input.updatedSince === undefined - ? undefined - : Option.map(DateTime.make(input.updatedSince), DateTime.formatIso); - if (since !== undefined && Option.isNone(since)) - return yield* new OrchestratorMcpFailure({ - code: "invalid_request", - message: "updatedSince must be an ISO-8601 time.", - }); - const rows = yield* sql`${select} WHERE 1 = 1 - ${input.status === undefined ? sql`` : sql`AND status = ${input.status}`} - ${input.ownerKind === undefined ? sql`` : sql`AND owner_kind = ${input.ownerKind}`} - ${input.ownerName === undefined ? sql`` : sql`AND owner_name = ${input.ownerName}`} - ${since === undefined ? sql`` : sql`AND updated_at >= ${Option.getOrThrow(since)}`} + const rows = + yield* sql`SELECT id, work_key AS key, changed_at AS "changedAt", changed_by AS "changedBy", changes_json AS changes + FROM work_ledger_history WHERE work_key = ${input.key} AND id > ${input.cursor ?? 0} + ORDER BY id LIMIT ${limit + 1}`; + const decoded = yield* Schema.decodeUnknownEffect( + Schema.Array( + WorkLedgerHistoryEntry.mapFields( + Struct.assign({ + changes: Schema.fromJsonString(WorkLedgerHistoryEntry.fields.changes), + }), + ), + ), + )(rows); + const items = decoded.slice(0, limit); + return { items, nextCursor: decoded.length > limit ? items.at(-1)!.id : null }; + }, Effect.mapError(storage)), + portfolio: Effect.fn("WorkLedger.portfolio")(function* (input) { + const limit = input.limit ?? 10; + const ledgerRows = yield* sql`${select} WHERE status NOT IN ('done', 'cancelled') ${input.cursor === undefined ? sql`` : sql`AND work_key > ${input.cursor}`} - ORDER BY work_key ASC LIMIT ${limit + 1}`.pipe( + ORDER BY work_key LIMIT ${limit + 1}`.pipe( Effect.flatMap(decode), Effect.mapError(storage), ); - const items = rows.slice(0, limit); - return { items, nextCursor: rows.length > limit ? items.at(-1)!.key : null }; + const now = DateTime.toEpochMillis(yield* DateTime.now); + const items = yield* Effect.forEach(ledgerRows.slice(0, limit), (item) => + Effect.gen(function* () { + if (item.ownerThreadId === null) + return { + item, + latestRunId: null, + latestRunStatus: null, + pendingRequestCount: 0, + resultExcerpt: null, + staleWorker: false, + }; + const [run] = yield* sql<{ + run_id: string; + status: string; + requested_at: string; + }>`SELECT run_id, status, requested_at + FROM orchestration_v2_projection_runs WHERE thread_id = ${item.ownerThreadId} ORDER BY ordinal DESC LIMIT 1`; + const [pending] = yield* sql<{ + count: number; + }>`SELECT COUNT(*) AS count FROM orchestration_v2_projection_runtime_requests + WHERE thread_id = ${item.ownerThreadId} AND status = 'pending'`; + const [result] = yield* sql<{ + text: string; + }>`SELECT substr(json_extract(payload_json, '$.text'), 1, 400) AS text + FROM orchestration_v2_projection_messages WHERE thread_id = ${item.ownerThreadId} AND role = 'assistant' + ${run === undefined ? sql`AND 0` : sql`AND run_id = ${run.run_id}`} + ORDER BY created_at DESC, message_id DESC LIMIT 1`; + return { + item, + latestRunId: run === undefined ? null : RunId.make(run.run_id), + latestRunStatus: run?.status ?? null, + pendingRequestCount: pending?.count ?? 0, + resultExcerpt: result?.text ?? null, + staleWorker: + now - Date.parse(run?.requested_at ?? item.createdAt) > 24 * 60 * 60 * 1000, + }; + }).pipe(Effect.mapError(storage)), + ); + return { items, nextCursor: ledgerRows.length > limit ? items.at(-1)!.item.key : null }; }), }); }); diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index ee957cd5534df..ff4053d5d64d5 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -684,6 +684,7 @@ export type OrchestrationV2ThreadLaunchWorkspaceStrategy = export const ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE = "workspace_preparation_failed"; export const OrchestrationV2Run = Schema.Struct({ + clientRequestId: Schema.optional(TrimmedNonEmptyString), id: RunId, threadId: ThreadId, ordinal: PositiveInt, @@ -3191,6 +3192,7 @@ export const OrchestrationV2Command = Schema.Union([ }), Schema.Struct({ type: Schema.Literal("message.dispatch"), + clientRequestId: Schema.optional(TrimmedNonEmptyString), notification: Schema.optional(OrchestrationV2Notification), /** Sends this pending guest message; the server uses the guest's text, not `text`. */ guestMessageId: Schema.optional(GuestMessageId), @@ -3366,6 +3368,7 @@ export const OrchestrationV2Command = Schema.Union([ }), Schema.Struct({ type: Schema.Literal("delegated_task.request"), + clientRequestId: Schema.optional(TrimmedNonEmptyString), ...OrchestrationV2CreationFields, commandId: CommandId, parentThreadId: ThreadId, diff --git a/packages/contracts/src/workLedger.ts b/packages/contracts/src/workLedger.ts index 9842dc6bd1fde..714a3a7f1f4cf 100644 --- a/packages/contracts/src/workLedger.ts +++ b/packages/contracts/src/workLedger.ts @@ -1,6 +1,12 @@ import * as Struct from "effect/Struct"; import * as Schema from "effect/Schema"; -import { IsoDateTime, NonNegativeInt, ThreadId, TrimmedNonEmptyString } from "./baseSchemas.ts"; +import { + IsoDateTime, + NonNegativeInt, + ThreadId, + RunId, + TrimmedNonEmptyString, +} from "./baseSchemas.ts"; const text = Schema.String.check(Schema.isMaxLength(8000)); const name = TrimmedNonEmptyString.check(Schema.isMaxLength(256)); @@ -11,6 +17,12 @@ export const WorkLedgerStatus = Schema.Literals([ "done", "cancelled", ]); +export const WorkLedgerWaitReason = Schema.NullOr( + Schema.Literals(["external", "natural", "decision"]), +); +const pageLimit = Schema.optional( + NonNegativeInt.check(Schema.isBetween({ minimum: 1, maximum: 100 })), +); const ownerKind = Schema.Literals(["jarvis", "vision", "t3", "human"]); const fields = { title: TrimmedNonEmptyString.check(Schema.isMaxLength(512)), @@ -19,6 +31,8 @@ const fields = { ownerThreadId: Schema.NullOr(ThreadId), status: WorkLedgerStatus, blocker: Schema.NullOr(text), + waitReason: WorkLedgerWaitReason, + nextCheckAt: Schema.NullOr(IsoDateTime), nextAction: Schema.NullOr(text), approvalNote: Schema.NullOr(text), links: Schema.Array(Schema.String.check(Schema.isMaxLength(4096))).check(Schema.isMaxLength(100)), @@ -26,6 +40,9 @@ const fields = { export const WorkLedgerItem = Schema.Struct({ key: name, ...fields, + lastRunId: Schema.NullOr(RunId), + lastRunStatus: Schema.NullOr(Schema.String), + lastRunAt: Schema.NullOr(IsoDateTime), createdAt: IsoDateTime, updatedAt: IsoDateTime, // A thread or session id, which can be longer than a display name. @@ -42,8 +59,10 @@ export const WorkLedgerQueryInput = Schema.Struct({ ownerKind: Schema.optional(ownerKind), ownerName: Schema.optional(name), updatedSince: Schema.optional(IsoDateTime), + waitReason: Schema.optional(WorkLedgerWaitReason), + due: Schema.optional(Schema.Boolean), cursor: Schema.optional(name), - limit: Schema.optional(NonNegativeInt.check(Schema.isBetween({ minimum: 1, maximum: 100 }))), + limit: pageLimit, }); export type WorkLedgerQueryInput = typeof WorkLedgerQueryInput.Type; export const WorkLedgerQueryResult = Schema.Struct({ @@ -51,3 +70,48 @@ export const WorkLedgerQueryResult = Schema.Struct({ nextCursor: Schema.NullOr(name), }); export type WorkLedgerQueryResult = typeof WorkLedgerQueryResult.Type; + +export const WorkLedgerUpsertResult = WorkLedgerItem.mapFields( + Struct.assign({ warnings: Schema.Array(Schema.String) }), +); +export type WorkLedgerUpsertResult = typeof WorkLedgerUpsertResult.Type; +export const WorkLedgerHistoryInput = Schema.Struct({ + key: name, + cursor: Schema.optional(NonNegativeInt), + limit: pageLimit, +}); +export type WorkLedgerHistoryInput = typeof WorkLedgerHistoryInput.Type; +export const WorkLedgerHistoryEntry = Schema.Struct({ + id: NonNegativeInt, + key: name, + changedAt: IsoDateTime, + changedBy: TrimmedNonEmptyString, + changes: Schema.Record( + Schema.String, + Schema.Struct({ before: Schema.Unknown, after: Schema.Unknown }), + ), +}); +export const WorkLedgerHistoryResult = Schema.Struct({ + items: Schema.Array(WorkLedgerHistoryEntry), + nextCursor: Schema.NullOr(NonNegativeInt), +}); +export type WorkLedgerHistoryResult = typeof WorkLedgerHistoryResult.Type; +export const PortfolioStatusInput = Schema.Struct({ + cursor: Schema.optional(name), + limit: pageLimit, +}); +export type PortfolioStatusInput = typeof PortfolioStatusInput.Type; +export const PortfolioStatusResult = Schema.Struct({ + items: Schema.Array( + Schema.Struct({ + item: WorkLedgerItem, + latestRunId: Schema.NullOr(RunId), + latestRunStatus: Schema.NullOr(Schema.String), + pendingRequestCount: NonNegativeInt, + resultExcerpt: Schema.NullOr(Schema.String), + staleWorker: Schema.Boolean, + }), + ), + nextCursor: Schema.NullOr(name), +}); +export type PortfolioStatusResult = typeof PortfolioStatusResult.Type; From 21beb0f034de1e1f3d229d141520d4e0a5b60f56 Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Wed, 11 Nov 2026 23:29:31 -0700 Subject: [PATCH 02/10] test(ledger): cover history due checks and run outcomes --- .../ForkMigrations/007_WorkLedgerPortfolio.ts | 6 ++--- .../server/src/persistence/WorkLedger.test.ts | 2 ++ apps/server/src/persistence/WorkLedger.ts | 26 ++++++++++--------- .../src/provider/agentContextBudget.json | 2 +- ...012-ledger-fields-history-and-portfolio.md | 19 ++++++++++++++ 5 files changed, 39 insertions(+), 16 deletions(-) create mode 100644 docs/jarvis/decisions/0012-ledger-fields-history-and-portfolio.md diff --git a/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts b/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts index de7d3875646fa..90e67a209521a 100644 --- a/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts +++ b/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts @@ -9,9 +9,9 @@ export default Effect.gen(function* () { 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 LIKE 'WAIT_EXTERNAL%' THEN 'external' - WHEN blocker LIKE 'WAIT_NATURAL%' THEN 'natural' - WHEN blocker LIKE 'NEEDS_DECISION%' THEN 'decision' ELSE NULL END`; + WHEN blocker GLOB 'WAIT_EXTERNAL*' THEN 'external' + WHEN blocker GLOB 'WAIT_NATURAL*' THEN 'natural' + WHEN blocker GLOB 'NEEDS_DECISION*' THEN 'decision' ELSE NULL END`; 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, diff --git a/apps/server/src/persistence/WorkLedger.test.ts b/apps/server/src/persistence/WorkLedger.test.ts index 12edcf477dbf5..0f0875dc2cffb 100644 --- a/apps/server/src/persistence/WorkLedger.test.ts +++ b/apps/server/src/persistence/WorkLedger.test.ts @@ -212,6 +212,7 @@ it.effect("backfills only recognized blocker prefixes during migration", () => ["b", "WAIT_NATURAL event"], ["c", "NEEDS_DECISION: approve"], ["d", "Other"], + ["e", "WAITXEXTERNAL"], ]) { yield* sql`INSERT INTO work_ledger (work_key, title, owner_kind, owner_name, status, blocker, links_json, created_at, updated_at, updated_by) VALUES (${key!}, 'Title', 't3', 'worker', 'blocked', ${blocker!}, '[]', '1970-01-01', '1970-01-01', 'caller')`; @@ -222,6 +223,7 @@ it.effect("backfills only recognized blocker prefixes during migration", () => { wait_reason: "natural" }, { wait_reason: "decision" }, { wait_reason: null }, + { wait_reason: null }, ]); expect(yield* sql`SELECT * FROM work_ledger_history`).toEqual([]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), diff --git a/apps/server/src/persistence/WorkLedger.ts b/apps/server/src/persistence/WorkLedger.ts index 05a8d48445728..bf6af7141a50f 100644 --- a/apps/server/src/persistence/WorkLedger.ts +++ b/apps/server/src/persistence/WorkLedger.ts @@ -48,6 +48,18 @@ const decode = Schema.decodeUnknownEffect( ), ), ); +const decodeHistory = Schema.decodeUnknownEffect( + Schema.Array( + WorkLedgerHistoryEntry.mapFields( + Struct.assign({ + changes: Schema.fromJsonString(WorkLedgerHistoryEntry.fields.changes), + }), + ), + ), +); +const encodeChanges = Schema.encodeEffect( + Schema.fromJsonString(WorkLedgerHistoryEntry.fields.changes), +); const encodeLinks = Schema.encodeEffect(Schema.fromJsonString(WorkLedgerItem.fields.links)); const isFailure = Schema.is(OrchestratorMcpFailure); const decodeItem = Schema.decodeUnknownEffect(WorkLedgerItem); @@ -196,9 +208,7 @@ const make = Effect.gen(function* () { { before: current?.[field as keyof WorkLedgerItem] ?? null, after: value }, ]), ); - const changesJson = yield* Schema.encodeEffect( - Schema.fromJsonString(WorkLedgerHistoryEntry.fields.changes), - )(changes); + const changesJson = yield* encodeChanges(changes); yield* sql`INSERT INTO work_ledger_history (work_key, changed_at, changed_by, changes_json) VALUES (${item.key}, ${now}, ${updatedBy}, ${changesJson})`; return { ...item, warnings }; @@ -213,15 +223,7 @@ const make = Effect.gen(function* () { yield* sql`SELECT id, work_key AS key, changed_at AS "changedAt", changed_by AS "changedBy", changes_json AS changes FROM work_ledger_history WHERE work_key = ${input.key} AND id > ${input.cursor ?? 0} ORDER BY id LIMIT ${limit + 1}`; - const decoded = yield* Schema.decodeUnknownEffect( - Schema.Array( - WorkLedgerHistoryEntry.mapFields( - Struct.assign({ - changes: Schema.fromJsonString(WorkLedgerHistoryEntry.fields.changes), - }), - ), - ), - )(rows); + const decoded = yield* decodeHistory(rows); const items = decoded.slice(0, limit); return { items, nextCursor: decoded.length > limit ? items.at(-1)!.id : null }; }, Effect.mapError(storage)), diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index a908df52c7576..f8b9adf6978b0 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5597, "codexAdditionalContextBytes": 7327, - "mcpToolListJsonBytes": 51125 + "mcpToolListJsonBytes": 52042 } diff --git a/docs/jarvis/decisions/0012-ledger-fields-history-and-portfolio.md b/docs/jarvis/decisions/0012-ledger-fields-history-and-portfolio.md new file mode 100644 index 0000000000000..11aa7aff26d71 --- /dev/null +++ b/docs/jarvis/decisions/0012-ledger-fields-history-and-portfolio.md @@ -0,0 +1,19 @@ +# Ledger fields, history and portfolio + +Accepted 2026-11-12. + +## Decision + +The ledger separates a work item's status from its reason for waiting. External waits depend on a vendor or another party. Natural waits depend on a real event. Decision waits need Chris's answer. Existing blocker prefixes become wait reasons during migration. A due check means the recorded next check time has arrived. It does not start work. + +Every upsert retains its caller, time, and changed values. A run started with a client request id equal to a work item's key records its terminal outcome on that item without changing the item's status. Finishing a run does not prove that the work is done. + +The portfolio joins open work with each owner thread's latest run, pending requests, and a short result excerpt. A worker with no run for over 24 hours is flagged for attention. That flag does not establish failure or authorize a restart. The web panel reads the same ledger and provides no editing controls. + +## Alternatives + +Keep wait prefixes in free text, infer completion from a successful run, or make the portfolio a second source of work state. + +## Why + +Structured waits keep vendor delays and real-world events out of Chris's decision queue. History makes coordination changes traceable. Run metadata and the portfolio show activity while leaving business completion with the work item's owner. From 5334cb4125e0e04049aa2af8dcbc82fea3dd52d1 Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Wed, 11 Nov 2026 23:37:47 -0700 Subject: [PATCH 03/10] feat(web): add a read-only work ledger panel --- apps/server/src/auth/RpcAuthorization.test.ts | 6 +- apps/server/src/auth/RpcAuthorization.ts | 1 + .../src/observability/RpcInstrumentation.ts | 1 + apps/server/src/ws.ts | 3 + .../settings/SettingsSidebarNav.tsx | 1 + .../settings/WorkLedgerSettings.tsx | 120 ++++++++++++++++++ .../src/components/settings/settingsSearch.ts | 11 ++ apps/web/src/routeTree.gen.ts | 21 +++ apps/web/src/routes/_chat.settings.tsx | 6 +- .../src/routes/_chat.settings.work-ledger.tsx | 6 + packages/client-runtime/src/state/server.ts | 4 + packages/contracts/src/rpc.ts | 10 ++ 12 files changed, 188 insertions(+), 2 deletions(-) create mode 100644 apps/web/src/components/settings/WorkLedgerSettings.tsx create mode 100644 apps/web/src/routes/_chat.settings.work-ledger.tsx diff --git a/apps/server/src/auth/RpcAuthorization.test.ts b/apps/server/src/auth/RpcAuthorization.test.ts index aab7445e9577e..863a016836291 100644 --- a/apps/server/src/auth/RpcAuthorization.test.ts +++ b/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); } }); diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 180f053f111f4..d041219669ea1 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/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, diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index f9a861cca23fb..a44c8ee92954c 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/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", diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 5257da599c4f2..e8dd4c0352fff 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1,3 +1,4 @@ +import * as WorkLedger from "./persistence/WorkLedger.ts"; import { OrchestrationDispatchCommandError } from "@t3tools/contracts"; import * as Context from "effect/Context"; import * as SecureRandom from "effect/crypto/SecureRandom"; @@ -1471,6 +1472,7 @@ export const layerWsRpc = ( ), Effect.orElseSucceed(() => null), ); + const ledger = yield* WorkLedger.WorkLedger.pipe(Effect.provide(WorkLedger.layer)); const usage = yield* UsageService.UsageService; const usageLimitSources = yield* UsageLimitSources.UsageLimitSources; const worktreeSetupTracker = yield* WorktreeSetupTracker.WorktreeSetupTracker; @@ -2634,6 +2636,7 @@ export const layerWsRpc = ( processResourceMonitor.readHistory(input), [WS_METHODS.serverGetResourceTelemetryHistory]: (input) => resourceTelemetry.readHistory(input), + [WS_METHODS.serverQueryWorkLedger]: (input) => ledger.query(input), [WS_METHODS.serverGetUsageSummary]: (input) => usage.readSummary(input), [WS_METHODS.serverRefreshUsageRates]: (_input) => usage.refreshRates, [WS_METHODS.serverRetryResourceTelemetry]: (_input) => resourceTelemetry.retry, diff --git a/apps/web/src/components/settings/SettingsSidebarNav.tsx b/apps/web/src/components/settings/SettingsSidebarNav.tsx index c34c48694ae87..efece3b57f1ca 100644 --- a/apps/web/src/components/settings/SettingsSidebarNav.tsx +++ b/apps/web/src/components/settings/SettingsSidebarNav.tsx @@ -101,6 +101,7 @@ const SETTINGS_SECTION_ICONS: Readonly< "/settings/threads": MessagesSquareIcon, "/settings/browser": GlobeIcon, "/settings/devices": SmartphoneIcon, + "/settings/work-ledger": MessagesSquareIcon, "/settings/scheduled-tasks": CalendarClockIcon, "/settings/source-control": GitBranchIcon, "/settings/worktrees": FolderTreeIcon, diff --git a/apps/web/src/components/settings/WorkLedgerSettings.tsx b/apps/web/src/components/settings/WorkLedgerSettings.tsx new file mode 100644 index 0000000000000..e49fae2ac991c --- /dev/null +++ b/apps/web/src/components/settings/WorkLedgerSettings.tsx @@ -0,0 +1,120 @@ +import { useState } from "react"; +import { Link } from "@tanstack/react-router"; +import type { EnvironmentId } from "@t3tools/contracts"; +import { scopeThreadRef } from "@t3tools/client-runtime/environment"; +import { buildThreadRouteParams } from "../../threadRoutes"; +import { useEnvironmentQuery } from "../../state/query"; +import { serverEnvironment } from "../../state/server"; +import { Button } from "../ui/button"; +import { SettingsPageContainer, SettingsSection } from "./settingsLayout"; +import { useSettingsScope } from "./SettingsScopeContext"; + +function LedgerList({ environmentId }: { readonly environmentId: EnvironmentId }) { + const [cursors, setCursors] = useState([]); + const cursor = cursors.at(-1); + const query = useEnvironmentQuery( + serverEnvironment.workLedger({ + environmentId, + input: { limit: 25, ...(cursor === undefined ? {} : { cursor }) }, + }), + ); + return ( + + Refresh + + } + > +

+ Read-only view of shared work and its next action. +

+ {query.error && ( +

+ {query.error} +

+ )} + {query.isPending && ( +

+ Loading work… +

+ )} + {query.data && ( + <> + {query.data.items.length === 0 && ( +

No work items.

+ )} +
    + {query.data.items.map((item) => ( +
  • +
    + {item.title} + + {item.status.replaceAll("_", " ")} + +
    +

    + {item.ownerName} · {item.ownerKind} + {item.waitReason ? ` · Waiting: ${item.waitReason}` : ""} +

    +

    + {item.nextAction || "No next action recorded."} +

    + {item.ownerThreadId && ( + + Open owner thread + + )} +
    • + ))} +
+
+ + +
+ + )} + + ); +} + +export function WorkLedgerSettings() { + const { environment, connectedEnvironments } = useSettingsScope(); + const selected = environment ?? connectedEnvironments[0]; + return ( + + {selected ? ( + + ) : ( +

+ Connect to an environment to read its work ledger. +

+ )} + + ); +} diff --git a/apps/web/src/components/settings/settingsSearch.ts b/apps/web/src/components/settings/settingsSearch.ts index 13f8c9bfc7f72..937cf6f96f471 100644 --- a/apps/web/src/components/settings/settingsSearch.ts +++ b/apps/web/src/components/settings/settingsSearch.ts @@ -30,6 +30,7 @@ export type SettingsPath = | "/settings/threads" | "/settings/browser" | "/settings/devices" + | "/settings/work-ledger" | "/settings/scheduled-tasks" | "/settings/source-control" | "/settings/worktrees" @@ -105,6 +106,7 @@ export const SETTINGS_SECTION_LABELS: Readonly> = { "/settings/threads": msg("Threads"), "/settings/browser": msg("Browser"), "/settings/devices": msg("Devices"), + "/settings/work-ledger": msg("Work ledger"), "/settings/scheduled-tasks": msg("Scheduled tasks"), "/settings/source-control": msg("Source control"), "/settings/worktrees": msg("Worktrees"), @@ -142,6 +144,7 @@ export const SETTINGS_NAV_GROUPS: ReadonlyArray<{ "/settings/threads", "/settings/browser", "/settings/devices", + "/settings/work-ledger", "/settings/scheduled-tasks", ], }, @@ -225,6 +228,13 @@ const SOURCE_CONTROL_HOST_TERMS = sourceControlClients.definitions .join(" "); export const SETTINGS_SEARCH_ITEMS = [ + { + id: "work-ledger", + title: "Work ledger", + to: "/settings/work-ledger", + environmentOnly: true, + searchTerms: ["shared work portfolio owner wait reason next action jarvis"], + }, { id: "storage-worktree-keep-when", targetId: "storage-worktrees", @@ -1362,6 +1372,7 @@ const SETTINGS_CATEGORY_SCOPES: Readonly ChatSettingsRoute, } as any) +const ChatSettingsWorkLedgerRoute = ChatSettingsWorkLedgerRouteImport.update({ + id: '/work-ledger', + path: '/work-ledger', + getParentRoute: () => ChatSettingsRoute, +} as any) const ChatSettingsVoiceRoute = ChatSettingsVoiceRouteImport.update({ id: '/voice', path: '/voice', @@ -315,6 +321,7 @@ export interface FileRoutesByFullPath { '/settings/storage': typeof ChatSettingsStorageRoute '/settings/threads': typeof ChatSettingsThreadsRoute '/settings/voice': typeof ChatSettingsVoiceRoute + '/settings/work-ledger': typeof ChatSettingsWorkLedgerRoute '/settings/worktrees': typeof ChatSettingsWorktreesRoute '/threads/$environmentId/$threadId': typeof ThreadsEnvironmentIdThreadIdRoute } @@ -359,6 +366,7 @@ export interface FileRoutesByTo { '/settings/storage': typeof ChatSettingsStorageRoute '/settings/threads': typeof ChatSettingsThreadsRoute '/settings/voice': typeof ChatSettingsVoiceRoute + '/settings/work-ledger': typeof ChatSettingsWorkLedgerRoute '/settings/worktrees': typeof ChatSettingsWorktreesRoute '/threads/$environmentId/$threadId': typeof ThreadsEnvironmentIdThreadIdRoute } @@ -405,6 +413,7 @@ export interface FileRoutesById { '/_chat/settings/storage': typeof ChatSettingsStorageRoute '/_chat/settings/threads': typeof ChatSettingsThreadsRoute '/_chat/settings/voice': typeof ChatSettingsVoiceRoute + '/_chat/settings/work-ledger': typeof ChatSettingsWorkLedgerRoute '/_chat/settings/worktrees': typeof ChatSettingsWorktreesRoute '/threads/$environmentId/$threadId': typeof ThreadsEnvironmentIdThreadIdRoute } @@ -451,6 +460,7 @@ export interface FileRouteTypes { | '/settings/storage' | '/settings/threads' | '/settings/voice' + | '/settings/work-ledger' | '/settings/worktrees' | '/threads/$environmentId/$threadId' fileRoutesByTo: FileRoutesByTo @@ -495,6 +505,7 @@ export interface FileRouteTypes { | '/settings/storage' | '/settings/threads' | '/settings/voice' + | '/settings/work-ledger' | '/settings/worktrees' | '/threads/$environmentId/$threadId' id: @@ -540,6 +551,7 @@ export interface FileRouteTypes { | '/_chat/settings/storage' | '/_chat/settings/threads' | '/_chat/settings/voice' + | '/_chat/settings/work-ledger' | '/_chat/settings/worktrees' | '/threads/$environmentId/$threadId' fileRoutesById: FileRoutesById @@ -663,6 +675,13 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof ChatSettingsWorktreesRouteImport parentRoute: typeof ChatSettingsRoute } + '/_chat/settings/work-ledger': { + id: '/_chat/settings/work-ledger' + path: '/work-ledger' + fullPath: '/settings/work-ledger' + preLoaderRoute: typeof ChatSettingsWorkLedgerRouteImport + parentRoute: typeof ChatSettingsRoute + } '/_chat/settings/voice': { id: '/_chat/settings/voice' path: '/voice' @@ -889,6 +908,7 @@ interface ChatSettingsRouteChildren { ChatSettingsStorageRoute: typeof ChatSettingsStorageRoute ChatSettingsThreadsRoute: typeof ChatSettingsThreadsRoute ChatSettingsVoiceRoute: typeof ChatSettingsVoiceRoute + ChatSettingsWorkLedgerRoute: typeof ChatSettingsWorkLedgerRoute ChatSettingsWorktreesRoute: typeof ChatSettingsWorktreesRoute } @@ -919,6 +939,7 @@ const ChatSettingsRouteChildren: ChatSettingsRouteChildren = { ChatSettingsStorageRoute: ChatSettingsStorageRoute, ChatSettingsThreadsRoute: ChatSettingsThreadsRoute, ChatSettingsVoiceRoute: ChatSettingsVoiceRoute, + ChatSettingsWorkLedgerRoute: ChatSettingsWorkLedgerRoute, ChatSettingsWorktreesRoute: ChatSettingsWorktreesRoute, } diff --git a/apps/web/src/routes/_chat.settings.tsx b/apps/web/src/routes/_chat.settings.tsx index 7758c9ca55919..19ec27f5368df 100644 --- a/apps/web/src/routes/_chat.settings.tsx +++ b/apps/web/src/routes/_chat.settings.tsx @@ -150,7 +150,11 @@ function SettingsRouteLayout() { return ( { // Send every axis so retain middleware recognizes "all" as an explicit target. void navigate({ diff --git a/apps/web/src/routes/_chat.settings.work-ledger.tsx b/apps/web/src/routes/_chat.settings.work-ledger.tsx new file mode 100644 index 0000000000000..4da24b6c180a0 --- /dev/null +++ b/apps/web/src/routes/_chat.settings.work-ledger.tsx @@ -0,0 +1,6 @@ +import { createFileRoute } from "@tanstack/react-router"; +import { WorkLedgerSettings } from "../components/settings/WorkLedgerSettings"; + +export const Route = createFileRoute("/_chat/settings/work-ledger")({ + component: WorkLedgerSettings, +}); diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index a935272d3d40f..09e707a586c46 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -1386,6 +1386,10 @@ export function createServerEnvironmentAtoms( execute: (input: EnvironmentRpcInput) => request(WS_METHODS.serverGetHostResources, input).pipe(Effect.timeout("5 seconds")), }), + workLedger: createEnvironmentRpcQueryAtomFamily(runtime, { + label: "environment-data:server:work-ledger", + tag: WS_METHODS.serverQueryWorkLedger, + }), processResourceHistory: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:server:process-resource-history", tag: WS_METHODS.serverGetProcessResourceHistory, diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 0f6af19188e18..a940d94bbe12b 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -1,3 +1,5 @@ +import { WorkLedgerQueryInput, WorkLedgerQueryResult } from "./workLedger.ts"; +import { OrchestratorMcpFailure } from "./orchestratorMcp.ts"; import { OrchestrationV2SearchThreadError, OrchestrationV2SearchThreadInput, @@ -546,6 +548,7 @@ export const WS_METHODS = { serverSignalProcess: "server.signalProcess", serverReportClientActivity: "server.reportClientActivity", serverGetUsageSummary: "server.getUsageSummary", + serverQueryWorkLedger: "server.queryWorkLedger", serverRefreshUsageRates: "server.refreshUsageRates", scheduledTasksSubscribe: "scheduledTasks.subscribe", @@ -994,6 +997,12 @@ const WsServerPrepareTriageRpc = Rpc.make(WS_METHODS.serverPrepareTriage, { error: Schema.Union([ServerPrepareTriageError, EnvironmentAuthorizationError]), }); +const WsServerQueryWorkLedgerRpc = Rpc.make(WS_METHODS.serverQueryWorkLedger, { + payload: WorkLedgerQueryInput, + success: WorkLedgerQueryResult, + error: Schema.Union([EnvironmentAuthorizationError, OrchestratorMcpFailure]), +}); + const WsServerGetUsageSummaryRpc = Rpc.make(WS_METHODS.serverGetUsageSummary, { payload: UsageSummaryInput, success: UsageSummary, @@ -2121,6 +2130,7 @@ export const WsRpcGroup = RpcGroup.make( WsServerCheckOtlpEndpointRpc, WsServerListTextGeneratorModelsRpc, WsServerPrepareTriageRpc, + WsServerQueryWorkLedgerRpc, WsServerGetUsageSummaryRpc, WsServerRefreshUsageRatesRpc, WsServerSignalProcessRpc, From fd639fda8dca8f9670e6349d6a3496046545364e Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Wed, 11 Nov 2026 23:40:24 -0700 Subject: [PATCH 04/10] fix(ledger): allow Claude read-only history and portfolio queries --- apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 77d3bf51eb7ab..ff47670a26fc1 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -2154,6 +2154,8 @@ export const CLAUDE_READ_ONLY_T3_MCP_ALLOWED_TOOLS: ReadonlyArray = [ "mcp__t3-code__t3_thread_list", "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", From 6f06ba8fda7eef0a2dee6ce90232e91a82c68c3f Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Wed, 11 Nov 2026 23:58:35 -0700 Subject: [PATCH 05/10] fix(ledger): retain scoped links and persist an explicit post marker --- apps/server/src/orchestration-v2/Orchestrator.ts | 1 + .../server/src/orchestration-v2/runtimeLayer.test.ts | 2 ++ apps/server/src/persistence/WorkLedger.test.ts | 11 +++++++++++ apps/server/src/persistence/WorkLedger.ts | 9 ++++++++- .../0012-ledger-fields-history-and-portfolio.md | 2 ++ docs/user/outside-agents.md | 12 ++++++++++++ packages/contracts/src/orchestrationV2.ts | 1 + 7 files changed, 37 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 8662344895311..196a39ee944d7 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6504,6 +6504,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : { senderThreadId: command.senderThreadId }), role: "user", posted: true, + postOnly: true, text: command.text, attachments: command.attachments, streaming: false, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index baf23091e322e..db327f52b3484 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -1076,6 +1076,8 @@ 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", diff --git a/apps/server/src/persistence/WorkLedger.test.ts b/apps/server/src/persistence/WorkLedger.test.ts index 0f0875dc2cffb..01db3c3c0d4fa 100644 --- a/apps/server/src/persistence/WorkLedger.test.ts +++ b/apps/server/src/persistence/WorkLedger.test.ts @@ -129,6 +129,17 @@ it.effect( expect((yield* ledger.query({ waitReason: null })).items).toHaveLength(1); expect((yield* ledger.query({ due: true })).items).toEqual([]); const before = yield* ledger.history({ key: "wait" }); + expect( + (yield* ledger + .upsert( + { + key: "wait", + ownerThreadId: ThreadId.make("t3-thread://v2/other-environment/worker"), + }, + "caller-c", + ) + .pipe(Effect.flip)).code, + ).toBe("invalid_request"); yield* ledger.upsert({ key: "wait", nextCheckAt: "invalid" }, "caller-c").pipe(Effect.flip); expect(yield* ledger.history({ key: "wait" })).toEqual(before); }).pipe(Effect.provide(WorkLedger.layer.pipe(Layer.provideMerge(Sqlite.layerMemory)))), diff --git a/apps/server/src/persistence/WorkLedger.ts b/apps/server/src/persistence/WorkLedger.ts index bf6af7141a50f..07bb376f396cf 100644 --- a/apps/server/src/persistence/WorkLedger.ts +++ b/apps/server/src/persistence/WorkLedger.ts @@ -146,6 +146,12 @@ const make = Effect.gen(function* () { const threadIds = new Set(); if (candidate.ownerThreadId !== null) { const ref = parseThreadLinkHref(candidate.ownerThreadId); + if (ref?.environmentId != null) + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: + "ownerThreadId must be an environment-local thread id or v1 link. Keep scoped thread references in links.", + }); if (ref !== null) candidate.ownerThreadId = ref.threadId; threadIds.add(candidate.ownerThreadId); } @@ -153,7 +159,8 @@ const make = Effect.gen(function* () { const href = link.match(/^\[[^\]]*\]\((.*)\)$/)?.[1] ?? link.replace(/^<|>$/g, ""); const ref = parseThreadLinkHref(href); if (ref === null) return link; - threadIds.add(ref.threadId); + if (ref.environmentId === null) threadIds.add(ref.threadId); + else warnings.push(`Scoped thread ${ref.threadId} was not checked locally.`); return formatThreadLink(ref).slice(1, -1); }); for (const threadId of threadIds) { diff --git a/docs/jarvis/decisions/0012-ledger-fields-history-and-portfolio.md b/docs/jarvis/decisions/0012-ledger-fields-history-and-portfolio.md index 11aa7aff26d71..b9cf1722e8d4c 100644 --- a/docs/jarvis/decisions/0012-ledger-fields-history-and-portfolio.md +++ b/docs/jarvis/decisions/0012-ledger-fields-history-and-portfolio.md @@ -8,6 +8,8 @@ The ledger separates a work item's status from its reason for waiting. External Every upsert retains its caller, time, and changed values. A run started with a client request id equal to a work item's key records its terminal outcome on that item without changing the item's status. Finishing a run does not prove that the work is done. +An owner thread belongs to the ledger's environment. References to other environments retain their scope as links. + The portfolio joins open work with each owner thread's latest run, pending requests, and a short result excerpt. A worker with no run for over 24 hours is flagged for attention. That flag does not establish failure or authorize a restart. The web panel reads the same ledger and provides no editing controls. ## Alternatives diff --git a/docs/user/outside-agents.md b/docs/user/outside-agents.md index 0a1029d755a8b..aa49aeb250a3a 100644 --- a/docs/user/outside-agents.md +++ b/docs/user/outside-agents.md @@ -273,3 +273,15 @@ Revoke one there to cut off its access immediately. New sign-ins renew with use and expire after 30 days without use. Clients receive a renewal token, which is replaced each time they use it. Reusing a replaced token cuts off the sign-in. Sign-ins created before rolling renewal keep their original expiry. + +## Track shared work + +The work ledger keeps shared work, its owner, blockers, and next action in one +place. Open **Settings → Work ledger** and choose an environment to read its +items or open an owner's thread. The panel is read-only. Owner threads belong +to the selected environment; references to other environments remain links. + +An assistant can update items with `ledger_upsert`, read their changes with +`ledger_history`, and check worker activity with `portfolio_status`. A wait +reason distinguishes an external dependency, a natural event, and a decision. +A due check or a completed agent run does not mark the work done. diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 2a0041855f58f..e8b283ed99a5b 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1311,6 +1311,7 @@ export const OrchestrationV2IdleCompaction = Schema.Literals(["beforeCacheExpire export type OrchestrationV2IdleCompaction = typeof OrchestrationV2IdleCompaction.Type; export const OrchestrationV2ConversationMessage = Schema.Struct({ + postOnly: Schema.optional(Schema.Boolean), posted: Schema.optional(Schema.Boolean), notification: Schema.optional(OrchestrationV2Notification), ...OrchestrationV2CreationFields, From be0c902e7c875294ae815770ae55eee4557cd559 Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 00:07:33 -0700 Subject: [PATCH 06/10] fix(server): skip malformed run payloads in ledger trigger --- .../src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts b/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts index 90e67a209521a..319b5c623bb6e 100644 --- a/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts +++ b/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts @@ -23,7 +23,8 @@ export default Effect.gen(function* () { 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 json_extract(NEW.payload_json, '$.clientRequestId') IS NOT NULL + 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) From 31ac0d5e6ed19d2840f9442be0d994c6239ae201 Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 00:37:51 -0700 Subject: [PATCH 07/10] fix(ledger): preserve legacy owner link destinations during migration --- .../ForkMigrations/007_WorkLedgerPortfolio.ts | 22 +++++++++++++++++++ .../server/src/persistence/WorkLedger.test.ts | 18 ++++++++++++++- 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts b/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts index 319b5c623bb6e..c271f85265b25 100644 --- a/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts +++ b/apps/server/src/persistence/ForkMigrations/007_WorkLedgerPortfolio.ts @@ -1,6 +1,12 @@ +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`; @@ -12,6 +18,22 @@ export default Effect.gen(function* () { 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, diff --git a/apps/server/src/persistence/WorkLedger.test.ts b/apps/server/src/persistence/WorkLedger.test.ts index 01db3c3c0d4fa..fbcefb6dc3b77 100644 --- a/apps/server/src/persistence/WorkLedger.test.ts +++ b/apps/server/src/persistence/WorkLedger.test.ts @@ -213,7 +213,7 @@ it.effect( }).pipe(Effect.provide(WorkLedger.layer.pipe(Layer.provideMerge(Sqlite.layerMemory)))), ); -it.effect("backfills only recognized blocker prefixes during migration", () => +it.effect("backfills recognized blockers and preserves legacy thread link destinations", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; yield* BaseLedgerMigration; @@ -228,7 +228,23 @@ it.effect("backfills only recognized blocker prefixes during migration", () => yield* sql`INSERT INTO work_ledger (work_key, title, owner_kind, owner_name, status, blocker, links_json, created_at, updated_at, updated_by) VALUES (${key!}, 'Title', 't3', 'worker', 'blocked', ${blocker!}, '[]', '1970-01-01', '1970-01-01', 'caller')`; } + yield* sql`UPDATE work_ledger SET owner_thread_id = 't3-thread://v1/worker-local' WHERE work_key = 'a'`; + yield* sql`UPDATE work_ledger SET owner_thread_id = 't3-thread://v2/remote/worker-remote', links_json = '["https://example.com"]' WHERE work_key = 'b'`; + yield* sql`UPDATE work_ledger SET owner_thread_id = 'https://app.t3.codes/threads/remote/worker%3Ahosted' WHERE work_key = 'c'`; + yield* sql`UPDATE work_ledger SET owner_thread_id = 'plain-thread' WHERE work_key = 'd'`; yield* LedgerMigration; + expect( + yield* sql`SELECT owner_thread_id, links_json FROM work_ledger ORDER BY work_key`, + ).toEqual([ + { owner_thread_id: "worker-local", links_json: "[]" }, + { + owner_thread_id: null, + links_json: '["https://example.com","t3-thread://v2/remote/worker-remote"]', + }, + { owner_thread_id: null, links_json: '["t3-thread://v2/remote/worker:hosted"]' }, + { owner_thread_id: "plain-thread", links_json: "[]" }, + { owner_thread_id: null, links_json: "[]" }, + ]); expect(yield* sql`SELECT wait_reason FROM work_ledger ORDER BY work_key`).toEqual([ { wait_reason: "external" }, { wait_reason: "natural" }, From ce3a72baa9371a1119f57f924a876061d463db50 Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 00:49:10 -0700 Subject: [PATCH 08/10] fix(mcp): persist post-only marker on posted turn items --- apps/server/src/orchestration-v2/Orchestrator.ts | 1 + apps/server/src/orchestration-v2/runtimeLayer.test.ts | 4 +++- packages/contracts/src/orchestrationV2.ts | 2 ++ 3 files changed, 6 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 196a39ee944d7..955e72cbfdec5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6538,6 +6538,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ? {} : { senderThreadId: command.senderThreadId }), inputIntent: "post", + postOnly: true, messageId: command.messageId, text: command.text, attachments: command.attachments, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index db327f52b3484..46056edd9cd31 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -1083,7 +1083,9 @@ it.layer(layerTest)("OrchestrationV2LayerLive", (it) => { creationSource: "mcp", senderThreadId: "sender", }); - 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" }, + ]); }), ); diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index e8b283ed99a5b..80722e016d4ee 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1638,6 +1638,7 @@ export const OrchestrationV2TurnItem = Schema.Union([ scheduledTaskId: Schema.optional(ScheduledTaskId), senderThreadId: Schema.optional(ThreadId), idleCompaction: Schema.optional(OrchestrationV2IdleCompaction), + postOnly: OrchestrationV2ConversationMessage.fields.postOnly, workflowScript: OrchestrationV2ConversationMessage.fields.workflowScript, freshContext: OrchestrationV2Run.fields.freshContext, inputIntent: OrchestrationV2UserMessageInputIntent, @@ -2487,6 +2488,7 @@ export const OrchestrationV2TurnItemJson = Schema.Union([ scheduledTaskId: Schema.optional(ScheduledTaskId), senderThreadId: Schema.optional(ThreadId), idleCompaction: Schema.optional(OrchestrationV2IdleCompaction), + postOnly: OrchestrationV2ConversationMessage.fields.postOnly, workflowScript: OrchestrationV2ConversationMessage.fields.workflowScript, freshContext: OrchestrationV2Run.fields.freshContext, inputIntent: OrchestrationV2UserMessageInputIntent, From bad9a4ead93be638478dd9ce13de0ab41dd5519b Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 01:00:43 -0700 Subject: [PATCH 09/10] fix(ledger): normalize bracketed links and consolidate guidance --- .../server/src/persistence/WorkLedger.test.ts | 14 ++++++++-- apps/server/src/persistence/WorkLedger.ts | 2 +- docs/user/outside-agents.md | 28 +++++++++---------- 3 files changed, 25 insertions(+), 19 deletions(-) diff --git a/apps/server/src/persistence/WorkLedger.test.ts b/apps/server/src/persistence/WorkLedger.test.ts index fbcefb6dc3b77..78c55e37ea5a8 100644 --- a/apps/server/src/persistence/WorkLedger.test.ts +++ b/apps/server/src/persistence/WorkLedger.test.ts @@ -96,15 +96,23 @@ it.effect( ownerKind: "t3", ownerName: "worker", ownerThreadId: ThreadId.make("t3-thread://v1/missing"), - links: ["https://app.t3.codes/threads/env-1/also-missing"], + links: [ + "https://app.t3.codes/threads/env-1/also-missing", + "[remote]()", + "[local]()", + ], waitReason: "external", nextCheckAt: "1970-01-01T01:00:00+01:00", }, "caller-a", ); expect(first.ownerThreadId).toBe("missing"); - expect(first.links).toEqual(["t3-thread://v2/env-1/also-missing"]); - expect(first.warnings).toHaveLength(2); + expect(first.links).toEqual([ + "t3-thread://v2/env-1/also-missing", + "t3-thread://v2/remote/thread", + "t3-thread://v1/angle-missing", + ]); + expect(first.warnings).toHaveLength(4); expect( (yield* ledger.query({ due: true, waitReason: "external" })).items.map((item) => item.key), ).toEqual(["wait"]); diff --git a/apps/server/src/persistence/WorkLedger.ts b/apps/server/src/persistence/WorkLedger.ts index 07bb376f396cf..d164b22020784 100644 --- a/apps/server/src/persistence/WorkLedger.ts +++ b/apps/server/src/persistence/WorkLedger.ts @@ -156,7 +156,7 @@ const make = Effect.gen(function* () { threadIds.add(candidate.ownerThreadId); } candidate.links = candidate.links.map((link) => { - const href = link.match(/^\[[^\]]*\]\((.*)\)$/)?.[1] ?? link.replace(/^<|>$/g, ""); + const href = (link.match(/^\[[^\]]*\]\((.*)\)$/)?.[1] ?? link).replace(/^<|>$/g, ""); const ref = parseThreadLinkHref(href); if (ref === null) return link; if (ref.environmentId === null) threadIds.add(ref.threadId); diff --git a/docs/user/outside-agents.md b/docs/user/outside-agents.md index aa49aeb250a3a..7e21d63f4f56c 100644 --- a/docs/user/outside-agents.md +++ b/docs/user/outside-agents.md @@ -167,9 +167,19 @@ ledger. Writing requires operate access and also works for agents inside T3. Use `status` to track `open`, `blocked`, `waiting_on_human`, `done`, or `cancelled` work. Record a blocker, next action, approval context, links, and an optional owning thread. An `approvalNote` records context and grants no approval. -`ledger_query` filters by `status`, `ownerKind`, `ownerName`, or `updatedSince`, which is inclusive, so skip rows you have already seen. -It returns up to 25 rows by default, ordered by key. Pass `nextCursor` as -`cursor` for the next page, or set `limit` up to 100. +Set `waitReason` to `external`, `natural`, or `decision` to distinguish a dependency, +a real-world event, and a question for Chris. Use `nextCheckAt` for the next due check. + +`ledger_query` filters by `status`, `ownerKind`, `ownerName`, `waitReason`, `due`, +or `updatedSince` (inclusive). It returns up to 25 rows by default, ordered by key. +Pass `nextCursor` as `cursor` for the next page, or set `limit` up to 100. +Read an item's changes with `ledger_history`, or check open work and its owners' +latest activity with `portfolio_status`. A due check or a completed agent run does +not mark the work done. + +Open **Settings → Work ledger** and choose an environment to read its items or +open an owner's thread. The panel is read-only. Owner threads belong to the +selected environment; references to other environments remain links. Use `bridge_status` to check this sign-in's remaining lifetime, subscription state, tunnel reachability, service start time, last successful webhook delivery, @@ -273,15 +283,3 @@ Revoke one there to cut off its access immediately. New sign-ins renew with use and expire after 30 days without use. Clients receive a renewal token, which is replaced each time they use it. Reusing a replaced token cuts off the sign-in. Sign-ins created before rolling renewal keep their original expiry. - -## Track shared work - -The work ledger keeps shared work, its owner, blockers, and next action in one -place. Open **Settings → Work ledger** and choose an environment to read its -items or open an owner's thread. The panel is read-only. Owner threads belong -to the selected environment; references to other environments remain links. - -An assistant can update items with `ledger_upsert`, read their changes with -`ledger_history`, and check worker activity with `portfolio_status`. A wait -reason distinguishes an external dependency, a natural event, and a decision. -A due check or a completed agent run does not mark the work done. From 821b8ab6b25770e60b618e3d97974a5c7a50e928 Mon Sep 17 00:00:00 2001 From: Chris OHearn Date: Thu, 12 Nov 2026 02:29:48 -0700 Subject: [PATCH 10/10] fix(server): keep ledger history and incremental results accurate --- .../ForkMigrations/010_WorkLedgerPortfolio.ts | 3 ++- apps/server/src/persistence/WorkLedger.test.ts | 12 ++++++++++++ apps/server/src/persistence/WorkLedger.ts | 2 +- 3 files changed, 15 insertions(+), 2 deletions(-) diff --git a/apps/server/src/persistence/ForkMigrations/010_WorkLedgerPortfolio.ts b/apps/server/src/persistence/ForkMigrations/010_WorkLedgerPortfolio.ts index c271f85265b25..38740d18aac10 100644 --- a/apps/server/src/persistence/ForkMigrations/010_WorkLedgerPortfolio.ts +++ b/apps/server/src/persistence/ForkMigrations/010_WorkLedgerPortfolio.ts @@ -49,7 +49,8 @@ export default Effect.gen(function* () { 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) + 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`); diff --git a/apps/server/src/persistence/WorkLedger.test.ts b/apps/server/src/persistence/WorkLedger.test.ts index df04035fe8212..bb1e7ab4bae90 100644 --- a/apps/server/src/persistence/WorkLedger.test.ts +++ b/apps/server/src/persistence/WorkLedger.test.ts @@ -131,6 +131,8 @@ it.effect( changedBy: "caller-a", changes: { title: { before: null, after: "Wait" } }, }); + expect(page.items[0]!.changes).not.toHaveProperty("lastRunId"); + expect(page.items[0]!.changes).not.toHaveProperty("approvalNote"); const rest = yield* ledger.history({ key: "wait", cursor: page.nextCursor! }); expect(rest.items[0]).toMatchObject({ changedBy: "caller-b", @@ -188,6 +190,11 @@ it.effect( (yield* ledger.query({})).items.find((item) => item.key === "run-key")!.lastRunId, ).toBeNull(); yield* sql`UPDATE orchestration_v2_projection_runs SET status = 'completed', completed_at = '1970-01-01T00:00:01.000Z' WHERE run_id = 'run-1'`; + expect( + (yield* ledger.query({ updatedSince: "1970-01-01T00:00:01.000Z" })).items.map( + (item) => item.key, + ), + ).toEqual(["run-key"]); yield* sql`INSERT INTO orchestration_v2_projection_messages (message_id, thread_id, run_id, role, streaming, created_at, updated_at, payload_json) VALUES ('result-1', 'worker-thread', 'run-1', 'assistant', 0, '1970-01-01T00:00:01.000Z', '1970-01-01T00:00:01.000Z', '{"text":"Completed the batch"}')`; @@ -215,6 +222,11 @@ it.effect( "caller", ); const page = yield* ledger.portfolio({ limit: 1 }); + yield* ledger.upsert({ key: "run-key" }, "caller"); + yield* sql`UPDATE orchestration_v2_projection_runs SET completed_at = '1970-01-01T00:00:02.000Z' WHERE run_id = 'run-1'`; + expect( + (yield* ledger.query({})).items.find((item) => item.key === "run-key")!.updatedAt, + ).toBe("1970-01-02T01:00:00.000Z"); expect(page.nextCursor).toBe("run-key"); expect((yield* ledger.portfolio({ cursor: page.nextCursor! })).items[0]).toMatchObject({ item: { key: "z" }, diff --git a/apps/server/src/persistence/WorkLedger.ts b/apps/server/src/persistence/WorkLedger.ts index 7c1ce4b61da87..b172ba69e159d 100644 --- a/apps/server/src/persistence/WorkLedger.ts +++ b/apps/server/src/persistence/WorkLedger.ts @@ -216,7 +216,7 @@ const make = Effect.gen(function* () { .filter( ([field, value]) => !["createdAt", "updatedAt", "updatedBy"].includes(field) && - JSON.stringify(current?.[field as keyof WorkLedgerItem]) !== + JSON.stringify(current?.[field as keyof WorkLedgerItem] ?? null) !== JSON.stringify(value), ) .map(([field, value]) => [