From 1fef20c96bdcc0dc7db4f8a06db67eb4b1b7963e Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 15 Jun 2026 09:15:56 -0400 Subject: [PATCH] fix(tasks): preserve requester agent attribution --- docs/automation/tasks.md | 2 +- docs/gateway/protocol.md | 4 +- src/agents/acp-spawn.test.ts | 31 +++++++++++ src/agents/acp-spawn.ts | 27 ++++++---- src/agents/subagent-registry-run-manager.ts | 2 + src/agents/subagent-spawn.test.ts | 8 ++- src/agents/subagent-spawn.ts | 1 + src/agents/tools/sessions-spawn-tool.test.ts | 3 ++ src/agents/tools/sessions-spawn-tool.ts | 2 + src/gateway/server-methods/artifacts.test.ts | 56 ++++++++++++++++++++ src/gateway/server-methods/artifacts.ts | 25 +++++++-- src/state/openclaw-state-db.generated.d.ts | 1 + src/state/openclaw-state-db.test.ts | 22 ++++++++ src/state/openclaw-state-db.ts | 1 + src/state/openclaw-state-schema.generated.ts | 1 + src/state/openclaw-state-schema.sql | 1 + src/tasks/detached-task-runtime-contract.ts | 1 + src/tasks/task-registry.store.sqlite.ts | 4 ++ src/tasks/task-registry.store.test.ts | 12 +++-- src/tasks/task-registry.ts | 26 ++++++++- src/tasks/task-registry.types.ts | 3 ++ src/tasks/task-status-access.ts | 6 ++- 22 files changed, 216 insertions(+), 23 deletions(-) diff --git a/docs/automation/tasks.md b/docs/automation/tasks.md index 030837633be..d4d6d42524a 100644 --- a/docs/automation/tasks.md +++ b/docs/automation/tasks.md @@ -360,7 +360,7 @@ A sweeper runs every **60 seconds** and handles four things: - A task may reference a `childSessionKey` (where work runs) and a `requesterSessionKey` (who started it). Sessions are conversation context; tasks are activity tracking on top of that. + A task may reference a `childSessionKey` (where work runs) and a `requesterSessionKey` (who started it). Its `agentId` identifies the agent executing the work, while the requester and owner fields preserve launch and control context. Sessions are conversation context; tasks are activity tracking on top of that. A task's `runId` links to the agent run doing the work. Agent lifecycle events (start, end, error) automatically update the task status - you do not need to manage the lifecycle manually. diff --git a/docs/gateway/protocol.md b/docs/gateway/protocol.md index b5ac296a375..49294cc09eb 100644 --- a/docs/gateway/protocol.md +++ b/docs/gateway/protocol.md @@ -542,7 +542,9 @@ runtime state. `TaskSummary` includes `id`, `status`, and optional metadata such as `kind`, `runtime`, `title`, `agentId`, `sessionKey`, `childSessionKey`, `ownerKey`, `runId`, `taskId`, `flowId`, `parentTaskId`, `sourceId`, timestamps, progress, -terminal summary, and sanitized error text. +terminal summary, and sanitized error text. `agentId` identifies the agent +executing the task; `sessionKey` and `ownerKey` preserve requester and control +context. ### Operator helper methods diff --git a/src/agents/acp-spawn.test.ts b/src/agents/acp-spawn.test.ts index c1afbad35d7..6ad77a4dfd3 100644 --- a/src/agents/acp-spawn.test.ts +++ b/src/agents/acp-spawn.test.ts @@ -2787,6 +2787,37 @@ describe("spawnAcpDirect", () => { expect(hoisted.startAcpSpawnParentStreamRelayMock).not.toHaveBeenCalled(); }); + it("persists separate requester and executor agents for global cross-agent tasks", async () => { + replaceSpawnConfig({ + ...hoisted.state.cfg, + session: { + ...hoisted.state.cfg.session, + scope: "global", + }, + }); + + const result = await spawnAcpDirect( + { + task: "Investigate flaky tests", + agentId: "codex", + }, + { + agentSessionKey: "global", + requesterAgentIdOverride: "research", + }, + ); + + expectAcceptedSpawn(result); + expect(hoisted.createRunningTaskRunMock).toHaveBeenCalledWith( + expect.objectContaining({ + ownerKey: "global", + childSessionKey: expect.stringMatching(/^agent:codex:acp:/), + agentId: "codex", + requesterAgentId: "research", + }), + ); + }); + it("does not implicitly stream for subagent requester sessions when heartbeat is disabled", async () => { replaceSpawnConfig({ ...hoisted.state.cfg, diff --git a/src/agents/acp-spawn.ts b/src/agents/acp-spawn.ts index 2440788d2d2..74243904926 100644 --- a/src/agents/acp-spawn.ts +++ b/src/agents/acp-spawn.ts @@ -162,6 +162,7 @@ function toGatewayImageAttachments( export type SpawnAcpContext = { agentSessionKey?: string; + requesterAgentIdOverride?: string; agentChannel?: string; agentAccountId?: string; agentTo?: string; @@ -766,6 +767,7 @@ function prepareAcpThreadBinding(params: { function resolveAcpSpawnRequesterState(params: { cfg: OpenClawConfig; parentSessionKey?: string; + requesterAgentId: string; targetAgentId: string; ctx: SpawnAcpContext; subagentStore?: SessionCapabilityStore; @@ -784,8 +786,6 @@ function resolveAcpSpawnRequesterState(params: { typeof params.ctx.agentThreadId === "string" ? Boolean(normalizeOptionalString(params.ctx.agentThreadId)) : params.ctx.agentThreadId != null; - const requesterAgentId = requesterParsedSession?.agentId; - return { parentSessionKey: params.parentSessionKey, isSubagentSession, @@ -796,17 +796,17 @@ function resolveAcpSpawnRequesterState(params: { sessionKey: params.parentSessionKey, }), heartbeatRelayRouteUsable: - params.parentSessionKey && requesterAgentId + params.parentSessionKey && params.requesterAgentId ? hasSessionLocalHeartbeatRelayRoute({ cfg: params.cfg, parentSessionKey: params.parentSessionKey, - requesterAgentId, + requesterAgentId: params.requesterAgentId, }) : false, origin: resolveRequesterOriginForChild({ cfg: params.cfg, targetAgentId: params.targetAgentId, - requesterAgentId: normalizeAgentId(requesterAgentId), + requesterAgentId: params.requesterAgentId, requesterChannel: params.ctx.agentChannel, requesterAccountId: params.ctx.agentAccountId, requesterTo: params.ctx.agentTo, @@ -820,6 +820,7 @@ function resolveAcpSpawnRequesterState(params: { function resolveAcpSubagentEnvelopeState(params: { cfg: OpenClawConfig; requesterSessionKey?: string; + requesterAgentId: string; targetAgentId: string; requestedAgentId?: string; subagentStore?: SessionCapabilityStore; @@ -860,9 +861,8 @@ function resolveAcpSubagentEnvelopeState(params: { }; } - const requesterAgentId = normalizeAgentId(parseAgentSessionKey(requesterSessionKey)?.agentId); const requireAgentId = - resolveAgentConfig(params.cfg, requesterAgentId)?.subagents?.requireAgentId ?? + resolveAgentConfig(params.cfg, params.requesterAgentId)?.subagents?.requireAgentId ?? params.cfg.agents?.defaults?.subagents?.requireAgentId ?? false; if (requireAgentId && !params.requestedAgentId?.trim()) { @@ -873,11 +873,11 @@ function resolveAcpSubagentEnvelopeState(params: { } const targetPolicy = resolveSubagentTargetPolicy({ - requesterAgentId, + requesterAgentId: params.requesterAgentId, targetAgentId: params.targetAgentId, requestedAgentId: params.requestedAgentId, allowAgents: - resolveAgentConfig(params.cfg, requesterAgentId)?.subagents?.allowAgents ?? + resolveAgentConfig(params.cfg, params.requesterAgentId)?.subagents?.allowAgents ?? params.cfg.agents?.defaults?.subagents?.allowAgents, configuredAgentIds: resolveConfiguredAcpSubagentTargetIds(params.cfg), }); @@ -1275,6 +1275,9 @@ export async function spawnAcpDirect( cfg, requesterSessionKey: ctx.agentSessionKey, }); + const requesterAgentId = normalizeAgentId( + ctx.requesterAgentIdOverride ?? parseAgentSessionKey(requesterInternalKey)?.agentId, + ); if (!isAcpEnabledByPolicy(cfg)) { return createAcpSpawnFailure({ status: "forbidden", @@ -1370,6 +1373,7 @@ export async function spawnAcpDirect( const requesterState = resolveAcpSpawnRequesterState({ cfg, parentSessionKey, + requesterAgentId, targetAgentId, ctx, subagentStore, @@ -1377,6 +1381,7 @@ export async function spawnAcpDirect( const subagentEnvelopeState = resolveAcpSubagentEnvelopeState({ cfg, requesterSessionKey: requesterInternalKey, + requesterAgentId, targetAgentId, requestedAgentId: params.agentId, subagentStore, @@ -1636,6 +1641,8 @@ export async function spawnAcpDirect( scopeKind: "session", requesterOrigin: requesterState.origin, childSessionKey: sessionKey, + agentId: targetAgentId, + requesterAgentId, runId: childRunId, label: params.label, task: params.task, @@ -1675,6 +1682,8 @@ export async function spawnAcpDirect( scopeKind: "session", requesterOrigin: requesterState.origin, childSessionKey: sessionKey, + agentId: targetAgentId, + requesterAgentId, runId: childRunId, label: params.label, task: params.task, diff --git a/src/agents/subagent-registry-run-manager.ts b/src/agents/subagent-registry-run-manager.ts index 2adfb15ccb1..398785f283e 100644 --- a/src/agents/subagent-registry-run-manager.ts +++ b/src/agents/subagent-registry-run-manager.ts @@ -148,6 +148,7 @@ export type RegisterSubagentRunParams = { task: string; taskName?: string; agentId?: string; + requesterAgentId?: string; cleanup: "delete" | "keep"; label?: string; model?: string; @@ -692,6 +693,7 @@ export function createSubagentRunManager(params: { label: registerParams.label, task: registerParams.task, agentId: registerParams.agentId, + requesterAgentId: registerParams.requesterAgentId, deliveryStatus: registerParams.expectsCompletionMessage === false ? "not_applicable" : "pending", startedAt: now, diff --git a/src/agents/subagent-spawn.test.ts b/src/agents/subagent-spawn.test.ts index 0abc2b76ed1..1c863991e5d 100644 --- a/src/agents/subagent-spawn.test.ts +++ b/src/agents/subagent-spawn.test.ts @@ -189,6 +189,9 @@ describe("spawnSubagentDirect seam flow", () => { it("registers the target agent id for cross-agent task attribution", async () => { hoisted.configOverride = createConfigOverride({ + session: { + scope: "global", + }, agents: { defaults: { workspace: os.tmpdir(), @@ -215,7 +218,8 @@ describe("spawnSubagentDirect seam flow", () => { agentId: "worker", }, { - agentSessionKey: "agent:main:main", + agentSessionKey: "global", + requesterAgentIdOverride: "main", }, ); @@ -224,6 +228,8 @@ describe("spawnSubagentDirect seam flow", () => { const registerInput = firstRegisteredSubagentRun(); expect(registerInput.childSessionKey).toBe(result.childSessionKey); expect(registerInput.agentId).toBe("worker"); + expect(registerInput.requesterSessionKey).toBe("global"); + expect(registerInput.requesterAgentId).toBe("main"); }); it("accepts a spawned run across session patching, runtime-model persistence, registry registration, and lifecycle emission", async () => { diff --git a/src/agents/subagent-spawn.ts b/src/agents/subagent-spawn.ts index 6834c853160..c67c0e26da7 100644 --- a/src/agents/subagent-spawn.ts +++ b/src/agents/subagent-spawn.ts @@ -1671,6 +1671,7 @@ export async function spawnSubagentDirect( task, taskName, agentId: targetAgentId, + requesterAgentId, cleanup, label: label || undefined, model: resolvedModel, diff --git a/src/agents/tools/sessions-spawn-tool.test.ts b/src/agents/tools/sessions-spawn-tool.test.ts index 9815ef55134..1cf0c030b66 100644 --- a/src/agents/tools/sessions-spawn-tool.test.ts +++ b/src/agents/tools/sessions-spawn-tool.test.ts @@ -556,6 +556,7 @@ describe("sessions_spawn tool", () => { registerAcpBackendForTest(); const tool = createSessionsSpawnTool({ agentSessionKey: "agent:main:main", + requesterAgentIdOverride: "main", agentChannel: "quietchat", agentAccountId: "default", agentTo: "channel:123", @@ -587,11 +588,13 @@ describe("sessions_spawn tool", () => { expect(spawnArgs.streamTo).toBe("parent"); const spawnContext = mockCallArg(hoisted.spawnAcpDirectMock, 0, 1, "spawnAcpDirect"); expect(spawnContext.agentSessionKey).toBe("agent:main:main"); + expect(spawnContext.requesterAgentIdOverride).toBe("main"); expect(hoisted.spawnSubagentDirectMock).not.toHaveBeenCalled(); const registration = mockCallArg(hoisted.registerSubagentRunMock, 0, 0, "registerSubagentRun"); expect(registration.runId).toBe("run-acp"); expect(registration.childSessionKey).toBe("agent:codex:acp:1"); expect(registration.requesterSessionKey).toBe("agent:main:main"); + expect(registration.requesterAgentId).toBe("main"); expect(registration.task).toBe("investigate the failing CI run"); expect(registration.cleanup).toBe("keep"); expect(registration.spawnMode).toBe("session"); diff --git a/src/agents/tools/sessions-spawn-tool.ts b/src/agents/tools/sessions-spawn-tool.ts index 93c7c592223..22e5c6240d3 100644 --- a/src/agents/tools/sessions-spawn-tool.ts +++ b/src/agents/tools/sessions-spawn-tool.ts @@ -400,6 +400,7 @@ export function createSessionsSpawnTool( }, { agentSessionKey: opts?.agentSessionKey, + requesterAgentIdOverride: opts?.requesterAgentIdOverride, agentChannel: opts?.agentChannel, agentAccountId: opts?.agentAccountId, agentTo: opts?.agentTo, @@ -447,6 +448,7 @@ export function createSessionsSpawnTool( requesterDisplayKey: ownership.completionRequesterDisplayKey, task, taskName, + requesterAgentId: opts?.requesterAgentIdOverride, cleanup: trackedCleanup, label: label || undefined, runTimeoutSeconds: result.runTimeoutSeconds, diff --git a/src/gateway/server-methods/artifacts.test.ts b/src/gateway/server-methods/artifacts.test.ts index 9c23505db58..4f22773157a 100644 --- a/src/gateway/server-methods/artifacts.test.ts +++ b/src/gateway/server-methods/artifacts.test.ts @@ -295,6 +295,7 @@ describe("artifacts RPC handlers", () => { hoisted.getTaskSessionLookupByIdForStatus.mockReturnValue({ agentId: "work", requesterSessionKey: "global", + ownerKey: "global", }); mockedMessages([assistantFileMessage({ title: "task.txt", taskId: "task-global" })]); @@ -547,6 +548,61 @@ describe("artifacts RPC handlers", () => { }); }); + it("keeps cross-agent task artifacts scoped to the requester transcript", async () => { + hoisted.getTaskSessionLookupByIdForStatus.mockReturnValue({ + requesterSessionKey: "agent:main:main", + ownerKey: "agent:main:main", + runId: "run-for-task-1", + agentId: "worker", + requesterAgentId: "main", + }); + mockedMessages([ + assistantImageMessage({ alt: "task-result.png", data: "dGFyZ2V0", taskId: "task-1" }), + ]); + + const { calls } = await listArtifacts( + { taskId: "task-1", agentId: "worker" }, + { id: "task-cross-agent-requester-session" }, + ); + + expect(hoisted.resolveSessionKeyForRun).not.toHaveBeenCalled(); + expect(hoisted.loadSessionEntry).toHaveBeenCalledWith("agent:main:main"); + expectFields(expectFirstArtifact(calls), { + taskId: "task-1", + sessionKey: "agent:main:main", + }); + }); + + it("uses the requester agent store for cross-agent global task artifacts", async () => { + hoisted.getTaskSessionLookupByIdForStatus.mockReturnValue({ + requesterSessionKey: "global", + ownerKey: "global", + runId: "run-for-task-1", + agentId: "worker", + requesterAgentId: "main", + }); + mockedMessages([ + assistantImageMessage({ alt: "task-result.png", data: "dGFyZ2V0", taskId: "task-1" }), + ]); + + const { calls } = await listArtifacts( + { taskId: "task-1", agentId: "worker" }, + { + id: "task-cross-agent-global-requester", + context: runtimeContext({ + session: { scope: "global" }, + agents: { list: [{ id: "main", default: true }, { id: "worker" }] }, + }), + }, + ); + + expect(hoisted.loadSessionEntry).toHaveBeenCalledWith("global", { agentId: "main" }); + expectFields(expectFirstArtifact(calls), { + taskId: "task-1", + sessionKey: "global", + }); + }); + it("derives taskId artifact scope from requesterSessionKey when task agentId is absent", async () => { hoisted.getTaskSessionLookupByIdForStatus.mockReturnValue({ requesterSessionKey: "agent:work:main", diff --git a/src/gateway/server-methods/artifacts.ts b/src/gateway/server-methods/artifacts.ts index 22578315e76..79f4a2e2814 100644 --- a/src/gateway/server-methods/artifacts.ts +++ b/src/gateway/server-methods/artifacts.ts @@ -422,8 +422,14 @@ function resolveQuerySession( if (query.taskId) { const task = getTaskSessionLookupByIdForStatus(query.taskId); const requesterSessionKey = asNonEmptyString(task?.requesterSessionKey); - const taskAgentId = - asNonEmptyString(task?.agentId) ?? resolveRequesterSessionAgentId(requesterSessionKey, cfg); + const ownerAgentId = parseAgentSessionKey(task?.ownerKey)?.agentId; + const requesterAgentId = + asNonEmptyString(task?.requesterAgentId) ?? + ownerAgentId ?? + (requesterSessionKey === "global" + ? undefined + : resolveRequesterSessionAgentId(requesterSessionKey, cfg)); + const taskAgentId = asNonEmptyString(task?.agentId) ?? requesterAgentId; if ( query.agentId && taskAgentId && @@ -431,11 +437,20 @@ function resolveQuerySession( ) { return undefined; } - const agentId = query.agentId ?? taskAgentId ?? resolveDefaultAgentId(cfg ?? {}); if (requesterSessionKey) { - const scopedSessionKey = resolveScopedArtifactSessionKey(requesterSessionKey, agentId, cfg); - return scopedSessionKey ? { sessionKey: scopedSessionKey, agentId } : undefined; + // task.agentId identifies the executor. requesterAgentId keeps global + // requester transcripts in the correct agent store across restarts. + const sessionAgentId = requesterAgentId ?? taskAgentId ?? resolveDefaultAgentId(cfg ?? {}); + const scopedSessionKey = resolveScopedArtifactSessionKey( + requesterSessionKey, + sessionAgentId, + cfg, + ); + return scopedSessionKey + ? { sessionKey: scopedSessionKey, agentId: sessionAgentId } + : undefined; } + const agentId = query.agentId ?? taskAgentId ?? resolveDefaultAgentId(cfg ?? {}); const runId = asNonEmptyString(task?.runId); const sessionKey = runId ? resolveSessionKeyForRun(runId, { agentId }) : undefined; const scopedSessionKey = resolveScopedArtifactSessionKey(sessionKey, agentId, cfg); diff --git a/src/state/openclaw-state-db.generated.d.ts b/src/state/openclaw-state-db.generated.d.ts index 16fe7d08c63..edbd88bb98d 100644 --- a/src/state/openclaw-state-db.generated.d.ts +++ b/src/state/openclaw-state-db.generated.d.ts @@ -849,6 +849,7 @@ export interface TaskRuns { parent_flow_id: string | null; parent_task_id: string | null; progress_summary: string | null; + requester_agent_id: string | null; requester_session_key: string | null; run_id: string | null; runtime: string; diff --git a/src/state/openclaw-state-db.test.ts b/src/state/openclaw-state-db.test.ts index 3f5cc7f9bf1..7730fd91c63 100644 --- a/src/state/openclaw-state-db.test.ts +++ b/src/state/openclaw-state-db.test.ts @@ -79,6 +79,28 @@ describe("openclaw state database", () => { expect(database.path).toBe(path.join(stateDir, "state", "openclaw.sqlite")); }); + it("adds requester agent attribution to existing task tables", () => { + const stateDir = createTempStateDir(); + const database = openOpenClawStateDatabase({ + env: { OPENCLAW_STATE_DIR: stateDir }, + }); + const databasePath = database.path; + closeOpenClawStateDatabaseForTest(); + + const { DatabaseSync } = requireNodeSqlite(); + const legacyDb = new DatabaseSync(databasePath); + legacyDb.exec("ALTER TABLE task_runs DROP COLUMN requester_agent_id"); + legacyDb.close(); + + const reopened = openOpenClawStateDatabase({ + env: { OPENCLAW_STATE_DIR: stateDir }, + }); + const columns = reopened.db.prepare("PRAGMA table_info(task_runs)").all() as Array<{ + name?: string; + }>; + expect(columns.some((column) => column.name === "requester_agent_id")).toBe(true); + }); + it("opens databases with early cron tables before creating cron indexes", () => { const stateDir = createTempStateDir(); const databasePath = path.join(stateDir, "state", "openclaw.sqlite"); diff --git a/src/state/openclaw-state-db.ts b/src/state/openclaw-state-db.ts index b7269dc068f..e75b260a905 100644 --- a/src/state/openclaw-state-db.ts +++ b/src/state/openclaw-state-db.ts @@ -857,6 +857,7 @@ function ensureAdditiveStateColumns(db: DatabaseSync): void { ensureColumn(db, "gateway_restart_sentinel", "continuation_json TEXT"); ensureColumn(db, "gateway_restart_sentinel", "doctor_hint TEXT"); ensureColumn(db, "gateway_restart_sentinel", "stats_json TEXT"); + ensureColumn(db, "task_runs", "requester_agent_id TEXT"); ensureColumn(db, "subagent_runs", "task_name TEXT"); } diff --git a/src/state/openclaw-state-schema.generated.ts b/src/state/openclaw-state-schema.generated.ts index bb4a67fe492..629437dc767 100644 --- a/src/state/openclaw-state-schema.generated.ts +++ b/src/state/openclaw-state-schema.generated.ts @@ -1003,6 +1003,7 @@ CREATE TABLE IF NOT EXISTS task_runs ( parent_flow_id TEXT, parent_task_id TEXT, agent_id TEXT, + requester_agent_id TEXT, run_id TEXT, label TEXT, task TEXT NOT NULL, diff --git a/src/state/openclaw-state-schema.sql b/src/state/openclaw-state-schema.sql index 2dd9d03690a..0409b2c3cb7 100644 --- a/src/state/openclaw-state-schema.sql +++ b/src/state/openclaw-state-schema.sql @@ -998,6 +998,7 @@ CREATE TABLE IF NOT EXISTS task_runs ( parent_flow_id TEXT, parent_task_id TEXT, agent_id TEXT, + requester_agent_id TEXT, run_id TEXT, label TEXT, task TEXT NOT NULL, diff --git a/src/tasks/detached-task-runtime-contract.ts b/src/tasks/detached-task-runtime-contract.ts index 04c030244af..5dfc0e1c7a1 100644 --- a/src/tasks/detached-task-runtime-contract.ts +++ b/src/tasks/detached-task-runtime-contract.ts @@ -23,6 +23,7 @@ export type DetachedTaskCreateParams = { childSessionKey?: string; parentTaskId?: string; agentId?: string; + requesterAgentId?: string; runId?: string; label?: string; task: string; diff --git a/src/tasks/task-registry.store.sqlite.ts b/src/tasks/task-registry.store.sqlite.ts index 4efc003e70a..c7ef7fd2fff 100644 --- a/src/tasks/task-registry.store.sqlite.ts +++ b/src/tasks/task-registry.store.sqlite.ts @@ -57,6 +57,7 @@ const TASK_RUN_SELECT_COLUMNS = [ "parent_flow_id", "parent_task_id", "agent_id", + "requester_agent_id", "run_id", "label", "task", @@ -109,6 +110,7 @@ function rowToTaskRecord(row: TaskRegistryRow): TaskRecord { ...(row.parent_flow_id ? { parentFlowId: row.parent_flow_id } : {}), ...(row.parent_task_id ? { parentTaskId: row.parent_task_id } : {}), ...(row.agent_id ? { agentId: row.agent_id } : {}), + ...(row.requester_agent_id ? { requesterAgentId: row.requester_agent_id } : {}), ...(row.run_id ? { runId: row.run_id } : {}), ...(row.label ? { label: row.label } : {}), task: row.task, @@ -150,6 +152,7 @@ function bindTaskRecordBase(record: TaskRecord): Insertable { parent_flow_id: record.parentFlowId ?? null, parent_task_id: record.parentTaskId ?? null, agent_id: record.agentId ?? null, + requester_agent_id: record.requesterAgentId ?? null, run_id: record.runId ?? null, label: record.label ?? null, task: record.task, @@ -238,6 +241,7 @@ function upsertTaskRow(db: DatabaseSync, row: Insertable): void { parent_flow_id: (eb) => eb.ref("excluded.parent_flow_id"), parent_task_id: (eb) => eb.ref("excluded.parent_task_id"), agent_id: (eb) => eb.ref("excluded.agent_id"), + requester_agent_id: (eb) => eb.ref("excluded.requester_agent_id"), run_id: (eb) => eb.ref("excluded.run_id"), label: (eb) => eb.ref("excluded.label"), task: (eb) => eb.ref("excluded.task"), diff --git a/src/tasks/task-registry.store.test.ts b/src/tasks/task-registry.store.test.ts index ef728eeaa06..ae398dc0ddc 100644 --- a/src/tasks/task-registry.store.test.ts +++ b/src/tasks/task-registry.store.test.ts @@ -522,15 +522,17 @@ describe("task-registry store runtime", () => { }); }); - it("persists inferred child-session agent ids in sqlite task rows", async () => { + it("persists executor and requester agent ids in sqlite task rows", async () => { await withOpenClawTestState( { layout: "state-only", prefix: "openclaw-task-agent-id-" }, async () => { const created = createTaskRecord({ runtime: "subagent", - ownerKey: "agent:main:main", + requesterSessionKey: "global", + ownerKey: "global", scopeKind: "session", childSessionKey: "agent:worker:subagent:child", + requesterAgentId: "main", runId: "run-worker-subagent-sqlite", task: "Inspect worker state", status: "running", @@ -543,20 +545,22 @@ describe("task-registry store runtime", () => { database.db, db .selectFrom("task_runs") - .select(["agent_id", "child_session_key", "owner_key"]) + .select(["agent_id", "requester_agent_id", "child_session_key", "owner_key"]) .where("task_id", "=", created.taskId), ); expect(row).toEqual({ agent_id: "worker", + requester_agent_id: "main", child_session_key: "agent:worker:subagent:child", - owner_key: "agent:main:main", + owner_key: "global", }); resetTaskRegistryForTests({ persist: false }); expect(findTaskByRunId("run-worker-subagent-sqlite")).toMatchObject({ taskId: created.taskId, agentId: "worker", + requesterAgentId: "main", }); }, ); diff --git a/src/tasks/task-registry.ts b/src/tasks/task-registry.ts index 4e58108de06..559cf9caf2c 100644 --- a/src/tasks/task-registry.ts +++ b/src/tasks/task-registry.ts @@ -14,7 +14,7 @@ import { formatErrorMessage } from "../infra/errors.js"; import { requestHeartbeat } from "../infra/heartbeat-wake.js"; import { enqueueSystemEvent } from "../infra/system-events.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; -import { parseAgentSessionKey } from "../routing/session-key.js"; +import { normalizeAgentId, parseAgentSessionKey } from "../routing/session-key.js"; import { normalizeDeliveryContext } from "../utils/delivery-context.shared.js"; import { isDeliverableMessageChannel } from "../utils/message-channel.js"; import { isChildlessCodexNativeSubagentTask } from "./codex-native-subagent-task.js"; @@ -857,6 +857,7 @@ function mergeExistingTaskForCreate( parentFlowId?: string; parentTaskId?: string; agentId?: string; + requesterAgentId?: string; label?: string; task: string; preferMetadata?: boolean; @@ -897,6 +898,9 @@ function mergeExistingTaskForCreate( if (params.agentId?.trim() && !existing.agentId?.trim()) { patch.agentId = params.agentId.trim(); } + if (params.requesterAgentId?.trim() && !existing.requesterAgentId?.trim()) { + patch.requesterAgentId = params.requesterAgentId.trim(); + } const nextLabel = params.label?.trim(); if (params.preferMetadata) { if (nextLabel && (normalizeOptionalString(existing.label) ?? "") !== nextLabel) { @@ -941,6 +945,19 @@ function resolveTaskAgentId(params: { ); } +function resolveTaskRequesterAgentId(params: { + explicitRequesterAgentId?: string; + ownerKey: string; + requesterSessionKey: string; +}): string | undefined { + const explicitRequesterAgentId = normalizeOptionalString(params.explicitRequesterAgentId); + return ( + (explicitRequesterAgentId ? normalizeAgentId(explicitRequesterAgentId) : undefined) ?? + parseAgentSessionKey(params.ownerKey)?.agentId ?? + parseAgentSessionKey(params.requesterSessionKey)?.agentId + ); +} + function taskTerminalDeliveryIdempotencyKey(task: TaskRecord): string { const outcome = task.status === "succeeded" ? (task.terminalOutcome ?? "default") : "default"; return `task-terminal:${task.taskId}:${task.status}:${outcome}`; @@ -1689,6 +1706,7 @@ export function createTaskRecord(params: { parentFlowId?: string; parentTaskId?: string; agentId?: string; + requesterAgentId?: string; runId?: string; label?: string; task: string; @@ -1719,6 +1737,11 @@ export function createTaskRecord(params: { ownerKey, requesterSessionKey, }); + const requesterAgentId = resolveTaskRequesterAgentId({ + explicitRequesterAgentId: params.requesterAgentId, + ownerKey, + requesterSessionKey, + }); assertTaskOwner({ ownerKey, scopeKind, @@ -1769,6 +1792,7 @@ export function createTaskRecord(params: { parentFlowId: normalizeOptionalString(params.parentFlowId), parentTaskId: normalizeOptionalString(params.parentTaskId), agentId, + requesterAgentId, runId: normalizeOptionalString(params.runId), label: normalizeOptionalString(params.label), task: params.task, diff --git a/src/tasks/task-registry.types.ts b/src/tasks/task-registry.types.ts index fd69f8346b3..7d4cd0c7915 100644 --- a/src/tasks/task-registry.types.ts +++ b/src/tasks/task-registry.types.ts @@ -125,6 +125,9 @@ export type TaskRecord = { parentFlowId?: string; parentTaskId?: string; agentId?: string; + /** Agent store for requester transcripts whose session key is unscoped, such as `global`. + * Task authorization remains keyed by ownerKey. */ + requesterAgentId?: string; runId?: string; label?: string; task: string; diff --git a/src/tasks/task-status-access.ts b/src/tasks/task-status-access.ts index c827eac3005..22c25f4d55d 100644 --- a/src/tasks/task-status-access.ts +++ b/src/tasks/task-status-access.ts @@ -11,13 +11,17 @@ import type { TaskRecord } from "./task-registry.types.js"; /** Returns only the session lookup fields needed by task status commands. */ export function getTaskSessionLookupByIdForStatus( taskId: string, -): Pick | undefined { +): + | Pick + | undefined { const task = getTaskById(taskId); return task ? { requesterSessionKey: task.requesterSessionKey, + ownerKey: task.ownerKey, ...(task.runId ? { runId: task.runId } : {}), ...(task.agentId ? { agentId: task.agentId } : {}), + ...(task.requesterAgentId ? { requesterAgentId: task.requesterAgentId } : {}), } : undefined; }