Skip to content

Commit 0bbac63

Browse files
steipetesteipete
committed
fix(tasks): migrate legacy agent attribution
1 parent 1fef20c commit 0bbac63

4 files changed

Lines changed: 435 additions & 16 deletions

File tree

‎src/commands/doctor-state-migrations.test.ts‎

Lines changed: 119 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -431,9 +431,9 @@ function writeLegacyTaskStateSidecars(root: string): {
431431
.prepare(
432432
`
433433
INSERT INTO task_runs (
434-
task_id, runtime, source_id, requester_session_key, child_session_key, run_id, task,
435-
status, delivery_status, notify_policy, created_at, last_event_at
436-
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
434+
task_id, runtime, source_id, requester_session_key, child_session_key, agent_id, run_id,
435+
task, status, delivery_status, notify_policy, created_at, last_event_at
436+
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
437437
`,
438438
)
439439
.run(
@@ -442,6 +442,7 @@ function writeLegacyTaskStateSidecars(root: string): {
442442
"nightly",
443443
"",
444444
"agent:main:cron:nightly",
445+
"ops",
445446
"legacy-task-run",
446447
"Legacy cron task",
447448
"running",
@@ -507,6 +508,36 @@ function writeLegacyTaskStateSidecars(root: string): {
507508
return { taskRunsPath, flowRunsPath };
508509
}
509510

511+
function appendLegacyCrossAgentTask(taskRunsPath: string): void {
512+
const sqlite = requireNodeSqlite();
513+
const db = new sqlite.DatabaseSync(taskRunsPath);
514+
try {
515+
db.prepare(
516+
`
517+
INSERT INTO task_runs (
518+
task_id, runtime, requester_session_key, child_session_key, agent_id, run_id, task,
519+
status, delivery_status, notify_policy, created_at, last_event_at
520+
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
521+
`,
522+
).run(
523+
"legacy-cross-agent",
524+
"subagent",
525+
"agent:main:main",
526+
"agent:worker:subagent:child",
527+
"main",
528+
"legacy-cross-agent-run",
529+
"Inspect worker state",
530+
"running",
531+
"pending",
532+
"done_only",
533+
130,
534+
140,
535+
);
536+
} finally {
537+
db.close();
538+
}
539+
}
540+
510541
async function detectAndRunMigrations(params: {
511542
root: string;
512543
cfg: OpenClawConfig;
@@ -2271,6 +2302,7 @@ describe("doctor legacy state migrations", () => {
22712302
ownerKey: "system:cron:nightly",
22722303
scopeKind: "system",
22732304
requesterSessionKey: "",
2305+
agentId: "ops",
22742306
runId: "legacy-task-run",
22752307
});
22762308
expect(taskState.deliveryStates.get("legacy-task")).toMatchObject({
@@ -2345,6 +2377,90 @@ describe("doctor legacy state migrations", () => {
23452377
});
23462378
});
23472379

2380+
it("canonicalizes cross-agent attribution while importing task sidecars", async () => {
2381+
const root = await makeTempRoot();
2382+
const { taskRunsPath } = writeLegacyTaskStateSidecars(root);
2383+
appendLegacyCrossAgentTask(taskRunsPath);
2384+
2385+
const result = await autoMigrateLegacyTaskStateSidecars({
2386+
env: { OPENCLAW_STATE_DIR: root } as NodeJS.ProcessEnv,
2387+
});
2388+
2389+
expect(result.warnings).toStrictEqual([]);
2390+
expect(result.changes).toContain("Migrated 2 task registry sidecar rows → shared SQLite state");
2391+
2392+
await withStateDir(root, async () => {
2393+
expect(loadTaskRegistryStateFromSqlite().tasks.get("legacy-cross-agent")).toMatchObject({
2394+
taskId: "legacy-cross-agent",
2395+
agentId: "worker",
2396+
requesterAgentId: "main",
2397+
requesterSessionKey: "agent:main:main",
2398+
childSessionKey: "agent:worker:subagent:child",
2399+
});
2400+
});
2401+
});
2402+
2403+
it("keeps task sidecars when only requester attribution conflicts", async () => {
2404+
const root = await makeTempRoot();
2405+
const { taskRunsPath } = writeLegacyTaskStateSidecars(root);
2406+
appendLegacyCrossAgentTask(taskRunsPath);
2407+
2408+
await withStateDir(root, async () => {
2409+
loadTaskRegistryStateFromSqlite();
2410+
closeOpenClawStateDatabaseForTest();
2411+
const sqlite = requireNodeSqlite();
2412+
const db = new sqlite.DatabaseSync(path.join(root, "state", "openclaw.sqlite"));
2413+
try {
2414+
db.prepare(
2415+
`INSERT INTO task_runs (
2416+
task_id,
2417+
runtime,
2418+
requester_session_key,
2419+
owner_key,
2420+
scope_kind,
2421+
child_session_key,
2422+
agent_id,
2423+
requester_agent_id,
2424+
run_id,
2425+
task,
2426+
status,
2427+
delivery_status,
2428+
notify_policy,
2429+
created_at,
2430+
last_event_at
2431+
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
2432+
).run(
2433+
"legacy-cross-agent",
2434+
"subagent",
2435+
"agent:main:main",
2436+
"agent:main:main",
2437+
"session",
2438+
"agent:worker:subagent:child",
2439+
"worker",
2440+
"other-requester",
2441+
"legacy-cross-agent-run",
2442+
"Inspect worker state",
2443+
"running",
2444+
"pending",
2445+
"done_only",
2446+
130,
2447+
140,
2448+
);
2449+
} finally {
2450+
db.close();
2451+
}
2452+
});
2453+
2454+
const result = await autoMigrateLegacyTaskStateSidecars({
2455+
env: { OPENCLAW_STATE_DIR: root } as NodeJS.ProcessEnv,
2456+
});
2457+
2458+
expect(result.warnings).toContain(
2459+
"Left task registry sidecar in place because 1 row already existed in shared state: legacy-cross-agent",
2460+
);
2461+
expect(fs.existsSync(taskRunsPath)).toBe(true);
2462+
});
2463+
23482464
it("keeps task sidecars when shared state already has conflicting task rows", async () => {
23492465
const root = await makeTempRoot();
23502466
const { taskRunsPath, flowRunsPath } = writeLegacyTaskStateSidecars(root);

‎src/infra/state-migrations.ts‎

Lines changed: 25 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -658,6 +658,19 @@ function normalizeLegacyTaskRow(row: Record<string, unknown>): SqliteBindRow {
658658
const ownerKey = ownerRaw || requesterRaw || `system:${runtime}:${sourceId || taskId}`;
659659
const scopeRaw = typeof row.scope_kind === "string" ? row.scope_kind : "";
660660
const scopeKind = scopeRaw === "system" || ownerKey.startsWith("system:") ? "system" : "session";
661+
const childSessionKey =
662+
typeof row.child_session_key === "string" ? row.child_session_key.trim() : "";
663+
const persistedAgentId = typeof row.agent_id === "string" ? row.agent_id.trim() : "";
664+
const isSpawnRuntime = runtime === "subagent" || runtime === "acp";
665+
const childAgentId = isSpawnRuntime ? parseAgentSessionKey(childSessionKey)?.agentId : undefined;
666+
const requesterAgentId =
667+
(typeof row.requester_agent_id === "string" ? row.requester_agent_id.trim() : "") ||
668+
(isSpawnRuntime
669+
? (parseAgentSessionKey(ownerKey)?.agentId ??
670+
parseAgentSessionKey(requesterRaw)?.agentId ??
671+
(childAgentId && persistedAgentId !== childAgentId ? persistedAgentId : ""))
672+
: "");
673+
const executorAgentId = requesterAgentId ? childAgentId || persistedAgentId : persistedAgentId;
661674
return {
662675
task_id: taskId,
663676
runtime,
@@ -666,10 +679,11 @@ function normalizeLegacyTaskRow(row: Record<string, unknown>): SqliteBindRow {
666679
requester_session_key: scopeKind === "system" ? "" : requesterRaw || ownerKey,
667680
owner_key: ownerKey,
668681
scope_kind: scopeKind,
669-
child_session_key: legacyBindValue(row.child_session_key),
682+
child_session_key: childSessionKey || null,
670683
parent_flow_id: legacyBindValue(row.parent_flow_id),
671684
parent_task_id: legacyBindValue(row.parent_task_id),
672-
agent_id: legacyBindValue(row.agent_id),
685+
agent_id: executorAgentId || null,
686+
requester_agent_id: requesterAgentId || null,
673687
run_id: legacyBindValue(row.run_id),
674688
label: legacyBindValue(row.label),
675689
task: legacyBindValue(row.task ?? ""),
@@ -760,6 +774,7 @@ function readLegacyTaskRows(sourcePath: string): SqliteBindRow[] {
760774
pickLegacyColumn(columns, "parent_flow_id"),
761775
pickLegacyColumn(columns, "parent_task_id"),
762776
pickLegacyColumn(columns, "agent_id"),
777+
pickLegacyColumn(columns, "requester_agent_id"),
763778
pickLegacyColumn(columns, "run_id"),
764779
pickLegacyColumn(columns, "label"),
765780
"task",
@@ -851,15 +866,15 @@ function insertTaskRunRowSql(db: DatabaseSync, row: SqliteBindRow): void {
851866
`
852867
INSERT INTO task_runs (
853868
task_id, runtime, task_kind, source_id, requester_session_key, owner_key, scope_kind,
854-
child_session_key, parent_flow_id, parent_task_id, agent_id, run_id, label, task, status,
855-
delivery_status, notify_policy, created_at, started_at, ended_at, last_event_at,
856-
cleanup_after, error, progress_summary, terminal_summary, terminal_outcome
869+
child_session_key, parent_flow_id, parent_task_id, agent_id, requester_agent_id, run_id,
870+
label, task, status, delivery_status, notify_policy, created_at, started_at, ended_at,
871+
last_event_at, cleanup_after, error, progress_summary, terminal_summary, terminal_outcome
857872
) VALUES (
858873
@task_id, @runtime, @task_kind, @source_id, @requester_session_key, @owner_key,
859-
@scope_kind, @child_session_key, @parent_flow_id, @parent_task_id, @agent_id, @run_id,
860-
@label, @task, @status, @delivery_status, @notify_policy, @created_at, @started_at,
861-
@ended_at, @last_event_at, @cleanup_after, @error, @progress_summary, @terminal_summary,
862-
@terminal_outcome
874+
@scope_kind, @child_session_key, @parent_flow_id, @parent_task_id, @agent_id,
875+
@requester_agent_id, @run_id, @label, @task, @status, @delivery_status, @notify_policy,
876+
@created_at, @started_at, @ended_at, @last_event_at, @cleanup_after, @error,
877+
@progress_summary, @terminal_summary, @terminal_outcome
863878
)
864879
`,
865880
).run(row);
@@ -933,6 +948,7 @@ async function migrateLegacyTaskRunsSidecar(params: {
933948
"parent_flow_id",
934949
"parent_task_id",
935950
"agent_id",
951+
"requester_agent_id",
936952
"run_id",
937953
"label",
938954
"task",

0 commit comments

Comments
 (0)