mirror of
https://github.com/rowboatlabs/rowboat.git
synced 2026-07-12 21:02:17 +02:00
feat(migration): port legacy runs into the turn/session runtime at startup
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 <runId>-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) <noreply@anthropic.com>
This commit is contained in:
Ramnique Singh
parent
a8420f99ce
commit
17f18a35a1
9 changed files with 1096 additions and 0 deletions
|
|
@ -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<ISessions>('sessions').initialize();
|
||||
|
|
|
|||
|
|
@ -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":[]}
|
||||
|
|
@ -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":[]}
|
||||
|
|
@ -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":[]}
|
||||
|
|
@ -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":[]}
|
||||
140
apps/x/packages/core/src/migrations/runs/convert.test.ts
Normal file
140
apps/x/packages/core/src/migrations/runs/convert.test.ts
Normal file
|
|
@ -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<typeof RunEvent>;
|
||||
|
||||
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");
|
||||
});
|
||||
});
|
||||
});
|
||||
539
apps/x/packages/core/src/migrations/runs/convert.ts
Normal file
539
apps/x/packages/core/src/migrations/runs/convert.ts
Normal file
|
|
@ -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<typeof RunEvent>;
|
||||
type StartT = z.infer<typeof StartEvent>;
|
||||
type TurnEventT = z.infer<typeof TurnEvent>;
|
||||
type SessionEventT = z.infer<typeof SessionEvent>;
|
||||
type UserMessageT = z.infer<typeof UserMessage>;
|
||||
type AssistantMessageT = z.infer<typeof AssistantMessage>;
|
||||
type ModelDescriptorT = z.infer<typeof ModelDescriptor>;
|
||||
|
||||
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<string, ToolCallPlan> = 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<typeof UseCase> | 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 };
|
||||
}
|
||||
191
apps/x/packages/core/src/migrations/runs/migrate.test.ts
Normal file
191
apps/x/packages/core/src/migrations/runs/migrate.test.ts
Normal file
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
165
apps/x/packages/core/src/migrations/runs/migrate.ts
Normal file
165
apps/x/packages/core/src/migrations/runs/migrate.ts
Normal file
|
|
@ -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/<runId>.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<T>(
|
||||
rootDir: string,
|
||||
id: string,
|
||||
events: T[],
|
||||
schema: z.ZodType<T>,
|
||||
): 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;
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue