mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-10 00:11:19 +00:00
fix(tasks): preserve requester agent attribution
This commit is contained in:
parent
77d6ad6f65
commit
1fef20c96b
22 changed files with 216 additions and 23 deletions
|
|
@ -360,7 +360,7 @@ A sweeper runs every **60 seconds** and handles four things:
|
|||
|
||||
</Accordion>
|
||||
<Accordion title="Tasks and sessions">
|
||||
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.
|
||||
</Accordion>
|
||||
<Accordion title="Tasks and agent runs">
|
||||
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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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 () => {
|
||||
|
|
|
|||
|
|
@ -1671,6 +1671,7 @@ export async function spawnSubagentDirect(
|
|||
task,
|
||||
taskName,
|
||||
agentId: targetAgentId,
|
||||
requesterAgentId,
|
||||
cleanup,
|
||||
label: label || undefined,
|
||||
model: resolvedModel,
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
1
src/state/openclaw-state-db.generated.d.ts
vendored
1
src/state/openclaw-state-db.generated.d.ts
vendored
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ export type DetachedTaskCreateParams = {
|
|||
childSessionKey?: string;
|
||||
parentTaskId?: string;
|
||||
agentId?: string;
|
||||
requesterAgentId?: string;
|
||||
runId?: string;
|
||||
label?: string;
|
||||
task: string;
|
||||
|
|
|
|||
|
|
@ -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<TaskRunsTable> {
|
|||
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<TaskRunsTable>): 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"),
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
});
|
||||
},
|
||||
);
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<TaskRecord, "requesterSessionKey" | "runId" | "agentId"> | undefined {
|
||||
):
|
||||
| Pick<TaskRecord, "requesterSessionKey" | "ownerKey" | "runId" | "agentId" | "requesterAgentId">
|
||||
| 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;
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue