From 17f18a35a1016decd2b4ab2a10e5f26a777cb549 Mon Sep 17 00:00:00 2001 From: Ramnique Singh <30795890+ramnique@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:56:04 +0530 Subject: [PATCH] feat(migration): port legacy runs into the turn/session runtime at startup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One-time startup migration that scans ~/.rowboat/runs/*.jsonl and converts each legacy run into the new event-sourced runtime, so existing chats and agent history are visible and continuable after the runtime rewrite. Mapping (a run is one whole conversation; turn boundaries are user messages): - copilot_chat run -> 1 session (sessionId = original runId) + one turn per user message (turn ids -tNNN). - every other run -> a single standalone turn whose id IS the original runId, so live-note (lastRunId) and background-task (runs.log) history views resolve it via sessions:getTurn with no renderer change. - code_session runs are skipped (Code mode still uses the old runtime). - convert.ts: pure convertRun(); synthesizes a reduceTurn/reduceSession-legal event log (exact request refs, permission replay, denials -> runtime isError results, reasoning parts preserved) and validates via the reducers before returning. - migrate.ts: defensive IO runner. Successful runs are moved to runs-archive/ (that move is the idempotency guard); failures are left in place (still served by the runs:fetch fallback) and retried next launch. Per-run try/catch never blocks boot. Writes config/runs-migration.json. - main.ts: runs before sessions.initialize() and logs an [runs-migration] summary (N turns across M sessions, skipped/failed counts). - Tests: 13 cases over 4 real (redacted) run fixtures — conversion, session chaining, deny->error, reasoning preservation, transcript fidelity, archive-on-success, skip, quarantine, idempotency, and read-back through the real FSTurnRepo/FSSessionRepo. Dry-run on real data: 19 runs -> 3 sessions + 21 turns, 0 failures. Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/x/apps/main/src/main.ts | 20 + .../runs/__fixtures__/background-task.jsonl | 15 + .../runs/__fixtures__/copilot-brew-deny.jsonl | 7 + .../__fixtures__/copilot-multi-turn.jsonl | 16 + .../__fixtures__/knowledge-single-turn.jsonl | 3 + .../core/src/migrations/runs/convert.test.ts | 140 +++++ .../core/src/migrations/runs/convert.ts | 539 ++++++++++++++++++ .../core/src/migrations/runs/migrate.test.ts | 191 +++++++ .../core/src/migrations/runs/migrate.ts | 165 ++++++ 9 files changed, 1096 insertions(+) create mode 100644 apps/x/packages/core/src/migrations/runs/__fixtures__/background-task.jsonl create mode 100644 apps/x/packages/core/src/migrations/runs/__fixtures__/copilot-brew-deny.jsonl create mode 100644 apps/x/packages/core/src/migrations/runs/__fixtures__/copilot-multi-turn.jsonl create mode 100644 apps/x/packages/core/src/migrations/runs/__fixtures__/knowledge-single-turn.jsonl create mode 100644 apps/x/packages/core/src/migrations/runs/convert.test.ts create mode 100644 apps/x/packages/core/src/migrations/runs/convert.ts create mode 100644 apps/x/packages/core/src/migrations/runs/migrate.test.ts create mode 100644 apps/x/packages/core/src/migrations/runs/migrate.ts diff --git a/apps/x/apps/main/src/main.ts b/apps/x/apps/main/src/main.ts index fb033413..9f17fdc8 100644 --- a/apps/x/apps/main/src/main.ts +++ b/apps/x/apps/main/src/main.ts @@ -36,6 +36,7 @@ import { backgroundTaskEventConsumer } from "@x/core/dist/background-tasks/event import { init as initLocalSites, shutdown as shutdownLocalSites } from "@x/core/dist/local-sites/server.js"; import { shutdown as shutdownAnalytics } from "@x/core/dist/analytics/posthog.js"; import { identifyIfSignedIn } from "@x/core/dist/analytics/identify.js"; +import { migrateRuns } from "@x/core/dist/migrations/runs/migrate.js"; import { initConfigs } from "@x/core/dist/config/initConfigs.js"; import { getAgentSlackCliStatus } from "@x/core/dist/slack/agent-slack-exec.js"; @@ -389,6 +390,25 @@ app.whenReady().then(async () => { // start runs watcher startRunsWatcher(); + // One-time: port legacy runs/*.jsonl into the new turn/session runtime. + // Must run BEFORE the session index is built so migrated sessions are picked + // up by the startup scan. Fully defensive — never blocks boot. + try { + const migration = migrateRuns(); + if (migration.scanned > 0) { + console.log( + `[runs-migration] migrated ${migration.migratedTurns} turn(s) across ` + + `${migration.migratedSessions} session(s) from ${migration.scanned} run(s) ` + + `(${migration.skipped} skipped, ${migration.failed.length} failed)`, + ); + for (const failure of migration.failed) { + console.warn(`[runs-migration] left in place (failed): ${failure.file} — ${failure.error}`); + } + } + } catch (error) { + console.error('[runs-migration] pass failed:', error); + } + // New runtime: build the in-memory session index (startup scan) before the // renderer can list sessions, then forward the session bus to windows. await container.resolve('sessions').initialize(); diff --git a/apps/x/packages/core/src/migrations/runs/__fixtures__/background-task.jsonl b/apps/x/packages/core/src/migrations/runs/__fixtures__/background-task.jsonl new file mode 100644 index 00000000..781f41c5 --- /dev/null +++ b/apps/x/packages/core/src/migrations/runs/__fixtures__/background-task.jsonl @@ -0,0 +1,15 @@ +{"type":"start","runId":"2026-07-02T15-55-29Z-0017565-000","agentName":"background-task-agent","model":"google/gemini-3.1-flash-lite","provider":"rowboat","permissionMode":"manual","useCase":"background_task_agent","subUseCase":"manual","subflow":[],"ts":"2026-07-02T15:55:29.568Z"} +{"runId":"2026-07-02T15-55-29Z-0017565-000","type":"message","messageId":"2026-07-02T15-55-29Z-0017565-001","message":{"role":"user","content":"Run the background task at `bg-tasks/mohali-weather-brief/`.\n\n**Time:** Thursday, July 2, 2026 at 9:25:29 PM GMT+5:30 (Asia/Calcutta)\n\n**Instructions:**\nGet and show the latest weather in Mohali in degree celsius\n\nYour task folder is `bg-tasks/mohali-weather-brief/`. The user-visible artifact is `bg-tasks/mohali-weather-brief/index.md` — read it with `file-readText` and update it with `file-editText` per the OUTPUT / ACTION mode rule. Do not touch `bg-tasks/mohali-weather-brief/task.yaml` (the runtime owns it).\n\n**Trigger:** Manual run (user-triggered — either the Run button in the Background Task detail view or the `run-background-task-agent` tool).","userMessageContext":{"currentDateTime":"Thursday, July 2, 2026 at 9:25 PM GMT+5:30"}},"subflow":[]} +{"runId":"2026-07-02T15-55-29Z-0017565-000","messageId":"2026-07-02T15-55-31Z-0017565-000","type":"message","message":{"role":"assistant","content":[{"type":"tool-call","toolCallId":"oFDM4XYN","toolName":"file-readText","arguments":{"query":"redacted"},"providerOptions":{"openrouter":{"reasoning_details":[{"type":"reasoning.encrypted","data":"EjQKMgEMOdbHK4P4fV56Kb79TR4GLeN0KXeYhT+YPCqiTE4fmExcz7koIh6pfWrr8i4wlW2X","id":"oFDM4XYN","format":"google-gemini-v1","index":0}]}}},{"type":"reasoning","text":"[REDACTED]"}],"providerOptions":{"openrouter":{"usage":{"promptTokens":12311,"promptTokensDetails":{"cachedTokens":0},"completionTokens":30,"completionTokensDetails":{"reasoningTokens":0},"cost":0,"totalTokens":12341,"costDetails":{"upstreamInferenceCost":0.00312275}},"provider":"Google AI Studio","reasoning_details":[{"type":"reasoning.encrypted","data":"EjQKMgEMOdbHK4P4fV56Kb79TR4GLeN0KXeYhT+YPCqiTE4fmExcz7koIh6pfWrr8i4wlW2X","id":"oFDM4XYN","format":"google-gemini-v1","index":0}]}}},"subflow":[]} +{"runId":"2026-07-02T15-55-29Z-0017565-000","type":"tool-invocation","toolCallId":"oFDM4XYN","toolName":"file-readText","input":"{\"query\": \"redacted\"}","subflow":[]} +{"runId":"2026-07-02T15-55-29Z-0017565-000","type":"tool-result","toolCallId":"oFDM4XYN","toolName":"file-readText","result":{"ok":true,"note":"redacted for fixture"},"subflow":[]} +{"runId":"2026-07-02T15-55-29Z-0017565-000","messageId":"2026-07-02T15-55-31Z-0017565-001","type":"message","message":{"role":"tool","content":"redacted for fixture","toolCallId":"oFDM4XYN","toolName":"file-readText"},"subflow":[]} +{"runId":"2026-07-02T15-55-29Z-0017565-000","messageId":"2026-07-02T15-55-33Z-0017565-000","type":"message","message":{"role":"assistant","content":[{"type":"tool-call","toolCallId":"OAs24vk5","toolName":"web-search","arguments":{"query":"redacted"},"providerOptions":{"openrouter":{"reasoning_details":[{"type":"reasoning.encrypted","data":"EjQKMgEMOdbHs6o9nrkXG1745HqUQkXR2/g4bvozLYGNP63eCEFEIDPzSKvW+CBAyOuSl46C","id":"OAs24vk5","format":"google-gemini-v1","index":0}]}}},{"type":"reasoning","text":"[REDACTED]"}],"providerOptions":{"openrouter":{"usage":{"promptTokens":12609,"promptTokensDetails":{"cachedTokens":8079},"completionTokens":21,"completionTokensDetails":{"reasoningTokens":0},"cost":0,"totalTokens":12630,"costDetails":{"upstreamInferenceCost":0.001365975}},"provider":"Google AI Studio","reasoning_details":[{"type":"reasoning.encrypted","data":"EjQKMgEMOdbHs6o9nrkXG1745HqUQkXR2/g4bvozLYGNP63eCEFEIDPzSKvW+CBAyOuSl46C","id":"OAs24vk5","format":"google-gemini-v1","index":0}]}}},"subflow":[]} +{"runId":"2026-07-02T15-55-29Z-0017565-000","type":"tool-invocation","toolCallId":"OAs24vk5","toolName":"web-search","input":"{\"query\": \"redacted\"}","subflow":[]} +{"runId":"2026-07-02T15-55-29Z-0017565-000","type":"tool-result","toolCallId":"OAs24vk5","toolName":"web-search","result":{"ok":true,"note":"redacted for fixture"},"subflow":[]} +{"runId":"2026-07-02T15-55-29Z-0017565-000","messageId":"2026-07-02T15-55-34Z-0017565-000","type":"message","message":{"role":"tool","content":"redacted for fixture","toolCallId":"OAs24vk5","toolName":"web-search"},"subflow":[]} +{"runId":"2026-07-02T15-55-29Z-0017565-000","messageId":"2026-07-02T15-55-36Z-0017565-000","type":"message","message":{"role":"assistant","content":[{"type":"tool-call","toolCallId":"xcoaRes7","toolName":"file-editText","arguments":{"query":"redacted"},"providerOptions":{"openrouter":{"reasoning_details":[{"type":"reasoning.encrypted","data":"EjQKMgEMOdbH+d29EP8/ZWH5EL3vSL+Nb9k5tsCVzMyINtH7vfhAvqeB7hBGY1L60w2BRWn4","id":"xcoaRes7","format":"google-gemini-v1","index":0}]}}},{"type":"reasoning","text":"[REDACTED]"}],"providerOptions":{"openrouter":{"usage":{"promptTokens":17340,"promptTokensDetails":{"cachedTokens":12157},"completionTokens":75,"completionTokensDetails":{"reasoningTokens":0},"cost":0,"totalTokens":17415,"costDetails":{"upstreamInferenceCost":0.001712175}},"provider":"Google AI Studio","reasoning_details":[{"type":"reasoning.encrypted","data":"EjQKMgEMOdbH+d29EP8/ZWH5EL3vSL+Nb9k5tsCVzMyINtH7vfhAvqeB7hBGY1L60w2BRWn4","id":"xcoaRes7","format":"google-gemini-v1","index":0}]}}},"subflow":[]} +{"runId":"2026-07-02T15-55-29Z-0017565-000","type":"tool-invocation","toolCallId":"xcoaRes7","toolName":"file-editText","input":"{\"query\": \"redacted\"}","subflow":[]} +{"runId":"2026-07-02T15-55-29Z-0017565-000","type":"tool-result","toolCallId":"xcoaRes7","toolName":"file-editText","result":{"ok":true,"note":"redacted for fixture"},"subflow":[]} +{"runId":"2026-07-02T15-55-29Z-0017565-000","messageId":"2026-07-02T15-55-36Z-0017565-001","type":"message","message":{"role":"tool","content":"redacted for fixture","toolCallId":"xcoaRes7","toolName":"file-editText"},"subflow":[]} +{"runId":"2026-07-02T15-55-29Z-0017565-000","messageId":"2026-07-02T15-55-37Z-0017565-000","type":"message","message":{"role":"assistant","content":[{"type":"text","text":"Updated — Mohali: 30.5°C · Clear Sky (as of 9:00 PM IST)."}],"providerOptions":{"openrouter":{"usage":{"promptTokens":17487,"promptTokensDetails":{"cachedTokens":12151},"completionTokens":26,"completionTokensDetails":{"reasoningTokens":0},"cost":0,"totalTokens":17513,"costDetails":{"upstreamInferenceCost":0.001676775}},"provider":"Google AI Studio","reasoning_details":[{"type":"reasoning.text","signature":"EjQKMgEMOdbHosBKKoAUZMNdqBPqUrsYCf2gYO8SL1TPjujCq2Zfl+Iia3fy5vIqcd8yY7TI","format":"google-gemini-v1","index":0}]}}},"subflow":[]} diff --git a/apps/x/packages/core/src/migrations/runs/__fixtures__/copilot-brew-deny.jsonl b/apps/x/packages/core/src/migrations/runs/__fixtures__/copilot-brew-deny.jsonl new file mode 100644 index 00000000..fe9ccad3 --- /dev/null +++ b/apps/x/packages/core/src/migrations/runs/__fixtures__/copilot-brew-deny.jsonl @@ -0,0 +1,7 @@ +{"type":"start","runId":"2026-07-02T15-29-15Z-0011690-000","agentName":"copilot","model":"anthropic/claude-opus-4.7","provider":"rowboat","permissionMode":"manual","useCase":"copilot_chat","subflow":[],"ts":"2026-07-02T15:29:15.130Z"} +{"runId":"2026-07-02T15-29-15Z-0011690-000","type":"message","messageId":"2026-07-02T15-29-15Z-0011690-001","message":{"role":"user","content":"run the following command:\nbrew install foobar","userMessageContext":{"currentDateTime":"Thursday, July 2, 2026 at 8:59 PM GMT+5:30","middlePane":{"kind":"empty"}}},"subflow":[]} +{"runId":"2026-07-02T15-29-15Z-0011690-000","messageId":"2026-07-02T15-29-17Z-0011690-002","type":"message","message":{"role":"assistant","content":[{"type":"tool-call","toolCallId":"toolu_bdrk_01Vq7Vjphw18SzBskW1F38QP","toolName":"executeCommand","arguments":{"query":"redacted"},"providerOptions":{"openrouter":{"reasoning_details":[]}}},{"type":"text","text":"I'll run that for you."}],"providerOptions":{"openrouter":{"usage":{"promptTokens":24249,"promptTokensDetails":{"cachedTokens":0},"completionTokens":79,"completionTokensDetails":{"reasoningTokens":0},"cost":0,"totalTokens":24328,"costDetails":{"upstreamInferenceCost":0.12322}},"provider":"Amazon Bedrock"}}},"subflow":[]} +{"runId":"2026-07-02T15-29-15Z-0011690-000","type":"tool-permission-request","toolCall":{"type":"tool-call","toolCallId":"toolu_bdrk_01Vq7Vjphw18SzBskW1F38QP","toolName":"executeCommand","arguments":{"query":"redacted"},"providerOptions":{"openrouter":{"reasoning_details":[]}}},"permission":{"kind":"command","commandNames":["brew"]},"subflow":[]} +{"subflow":[],"toolCallId":"toolu_bdrk_01Vq7Vjphw18SzBskW1F38QP","response":"deny","runId":"2026-07-02T15-29-15Z-0011690-000","type":"tool-permission-response"} +{"runId":"2026-07-02T15-29-15Z-0011690-000","messageId":"2026-07-02T15-29-19Z-0011690-002","type":"message","message":{"role":"tool","content":"redacted for fixture","toolCallId":"toolu_bdrk_01Vq7Vjphw18SzBskW1F38QP","toolName":"executeCommand"},"subflow":[]} +{"runId":"2026-07-02T15-29-15Z-0011690-000","messageId":"2026-07-02T15-29-24Z-0011690-000","type":"message","message":{"role":"assistant","content":[{"type":"text","text":"The command was denied by the approval prompt, so `brew install foobar` didn't run. If you'd like to try again, approve the prompt when it appears — or let me know if you want me to try a different package name (Homebrew doesn't have an official `foobar` formula, so it would likely fail with \"No available formula\" anyway)."}],"providerOptions":{"openrouter":{"usage":{"promptTokens":24357,"promptTokensDetails":{"cachedTokens":0},"completionTokens":104,"completionTokensDetails":{"reasoningTokens":0},"cost":0,"totalTokens":24461,"costDetails":{"upstreamInferenceCost":0.124385}},"provider":"Amazon Bedrock"}}},"subflow":[]} diff --git a/apps/x/packages/core/src/migrations/runs/__fixtures__/copilot-multi-turn.jsonl b/apps/x/packages/core/src/migrations/runs/__fixtures__/copilot-multi-turn.jsonl new file mode 100644 index 00000000..e2b79ede --- /dev/null +++ b/apps/x/packages/core/src/migrations/runs/__fixtures__/copilot-multi-turn.jsonl @@ -0,0 +1,16 @@ +{"type":"start","runId":"2026-07-02T15-27-08Z-0011690-000","agentName":"copilot","model":"anthropic/claude-opus-4.7","provider":"rowboat","permissionMode":"auto","useCase":"copilot_chat","subflow":[],"ts":"2026-07-02T15:27:08.939Z"} +{"runId":"2026-07-02T15-27-08Z-0011690-000","type":"message","messageId":"2026-07-02T15-27-09Z-0011690-000","message":{"role":"user","content":"yo","userMessageContext":{"currentDateTime":"Thursday, July 2, 2026 at 8:57 PM GMT+5:30","middlePane":{"kind":"empty"}}},"subflow":[]} +{"runId":"2026-07-02T15-27-08Z-0011690-000","messageId":"2026-07-02T15-27-11Z-0011690-002","type":"message","message":{"role":"assistant","content":[{"type":"text","text":"Yo! What's up?"}],"providerOptions":{"openrouter":{"usage":{"promptTokens":24239,"promptTokensDetails":{"cachedTokens":0},"completionTokens":12,"completionTokensDetails":{"reasoningTokens":0},"cost":0,"totalTokens":24251,"costDetails":{"upstreamInferenceCost":0.121495}},"provider":"Amazon Bedrock"}}},"subflow":[]} +{"runId":"2026-07-02T15-27-08Z-0011690-000","type":"message","messageId":"2026-07-02T15-27-25Z-0011690-000","message":{"role":"user","content":"whats the weather in Mohali?","userMessageContext":{"currentDateTime":"Thursday, July 2, 2026 at 8:57 PM GMT+5:30","middlePane":{"kind":"empty"}}},"subflow":[]} +{"runId":"2026-07-02T15-27-08Z-0011690-000","messageId":"2026-07-02T15-27-30Z-0011690-000","type":"message","message":{"role":"assistant","content":[{"type":"tool-call","toolCallId":"toolu_bdrk_013dC9KmbiBbMH7L2a8wJ3HG","toolName":"web-search","arguments":{"query":"redacted"},"providerOptions":{"openrouter":{"reasoning_details":[]}}},{"type":"text","text":"Quick answer coming up — but \"what's the weather\" is exactly the kind of thing that decays fast. Want me to set up a background task that keeps a fresh Mohali forecast in a note (say, refreshed every morning)? Say the word.\n\nFor right now:"}],"providerOptions":{"openrouter":{"usage":{"promptTokens":24325,"promptTokensDetails":{"cachedTokens":0},"completionTokens":176,"completionTokensDetails":{"reasoningTokens":0},"cost":0,"totalTokens":24501,"costDetails":{"upstreamInferenceCost":0.126025}},"provider":"Amazon Bedrock"}}},"subflow":[]} +{"runId":"2026-07-02T15-27-08Z-0011690-000","type":"tool-invocation","toolCallId":"toolu_bdrk_013dC9KmbiBbMH7L2a8wJ3HG","toolName":"web-search","input":"{\"query\": \"redacted\"}","subflow":[]} +{"runId":"2026-07-02T15-27-08Z-0011690-000","type":"tool-result","toolCallId":"toolu_bdrk_013dC9KmbiBbMH7L2a8wJ3HG","toolName":"web-search","result":{"ok":true,"note":"redacted for fixture"},"subflow":[]} +{"runId":"2026-07-02T15-27-08Z-0011690-000","messageId":"2026-07-02T15-27-31Z-0011690-000","type":"message","message":{"role":"tool","content":"redacted for fixture","toolCallId":"toolu_bdrk_013dC9KmbiBbMH7L2a8wJ3HG","toolName":"web-search"},"subflow":[]} +{"runId":"2026-07-02T15-27-08Z-0011690-000","messageId":"2026-07-02T15-27-40Z-0011690-000","type":"message","message":{"role":"assistant","content":[{"type":"text","text":"### Mohali right now (Thu, Jul 2, ~9 PM IST)\n- **~30°C (87°F)**, feels like **~37°C (99°F)** — humid\n- Clear sky, humidity **76%**, light ESE wind ~3 km/h\n- **Today's range:** 27°C → 33°C, 0% rain\n- **Sunset:** 7:29 PM (already down)\n\n### Next few days\n- **Fri Jul 3:** hot & overcast, up to ~34°C\n- **Sat–Wed:** monsoon kicks in — thunderstorms most days, highs ~32–34°C\n\nBasically: sticky tonight, storms rolling in over the weekend."}],"providerOptions":{"openrouter":{"usage":{"promptTokens":31150,"promptTokensDetails":{"cachedTokens":0},"completionTokens":231,"completionTokensDetails":{"reasoningTokens":0},"cost":0,"totalTokens":31381,"costDetails":{"upstreamInferenceCost":0.161525}},"provider":"Amazon Bedrock"}}},"subflow":[]} +{"runId":"2026-07-02T15-27-08Z-0011690-000","type":"message","messageId":"2026-07-02T15-27-45Z-0011690-000","message":{"role":"user","content":"list my recent downloads","userMessageContext":{"currentDateTime":"Thursday, July 2, 2026 at 8:57 PM GMT+5:30","middlePane":{"kind":"empty"}}},"subflow":[]} +{"runId":"2026-07-02T15-27-08Z-0011690-000","messageId":"2026-07-02T15-28-31Z-0011690-000","type":"message","message":{"role":"assistant","content":[{"type":"tool-call","toolCallId":"toolu_bdrk_01HJvA2Yi1eMWmtr5v5KaNe4","toolName":"file-list","arguments":{"query":"redacted"},"providerOptions":{"openrouter":{"reasoning_details":[]}}}],"providerOptions":{"openrouter":{"usage":{"promptTokens":31451,"promptTokensDetails":{"cachedTokens":0},"completionTokens":92,"completionTokensDetails":{"reasoningTokens":0},"cost":0,"totalTokens":31543,"costDetails":{"upstreamInferenceCost":0.159555}},"provider":"Amazon Bedrock"}}},"subflow":[]} +{"runId":"2026-07-02T15-27-08Z-0011690-000","type":"tool-permission-auto-decision","toolCallId":"toolu_bdrk_01HJvA2Yi1eMWmtr5v5KaNe4","toolCall":{"type":"tool-call","toolCallId":"toolu_bdrk_01HJvA2Yi1eMWmtr5v5KaNe4","toolName":"file-list","arguments":{"query":"redacted"},"providerOptions":{"openrouter":{"reasoning_details":[]}}},"permission":{"kind":"file","operation":"list","paths":["/Users/testuser/Downloads"],"pathPrefix":"/Users/testuser/Downloads"},"decision":"allow","reason":"The user explicitly requested to list files in their Downloads directory, and the tool call is scoped to that directory.","subflow":[]} +{"runId":"2026-07-02T15-27-08Z-0011690-000","type":"tool-invocation","toolCallId":"toolu_bdrk_01HJvA2Yi1eMWmtr5v5KaNe4","toolName":"file-list","input":"{\"query\": \"redacted\"}","subflow":[]} +{"runId":"2026-07-02T15-27-08Z-0011690-000","type":"tool-result","toolCallId":"toolu_bdrk_01HJvA2Yi1eMWmtr5v5KaNe4","toolName":"file-list","result":{"ok":true,"note":"redacted for fixture"},"subflow":[]} +{"runId":"2026-07-02T15-27-08Z-0011690-000","messageId":"2026-07-02T15-28-33Z-0011690-000","type":"message","message":{"role":"tool","content":"redacted for fixture","toolCallId":"toolu_bdrk_01HJvA2Yi1eMWmtr5v5KaNe4","toolName":"file-list"},"subflow":[]} +{"runId":"2026-07-02T15-27-08Z-0011690-000","messageId":"2026-07-02T15-28-57Z-0011690-000","type":"message","message":{"role":"assistant","content":[{"type":"text","text":"Your Downloads folder is *massive* — hundreds of files going back to 2024. Here are the **10 most recent** (all from the past few days, June 2026):\n\n### Most recent downloads\n1. `Punjab-main---Punjab-main.pdf` — 36 MB, Jul 1 (~10 PM)\n2. `Punjab-main-Punjab-main-page-9.pdf` — 339 KB, Jul 1\n3. `89227HC00FLG.pdf` — 46 KB, Jun 29\n4. `26070100012297ICIC_ChallanReceipt.pdf` — 422 KB, Jun 27\n5. `AnnexureD.pdf` — 81 KB, Jun 26\n6. `18997H8016V0.pdf` — 46 KB, Jun 24\n7. `53377H801DNU.pdf` — 46 KB, Jun 24\n8. `76687H8019D2.pdf` — 46 KB, Jun 24\n9. `25327H800NXB.pdf` — 46 KB, Jun 24\n10. `49137H8010X2.pdf` — 46 KB, Jun 24\n\n```filepath\n~/Downloads\n```\n\nHeads up — your Downloads folder is a graveyard: multiple `(1)`, `(2)`, `(3)` duplicates, ~30 old Rowboat .dmg builds, ISOs, obsolete installers. Rough estimate: several GB of easy wins. Say the word and I'll draft a cleanup plan (dedupe, archive old builds, sweep installers)."}],"providerOptions":{"openrouter":{"usage":{"promptTokens":146280,"promptTokensDetails":{"cachedTokens":0},"completionTokens":518,"completionTokensDetails":{"reasoningTokens":0},"cost":0,"totalTokens":146798,"costDetails":{"upstreamInferenceCost":0.74435}},"provider":"Amazon Bedrock"}}},"subflow":[]} diff --git a/apps/x/packages/core/src/migrations/runs/__fixtures__/knowledge-single-turn.jsonl b/apps/x/packages/core/src/migrations/runs/__fixtures__/knowledge-single-turn.jsonl new file mode 100644 index 00000000..bf1109b2 --- /dev/null +++ b/apps/x/packages/core/src/migrations/runs/__fixtures__/knowledge-single-turn.jsonl @@ -0,0 +1,3 @@ +{"type":"start","runId":"2026-07-02T15-26-22Z-0011690-000","agentName":"note_creation","model":"google/gemini-3.1-flash-lite","provider":"rowboat","permissionMode":"manual","useCase":"knowledge_sync","subUseCase":"build_graph","subflow":[],"ts":"2026-07-02T15:26:22.103Z"} +{"runId":"2026-07-02T15-26-22Z-0011690-000","type":"message","messageId":"2026-07-02T15-26-22Z-0011690-001","message":{"role":"user","content":"Process the following 1 source files and create/update obsidian notes.\n\n**Instructions:**\n- Use the KNOWLEDGE BASE INDEX below to resolve entities - DO NOT grep/search for existing notes\n- Extract entities and relationships\n\n**KNOWLEDGE BASE INDEX:**\n- [[Acme Corp]]\n- [[Sample Project]]\n\n**Source file: sample-note.md**\nMeeting with Acme Corp about the Sample Project. Follow up next week.\n","userMessageContext":{"currentDateTime":"Thursday, July 2, 2026 at 8:56 PM GMT+5:30"}},"subflow":[]} +{"runId":"2026-07-02T15-26-22Z-0011690-000","messageId":"2026-07-02T15-26-24Z-0011690-000","type":"message","message":{"role":"assistant","content":[{"type":"text","text":"Created/updated 1 note: [[Acme Corp]] with a reference to [[Sample Project]]."}],"providerOptions":{"openrouter":{"usage":{"promptTokens":19366,"promptTokensDetails":{"cachedTokens":0},"completionTokens":84,"completionTokensDetails":{"reasoningTokens":0},"cost":0,"totalTokens":19450,"costDetails":{"upstreamInferenceCost":0.0049675}},"provider":"Google AI Studio","reasoning_details":[{"type":"reasoning.text","signature":"EjQKMgEMOdbHC3J8//5gZWhEIrTstK75K+2cLu//G8KXeokib7MUOAXX6oLYt9RyB+jnT6JE","format":"google-gemini-v1","index":0}]}}},"subflow":[]} diff --git a/apps/x/packages/core/src/migrations/runs/convert.test.ts b/apps/x/packages/core/src/migrations/runs/convert.test.ts new file mode 100644 index 00000000..0f6e889f --- /dev/null +++ b/apps/x/packages/core/src/migrations/runs/convert.test.ts @@ -0,0 +1,140 @@ +import { describe, it, expect } from "vitest"; +import fs from "node:fs"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { z } from "zod"; +import { RunEvent } from "@x/shared/dist/runs.js"; +import { reduceTurn, turnTranscript } from "@x/shared/dist/turns.js"; +import { reduceSession } from "@x/shared/dist/sessions.js"; +import { convertRun } from "./convert.js"; + +const fixturesDir = path.join( + path.dirname(fileURLToPath(import.meta.url)), + "__fixtures__", +); + +type RunEventT = z.infer; + +function loadFixture(name: string): { log: RunEventT[]; runId: string } { + const raw = fs.readFileSync(path.join(fixturesDir, name), "utf-8"); + const log = raw + .split("\n") + .filter((line) => line.trim().length > 0) + .map((line) => RunEvent.parse(JSON.parse(line))); + const start = log[0]; + if (start.type !== "start") throw new Error("fixture missing start event"); + return { log, runId: start.runId }; +} + +describe("convertRun", () => { + it("non-copilot run -> a single standalone turn whose id is the runId", () => { + const { log, runId } = loadFixture("knowledge-single-turn.jsonl"); + const result = convertRun(log, runId); + + expect(result.isSession).toBe(false); + expect(result.session).toBeUndefined(); + expect(result.turns).toHaveLength(1); + expect(result.turns[0].turnId).toBe(runId); + + const state = reduceTurn(result.turns[0].events); + expect(state.terminal?.type).toBe("turn_completed"); + expect(state.definition.sessionId).toBeNull(); + // Standalone turn has inline (empty) context. + expect(Array.isArray(state.definition.context)).toBe(true); + }); + + it("background-task run -> standalone turn, reasoning parts preserved", () => { + const { log, runId } = loadFixture("background-task.jsonl"); + const result = convertRun(log, runId); + + expect(result.isSession).toBe(false); + expect(result.turns[0].turnId).toBe(runId); + + const state = reduceTurn(result.turns[0].events); + // The bg-task run interleaves tool-call + reasoning parts; they survive + // verbatim on the completed model messages. + const hasReasoning = state.modelCalls.some( + (c) => + Array.isArray(c.response?.content) && + c.response.content.some((p) => p.type === "reasoning"), + ); + expect(hasReasoning).toBe(true); + // Every tool call resolved with a terminal result. + expect(state.toolCalls.every((tc) => tc.result)).toBe(true); + }); + + it("copilot run -> session + one turn per user message, chained by previousTurnId", () => { + const { log, runId } = loadFixture("copilot-multi-turn.jsonl"); + const result = convertRun(log, runId); + + expect(result.isSession).toBe(true); + expect(result.session?.sessionId).toBe(runId); + // Fixture has 3 user messages. + expect(result.turns).toHaveLength(3); + + // Session reduces and lists the turns in order. + const sessionState = reduceSession(result.session!.events); + expect(sessionState.turns.map((t) => t.turnId)).toEqual( + result.turns.map((t) => t.turnId), + ); + expect(sessionState.turns.map((t) => t.sessionSeq)).toEqual([1, 2, 3]); + + // First turn has inline context; later turns reference their predecessor. + const states = result.turns.map((t) => reduceTurn(t.events)); + expect(Array.isArray(states[0].definition.context)).toBe(true); + expect(states[1].definition.context).toEqual({ + previousTurnId: result.turns[0].turnId, + }); + expect(states[2].definition.context).toEqual({ + previousTurnId: result.turns[1].turnId, + }); + // Every turn is completed and its session id points back to the session. + for (const s of states) { + expect(s.terminal?.type).toBe("turn_completed"); + expect(s.definition.sessionId).toBe(runId); + } + }); + + it("copilot deny run -> runtime tool result flagged as error", () => { + const { log, runId } = loadFixture("copilot-brew-deny.jsonl"); + const result = convertRun(log, runId); + + expect(result.isSession).toBe(true); + const state = reduceTurn(result.turns[0].events); + + const denied = state.toolCalls.find( + (tc) => tc.result?.source === "runtime", + ); + expect(denied).toBeDefined(); + expect(denied?.result?.result.isError).toBe(true); + // A denied call is resolved but never invoked. + expect(denied?.permission?.resolved?.decision).toBe("deny"); + expect(denied?.invocation).toBeUndefined(); + }); + + it("turnTranscript reproduces the original conversation slice", () => { + const { log, runId } = loadFixture("copilot-multi-turn.jsonl"); + const result = convertRun(log, runId); + + // Reconstruct expected per-turn user/assistant text from the raw log. + const userTexts: string[] = []; + for (const e of log) { + if (e.type === "message" && e.message.role === "user") { + const c = e.message.content; + userTexts.push(typeof c === "string" ? c : JSON.stringify(c)); + } + } + + result.turns.forEach((turn, i) => { + const transcript = turnTranscript(reduceTurn(turn.events)); + // The first message of each turn's transcript is its user input. + expect(transcript[0].role).toBe("user"); + const c = transcript[0].content; + expect(typeof c === "string" ? c : JSON.stringify(c)).toBe( + userTexts[i], + ); + // The last contributed message is an assistant response. + expect(transcript[transcript.length - 1].role).toBe("assistant"); + }); + }); +}); diff --git a/apps/x/packages/core/src/migrations/runs/convert.ts b/apps/x/packages/core/src/migrations/runs/convert.ts new file mode 100644 index 00000000..5e1906c2 --- /dev/null +++ b/apps/x/packages/core/src/migrations/runs/convert.ts @@ -0,0 +1,539 @@ +import { z } from "zod"; +import { + RunEvent, + StartEvent, + type UseCase, +} from "@x/shared/dist/runs.js"; +import type { + AssistantMessage, + UserMessage, +} from "@x/shared/dist/message.js"; +import { + assistantRef, + reduceTurn, + toolResultRef, + TurnEvent, + type ModelDescriptor, +} from "@x/shared/dist/turns.js"; +import { reduceSession, SessionEvent } from "@x/shared/dist/sessions.js"; + +// One-time conversion of a legacy `run.jsonl` log into the new event-sourced +// turn/session runtime. Pure and I/O-free so it is exhaustively testable +// against real fixtures; the runner (migrate.ts) owns the filesystem side. +// +// Mapping (decided in planning): +// - A run is one whole conversation. Turn boundaries are user messages. +// - copilot_chat runs -> one session + one turn per user message. +// - every other run -> a single standalone turn (sessionId: null) whose +// turnId is the ORIGINAL runId, so the live-note / +// background-task history views (which store that id +// and already fetch turn-first via sessions:getTurn) +// resolve it with no code change. +// - code_session runs are skipped by the runner; nothing to migrate there. +// +// The synthesized turn log is validated against reduceTurn before it is +// returned; anything the reducer would reject throws and the runner quarantines +// the run instead of writing a corrupt turn. + +type RunEventT = z.infer; +type StartT = z.infer; +type TurnEventT = z.infer; +type SessionEventT = z.infer; +type UserMessageT = z.infer; +type AssistantMessageT = z.infer; +type ModelDescriptorT = z.infer; + +export interface ConvertedTurn { + turnId: string; + events: TurnEventT[]; +} + +export interface ConvertedSession { + sessionId: string; + events: SessionEventT[]; +} + +export interface ConvertResult { + session?: ConvertedSession; + turns: ConvertedTurn[]; + /** True for copilot_chat runs (session + turns); false for standalone. */ + isSession: boolean; +} + +export class RunConversionError extends Error { + constructor(message: string) { + super(message); + this.name = "RunConversionError"; + } +} + +// A tool-call part lifted off an assistant message, with everything the run +// recorded about how it resolved. +interface ToolCallPlan { + toolCallId: string; + toolName: string; + input: unknown; + // Permission provenance from the legacy run, if any. + autoDecision?: { decision: "allow" | "deny"; reason: string; request: unknown }; + manualRequest?: { request: unknown }; + manualResponse?: { decision: "allow" | "deny" }; + // Execution outcome from the legacy run, if any. + result?: { output: unknown; isError: boolean }; + // Tool-role message content for this call (denial text fallback). + toolMessage?: string; +} + +interface ModelCallPlan { + message: AssistantMessageT; + toolCalls: ToolCallPlan[]; +} + +interface TurnPlan { + input: UserMessageT; + ts: string; + modelCalls: ModelCallPlan[]; +} + +function isToolCallResultObject(value: unknown): value is { success?: boolean } { + return typeof value === "object" && value !== null; +} + +function inferIsError(result: unknown): boolean { + if (isToolCallResultObject(result)) { + if (result.success === false) return true; + if ("error" in result && (result as { error?: unknown }).error) return true; + } + return false; +} + +function modelDescriptor(start: StartT): ModelDescriptorT { + // The new ModelDescriptor keeps provider as the name and model as the full + // id (no "/" split) — same convention the live agent resolver writes. + return { provider: start.provider, model: start.model }; +} + +function titleFrom(input: UserMessageT): string | undefined { + const content = input.content; + let text: string | undefined; + if (typeof content === "string") { + text = content; + } else if (Array.isArray(content)) { + const firstText = content.find( + (p): p is { type: "text"; text: string } => p.type === "text", + ); + text = firstText?.text; + } + if (!text) return undefined; + const trimmed = text.trim().replace(/\s+/g, " "); + if (!trimmed) return undefined; + return trimmed.length > 80 ? `${trimmed.slice(0, 77)}...` : trimmed; +} + +// Group the run log into turns keyed off user-message boundaries, folding the +// interleaved tool / permission events onto the tool-call parts they belong to. +function planTurns(log: RunEventT[], start: StartT): TurnPlan[] { + const turns: TurnPlan[] = []; + let current: TurnPlan | undefined; + // Per-turn lookups keyed by toolCallId, resolved when we emit each call. + let byId: Map = new Map(); + + const ensureToolPlan = (toolCallId: string): ToolCallPlan | undefined => + byId.get(toolCallId); + + for (const event of log) { + switch (event.type) { + case "message": { + const msg = event.message; + if (msg.role === "user") { + current = { + input: msg, + ts: event.ts ?? start.ts ?? new Date(0).toISOString(), + modelCalls: [], + }; + byId = new Map(); + turns.push(current); + } else if (msg.role === "assistant") { + if (!current) break; + const toolCalls: ToolCallPlan[] = []; + const parts = Array.isArray(msg.content) ? msg.content : []; + for (const part of parts) { + if (part.type !== "tool-call") continue; + const plan: ToolCallPlan = { + toolCallId: part.toolCallId, + toolName: part.toolName, + input: part.arguments, + }; + toolCalls.push(plan); + byId.set(part.toolCallId, plan); + } + current.modelCalls.push({ message: msg, toolCalls }); + } else if (msg.role === "tool") { + const plan = current && ensureToolPlan(msg.toolCallId); + if (plan) plan.toolMessage = msg.content; + } + break; + } + case "tool-result": { + if (!current || !event.toolCallId) break; + const plan = ensureToolPlan(event.toolCallId); + if (plan) { + plan.result = { + output: event.result ?? null, + isError: inferIsError(event.result), + }; + } + break; + } + case "tool-permission-auto-decision": { + if (!current) break; + const plan = ensureToolPlan(event.toolCallId); + if (plan) { + plan.autoDecision = { + decision: event.decision, + reason: event.reason, + request: event.permission ?? event.toolCall, + }; + } + break; + } + case "tool-permission-request": { + if (!current) break; + const plan = ensureToolPlan(event.toolCall.toolCallId); + if (plan) { + plan.manualRequest = { + request: event.permission ?? event.toolCall, + }; + } + break; + } + case "tool-permission-response": { + if (!current) break; + const plan = ensureToolPlan(event.toolCallId); + if (plan) { + plan.manualResponse = { + decision: event.response === "approve" ? "allow" : "deny", + }; + } + break; + } + default: + // start, tool-invocation (input taken from the assistant part), + // llm-stream-event, run-processing-*, spawn-subflow, ask-human-*, + // code-* and error events do not contribute to the transcript. + break; + } + } + + return turns; +} + +// Resolve a tool call's permission story + terminal result into the ordered +// list of new turn events. Throws (-> quarantine) if the run left the call +// dangling (no result and no denial), which the reducer cannot represent. +function toolEvents( + turnId: string, + ts: string, + plan: ToolCallPlan, +): TurnEventT[] { + const events: TurnEventT[] = []; + const hasPermission = + plan.autoDecision || plan.manualRequest || plan.manualResponse; + + let decision: "allow" | "deny" | undefined; + if (hasPermission) { + events.push({ + type: "tool_permission_required", + turnId, + ts, + toolCallId: plan.toolCallId, + toolName: plan.toolName, + request: (plan.autoDecision?.request ?? + plan.manualRequest?.request ?? + {}) as unknown, + } as TurnEventT); + if (plan.autoDecision) { + events.push({ + type: "tool_permission_classified", + turnId, + ts, + toolCallId: plan.toolCallId, + decision: plan.autoDecision.decision, + reason: plan.autoDecision.reason, + }); + } + // A human response is the effective decision; otherwise the classifier's. + if (plan.manualResponse) { + decision = plan.manualResponse.decision; + events.push({ + type: "tool_permission_resolved", + turnId, + ts, + toolCallId: plan.toolCallId, + decision, + source: "human", + }); + } else if (plan.autoDecision) { + decision = plan.autoDecision.decision; + events.push({ + type: "tool_permission_resolved", + turnId, + ts, + toolCallId: plan.toolCallId, + decision, + source: "classifier", + reason: plan.autoDecision.reason, + }); + } else { + // Only a manual request, never answered — the run ended waiting. + throw new RunConversionError( + `tool call ${plan.toolCallId} has an unresolved permission request`, + ); + } + } + + if (decision === "deny") { + const denial = + plan.toolMessage ?? "Unable to execute this tool: Permission was denied"; + events.push({ + type: "tool_result", + turnId, + ts, + toolCallId: plan.toolCallId, + toolName: plan.toolName, + source: "runtime", + result: { output: denial, isError: true }, + }); + return events; + } + + // Allowed (explicitly or by absence of a permission gate): the run must have + // recorded a terminal result, else the call is dangling and unrepresentable. + if (!plan.result) { + if (plan.toolMessage !== undefined) { + // Fall back to the tool-role message when the raw result event is + // missing (older logs). + events.push( + invocationEvent(turnId, ts, plan), + { + type: "tool_result", + turnId, + ts, + toolCallId: plan.toolCallId, + toolName: plan.toolName, + source: "sync", + result: { output: plan.toolMessage, isError: false }, + }, + ); + return events; + } + throw new RunConversionError( + `tool call ${plan.toolCallId} has no terminal result`, + ); + } + + events.push(invocationEvent(turnId, ts, plan), { + type: "tool_result", + turnId, + ts, + toolCallId: plan.toolCallId, + toolName: plan.toolName, + source: "sync", + result: { + output: (plan.result.output ?? null) as unknown, + isError: plan.result.isError, + }, + } as TurnEventT); + return events; +} + +function invocationEvent( + turnId: string, + ts: string, + plan: ToolCallPlan, +): TurnEventT { + return { + type: "tool_invocation_requested", + turnId, + ts, + toolCallId: plan.toolCallId, + toolId: plan.toolName, + toolName: plan.toolName, + execution: "sync", + input: (plan.input ?? null) as unknown, + } as TurnEventT; +} + +// Build one reduceTurn-legal turn log from a planned turn. +function buildTurn( + turnId: string, + sessionId: string | null, + previousTurnId: string | null, + start: StartT, + plan: TurnPlan, +): ConvertedTurn { + const ts = plan.ts; + const model = modelDescriptor(start); + const events: TurnEventT[] = []; + + events.push({ + type: "turn_created", + schemaVersion: 1, + turnId, + ts, + sessionId, + agent: { + requested: { agentId: start.agentName }, + resolved: { + agentId: start.agentName, + systemPrompt: "", + model, + tools: [], + }, + }, + context: previousTurnId ? { previousTurnId } : [], + input: plan.input, + config: { + autoPermission: start.permissionMode === "auto", + humanAvailable: true, + maxModelCalls: Math.max(20, plan.modelCalls.length + 5), + }, + }); + + if (plan.modelCalls.length === 0) { + throw new RunConversionError("turn has no assistant response"); + } + const last = plan.modelCalls[plan.modelCalls.length - 1]; + if (last.toolCalls.length > 0) { + // The final response still has open tool calls (interrupted run); the + // turn cannot be completed cleanly. + throw new RunConversionError("final model call has unresolved tool calls"); + } + + plan.modelCalls.forEach((call, index) => { + // Request references: only what is NEW since the previous model call. + let contextRef: { previousTurnId: string } | undefined; + let messages: string[]; + if (index === 0) { + if (previousTurnId) { + contextRef = { previousTurnId }; + messages = ["input"]; + } else { + messages = ["input"]; + } + } else { + const prev = plan.modelCalls[index - 1]; + messages = [ + assistantRef(index - 1), + ...prev.toolCalls.map((tc) => toolResultRef(tc.toolCallId)), + ]; + } + events.push({ + type: "model_call_requested", + turnId, + ts, + modelCallIndex: index, + request: { + ...(contextRef ? { contextRef } : {}), + messages, + parameters: {}, + }, + }); + events.push({ + type: "model_call_completed", + turnId, + ts, + modelCallIndex: index, + message: call.message, + finishReason: call.toolCalls.length > 0 ? "tool-calls" : "stop", + usage: {}, + }); + for (const tc of call.toolCalls) { + events.push(...toolEvents(turnId, ts, tc)); + } + }); + + events.push({ + type: "turn_completed", + turnId, + ts, + output: last.message, + finishReason: "stop", + usage: {}, + }); + + // The definition of a well-formed migration: the reducer accepts it. + reduceTurn(events); + return { turnId, events }; +} + +// A copilot run's session turns get fresh ids derived from each initiating user +// message id (unique, sortable, date-prefixed — satisfies the repo id regex). +function turnIdFor(runId: string, index: number): string { + return `${runId}-t${String(index).padStart(3, "0")}`; +} + +/** + * Convert one legacy run log into the new runtime shape. `runId` is the run's + * id (== the log's start.runId); pass it explicitly so the caller controls id + * preservation. Throws RunConversionError / TurnCorruptionError on anything the + * new reducers would reject. + */ +export function convertRun(log: RunEventT[], runId: string): ConvertResult { + const first = log[0]; + if (!first || first.type !== "start") { + throw new RunConversionError("run log does not begin with a start event"); + } + const start = first as StartT; + const useCase: z.infer | undefined = start.useCase; + + const plans = planTurns(log, start); + if (plans.length === 0) { + throw new RunConversionError("run has no user messages"); + } + + if (useCase === "copilot_chat") { + // Session id reuses the original runId; turns get derived ids. + const sessionId = runId; + const turns: ConvertedTurn[] = []; + const sessionEvents: SessionEventT[] = []; + const title = titleFrom(plans[0].input); + sessionEvents.push({ + type: "session_created", + schemaVersion: 1, + sessionId, + ts: plans[0].ts, + ...(title ? { title } : {}), + }); + let previousTurnId: string | null = null; + plans.forEach((plan, i) => { + const turnId = turnIdFor(runId, i); + const turn = buildTurn(turnId, sessionId, previousTurnId, start, plan); + turns.push(turn); + sessionEvents.push({ + type: "turn_appended", + sessionId, + ts: plan.ts, + turnId, + sessionSeq: i + 1, + agentId: start.agentName, + model: modelDescriptor(start), + }); + previousTurnId = turnId; + }); + reduceSession(sessionEvents); + return { + session: { sessionId, events: sessionEvents }, + turns, + isSession: true, + }; + } + + // Non-copilot: a single standalone turn whose id IS the runId. + if (plans.length > 1) { + throw new RunConversionError( + `non-copilot run has ${plans.length} user messages; expected a single turn`, + ); + } + const turn = buildTurn(runId, null, null, start, plans[0]); + return { turns: [turn], isSession: false }; +} diff --git a/apps/x/packages/core/src/migrations/runs/migrate.test.ts b/apps/x/packages/core/src/migrations/runs/migrate.test.ts new file mode 100644 index 00000000..c28440ac --- /dev/null +++ b/apps/x/packages/core/src/migrations/runs/migrate.test.ts @@ -0,0 +1,191 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { reduceTurn } from "@x/shared/dist/turns.js"; +import { reduceSession } from "@x/shared/dist/sessions.js"; +import { FSTurnRepo } from "../../turns/fs-repo.js"; +import { FSSessionRepo } from "../../sessions/fs-repo.js"; +import { migrateRuns } from "./migrate.js"; + +const fixturesDir = path.join( + path.dirname(fileURLToPath(import.meta.url)), + "__fixtures__", +); + +let tmp: string; +let runsDir: string; +let turnsRootDir: string; +let sessionsRootDir: string; +let archiveDir: string; +let logFile: string; + +function deps() { + return { + runsDir, + turnsRootDir, + sessionsRootDir, + archiveDir, + logFile, + now: () => "2026-07-02T00:00:00.000Z", + logger: () => undefined, + }; +} + +function copyFixture(name: string, targetName?: string): string { + const dest = path.join(runsDir, targetName ?? name); + fs.copyFileSync(path.join(fixturesDir, name), dest); + return dest; +} + +// Turn/session files nest under YYYY/MM/DD; find one by id anywhere in a tree. +function findFile(root: string, id: string): string | undefined { + const want = `${id}.jsonl`; + const walk = (dir: string): string | undefined => { + let entries: fs.Dirent[]; + try { + entries = fs.readdirSync(dir, { withFileTypes: true }); + } catch { + return undefined; + } + for (const entry of entries) { + const full = path.join(dir, entry.name); + if (entry.isDirectory()) { + const hit = walk(full); + if (hit) return hit; + } else if (entry.name === want) { + return full; + } + } + return undefined; + }; + return walk(root); +} + +beforeEach(() => { + tmp = fs.mkdtempSync(path.join(os.tmpdir(), "runs-migration-")); + runsDir = path.join(tmp, "runs"); + turnsRootDir = path.join(tmp, "storage", "turns"); + sessionsRootDir = path.join(tmp, "storage", "sessions"); + archiveDir = path.join(tmp, "runs-archive"); + logFile = path.join(tmp, "config", "runs-migration.json"); + fs.mkdirSync(runsDir, { recursive: true }); +}); + +afterEach(() => { + fs.rmSync(tmp, { recursive: true, force: true }); +}); + +describe("migrateRuns", () => { + it("migrates a non-copilot run to a standalone turn keyed by the runId, and archives the source", () => { + copyFixture("knowledge-single-turn.jsonl"); + const runId = "2026-07-02T15-26-22Z-0011690-000"; + + const summary = migrateRuns(deps()); + + expect(summary.scanned).toBe(1); + expect(summary.migratedTurns).toBe(1); + expect(summary.migratedSessions).toBe(0); + expect(summary.failed).toHaveLength(0); + // Turn file exists at the runId; no session written. + expect(findFile(turnsRootDir, runId)).toBeDefined(); + expect(fs.existsSync(sessionsRootDir)).toBe(false); + // Source archived out of the scan dir. + expect(fs.existsSync(path.join(runsDir, "knowledge-single-turn.jsonl"))).toBe( + false, + ); + expect( + fs.existsSync(path.join(archiveDir, "knowledge-single-turn.jsonl")), + ).toBe(true); + }); + + it("migrates a copilot run to a session + turns", () => { + copyFixture("copilot-multi-turn.jsonl"); + const runId = "2026-07-02T15-27-08Z-0011690-000"; + + const summary = migrateRuns(deps()); + + expect(summary.migratedSessions).toBe(1); + expect(summary.migratedTurns).toBe(3); + expect(findFile(sessionsRootDir, runId)).toBeDefined(); + expect(findFile(turnsRootDir, `${runId}-t000`)).toBeDefined(); + expect(findFile(turnsRootDir, `${runId}-t002`)).toBeDefined(); + }); + + it("writes files the real FS repos can read back and reduce", async () => { + copyFixture("copilot-multi-turn.jsonl"); + const runId = "2026-07-02T15-27-08Z-0011690-000"; + migrateRuns(deps()); + + const turnRepo = new FSTurnRepo({ turnsRootDir }); + const sessionRepo = new FSSessionRepo({ sessionsRootDir }); + + // The app's actual read path: repo.read (strict trailing-newline + id + // checks) then reduce. + const sessionEvents = await sessionRepo.read(runId); + const sessionState = reduceSession(sessionEvents); + expect(sessionState.turns).toHaveLength(3); + + const turnEvents = await turnRepo.read(`${runId}-t000`); + const turnState = reduceTurn(turnEvents); + expect(turnState.terminal?.type).toBe("turn_completed"); + }); + + it("skips code_session runs and leaves them in place", () => { + // Rewrite a fixture's useCase to code_session. + const raw = fs + .readFileSync(path.join(fixturesDir, "knowledge-single-turn.jsonl"), "utf-8") + .split("\n") + .filter((l) => l.trim()); + const start = JSON.parse(raw[0]); + start.useCase = "code_session"; + raw[0] = JSON.stringify(start); + fs.writeFileSync(path.join(runsDir, "code.jsonl"), raw.join("\n") + "\n"); + + const summary = migrateRuns(deps()); + + expect(summary.skipped).toBe(1); + expect(summary.migratedTurns).toBe(0); + // Left in place, not archived. + expect(fs.existsSync(path.join(runsDir, "code.jsonl"))).toBe(true); + expect(fs.existsSync(path.join(archiveDir, "code.jsonl"))).toBe(false); + }); + + it("leaves a malformed run in place and records the failure", () => { + fs.writeFileSync(path.join(runsDir, "broken.jsonl"), "not json\n"); + + const summary = migrateRuns(deps()); + + expect(summary.failed).toHaveLength(1); + expect(summary.failed[0].file).toBe("broken.jsonl"); + expect(fs.existsSync(path.join(runsDir, "broken.jsonl"))).toBe(true); + }); + + it("is idempotent — a second pass finds nothing left to migrate", () => { + copyFixture("background-task.jsonl"); + + const first = migrateRuns(deps()); + expect(first.migratedTurns).toBe(1); + + const second = migrateRuns(deps()); + expect(second.scanned).toBe(0); + expect(second.migratedTurns).toBe(0); + }); + + it("writes a migration log with counts", () => { + copyFixture("copilot-brew-deny.jsonl"); + migrateRuns(deps()); + + expect(fs.existsSync(logFile)).toBe(true); + const parsed = JSON.parse(fs.readFileSync(logFile, "utf-8")); + expect(parsed.ranAt).toBe("2026-07-02T00:00:00.000Z"); + expect(parsed.migratedSessions).toBe(1); + }); + + it("does nothing gracefully when the runs dir is absent", () => { + fs.rmSync(runsDir, { recursive: true, force: true }); + const summary = migrateRuns(deps()); + expect(summary.scanned).toBe(0); + }); +}); diff --git a/apps/x/packages/core/src/migrations/runs/migrate.ts b/apps/x/packages/core/src/migrations/runs/migrate.ts new file mode 100644 index 00000000..4fbe5dcc --- /dev/null +++ b/apps/x/packages/core/src/migrations/runs/migrate.ts @@ -0,0 +1,165 @@ +import fs from "node:fs"; +import path from "node:path"; +import { z } from "zod"; +import { WorkDir } from "../../config/config.js"; +import { RunEvent } from "@x/shared/dist/runs.js"; +import { TurnEvent } from "@x/shared/dist/turns.js"; +import { SessionEvent } from "@x/shared/dist/sessions.js"; +import { convertRun } from "./convert.js"; + +// One-time migration that ports legacy `runs/.jsonl` logs into the new +// turn/session runtime at startup. Runs before the session index is built. +// +// Design (see convert.ts for the mapping): +// - Successfully migrated runs are MOVED to `runs-archive/`. That move is the +// idempotency guard — migrated files leave the scan directory, so a re-run +// never re-processes them and no separate marker is needed. +// - Any run that fails to convert is LEFT in `runs/` (still served by the +// `runs:fetch` fallback the history views keep) and retried next launch. +// - code_session runs are skipped and left in place — Code mode still uses +// the old runtime, so those files must stay. +// Every step is defensive; a bad run (or the whole pass) never blocks boot. + +const ID_PATTERN = /^(\d{4})-(\d{2})-(\d{2})T[A-Za-z0-9-]+$/; + +export interface MigrateRunsDeps { + runsDir?: string; + turnsRootDir?: string; + sessionsRootDir?: string; + archiveDir?: string; + logFile?: string; + now?: () => string; + logger?: (message: string) => void; +} + +export interface MigrationSummary { + scanned: number; + migratedSessions: number; + migratedTurns: number; + skipped: number; + failed: Array<{ file: string; error: string }>; +} + +function idPath(rootDir: string, id: string): string { + const match = ID_PATTERN.exec(id); + if (!match) throw new Error(`invalid id: ${id}`); + const [, year, month, day] = match; + return path.join(rootDir, year, month, day, `${id}.jsonl`); +} + +// Mirrors the FS repos' serialization: revalidate each event, one JSON object +// per line, trailing newline. Overwrites any existing target so a re-run after +// a mid-migration crash converges cleanly (historical ids never collide with +// live turns, whose ids are minted from the current clock). +function writeJsonl( + rootDir: string, + id: string, + events: T[], + schema: z.ZodType, +): void { + const file = idPath(rootDir, id); + fs.mkdirSync(path.dirname(file), { recursive: true }); + const payload = + events.map((event) => JSON.stringify(schema.parse(event))).join("\n") + + "\n"; + fs.writeFileSync(file, payload); +} + +export function migrateRuns(deps: MigrateRunsDeps = {}): MigrationSummary { + const runsDir = deps.runsDir ?? path.join(WorkDir, "runs"); + const turnsRootDir = + deps.turnsRootDir ?? path.join(WorkDir, "storage", "turns"); + const sessionsRootDir = + deps.sessionsRootDir ?? path.join(WorkDir, "storage", "sessions"); + const archiveDir = deps.archiveDir ?? path.join(WorkDir, "runs-archive"); + const logFile = + deps.logFile ?? path.join(WorkDir, "config", "runs-migration.json"); + const log = + deps.logger ?? ((message: string) => console.log(`[runs-migration] ${message}`)); + + const summary: MigrationSummary = { + scanned: 0, + migratedSessions: 0, + migratedTurns: 0, + skipped: 0, + failed: [], + }; + + let files: string[]; + try { + files = fs + .readdirSync(runsDir) + .filter((name) => name.endsWith(".jsonl")) + .sort(); + } catch { + // No runs directory — nothing to migrate. + return summary; + } + + for (const file of files) { + summary.scanned++; + const source = path.join(runsDir, file); + try { + const raw = fs.readFileSync(source, "utf-8"); + const logEvents = raw + .split("\n") + .filter((line) => line.trim().length > 0) + .map((line) => RunEvent.parse(JSON.parse(line))); + + const start = logEvents[0]; + if (!start || start.type !== "start") { + throw new Error("run log does not begin with a start event"); + } + if (start.useCase === "code_session") { + // Code mode still uses the old runtime; leave the file in place. + summary.skipped++; + continue; + } + + const result = convertRun(logEvents, start.runId); + for (const turn of result.turns) { + writeJsonl(turnsRootDir, turn.turnId, turn.events, TurnEvent); + } + if (result.session) { + writeJsonl( + sessionsRootDir, + result.session.sessionId, + result.session.events, + SessionEvent, + ); + } + + // Success — archive the original out of the scan directory. + fs.mkdirSync(archiveDir, { recursive: true }); + fs.renameSync(source, path.join(archiveDir, file)); + + if (result.isSession) summary.migratedSessions++; + summary.migratedTurns += result.turns.length; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + summary.failed.push({ file, error: message }); + log(`failed to migrate ${file}: ${message}`); + } + } + + try { + fs.mkdirSync(path.dirname(logFile), { recursive: true }); + fs.writeFileSync( + logFile, + JSON.stringify( + { ranAt: deps.now?.() ?? new Date().toISOString(), ...summary }, + null, + 2, + ), + ); + } catch { + // Observability only; never fatal. + } + + log( + `scanned ${summary.scanned}, sessions ${summary.migratedSessions}, ` + + `turns ${summary.migratedTurns}, skipped ${summary.skipped}, ` + + `failed ${summary.failed.length}`, + ); + return summary; +}