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/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 04ab43e2cadde..d85232a1f5b93 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/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, diff --git a/apps/server/src/mcp/ThreadLaunchMcpService.ts b/apps/server/src/mcp/ThreadLaunchMcpService.ts index 7b7bd29101f74..146d4879409f8 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 }), ...McpInvocationContext.messageSenderFor(scope), 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/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/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 6446e81dae09a..5a5d22986b449 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -2155,6 +2155,8 @@ export const CLAUDE_READ_ONLY_T3_MCP_ALLOWED_TOOLS: ReadonlyArray = [ "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", diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index de8c382f50cc9..f4e5c97aada3a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/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, diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index ff87660ab7215..05f68648e333f 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/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, diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index d0438eb323be6..2da6e2a877ff5 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/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, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index a12165c24872a..3ddcc51601de5 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/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" }, + ]); }), ); diff --git a/apps/server/src/persistence/ForkMigrations.ts b/apps/server/src/persistence/ForkMigrations.ts index 0a927b2df1c7f..7a8eaa67bd145 100644 --- a/apps/server/src/persistence/ForkMigrations.ts +++ b/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"; diff --git a/apps/server/src/persistence/ForkMigrations/010_WorkLedgerPortfolio.ts b/apps/server/src/persistence/ForkMigrations/010_WorkLedgerPortfolio.ts new file mode 100644 index 0000000000000..38740d18aac10 --- /dev/null +++ b/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`); + } +}); diff --git a/apps/server/src/persistence/WorkLedger.test.ts b/apps/server/src/persistence/WorkLedger.test.ts index c3507c083361e..bb1e7ab4bae90 100644 --- a/apps/server/src/persistence/WorkLedger.test.ts +++ b/apps/server/src/persistence/WorkLedger.test.ts @@ -3,6 +3,10 @@ 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/010_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"; @@ -28,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, @@ -78,5 +86,198 @@ 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", + "[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", + "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"]); + 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" } }, + }); + 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", + 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" }); + 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)))), +); + +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'`; + 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"}')`; + 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 }); + 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" }, + 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 recognized blockers and preserves legacy thread link destinations", () => + 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"], + ["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')`; + } + 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" }, + { 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 5fe8cbce56a80..b172ba69e159d 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"; @@ -23,7 +31,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; @@ -37,6 +51,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); @@ -47,13 +73,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({ forThread: Effect.fn("WorkLedger.forThread")((threadId) => sql`${select} WHERE owner_thread_id = ${threadId} AND status NOT IN ('done', 'cancelled') ORDER BY work_key LIMIT 26`.pipe( @@ -72,6 +126,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, @@ -83,6 +142,42 @@ 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?.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); + } + candidate.links = candidate.links.map((link) => { + 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); + else warnings.push(`Scoped thread ${ref.threadId} was not checked locally.`); + 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( () => @@ -93,40 +188,109 @@ 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] ?? null) !== + JSON.stringify(value), + ) + .map(([field, value]) => [ + field, + { before: current?.[field as keyof WorkLedgerItem] ?? null, after: value }, + ]), ); - return item; + 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 }; }), ) .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* decodeHistory(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/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index 655e4f024eb2f..9de8bd4e7fa4a 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5597, "codexAdditionalContextBytes": 7327, - "mcpToolListJsonBytes": 54842 + "mcpToolListJsonBytes": 55759 } 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/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..b9cf1722e8d4c --- /dev/null +++ b/docs/jarvis/decisions/0012-ledger-fields-history-and-portfolio.md @@ -0,0 +1,21 @@ +# 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. + +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 + +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. diff --git a/docs/user/outside-agents.md b/docs/user/outside-agents.md index e923eb34a9802..f531513a0026b 100644 --- a/docs/user/outside-agents.md +++ b/docs/user/outside-agents.md @@ -190,9 +190,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, 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/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index eac913d5a1ed1..de552ddeaa0be 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, @@ -3229,6 +3230,7 @@ export const OrchestrationV2Command = Schema.Union([ }), Schema.Struct({ type: Schema.Literal("message.dispatch"), + clientRequestId: Schema.optional(TrimmedNonEmptyString), postOnly: Schema.optional(Schema.Boolean), notification: Schema.optional(OrchestrationV2Notification), /** Sends this pending guest message; the server uses the guest's text, not `text`. */ @@ -3406,6 +3408,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/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, 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;