mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-27 12:56:01 -06:00
refactor(tasks): split task registry internals (#113279)
* refactor(tasks): split task registry internals * test(tasks): align registry ownership boundary
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
741ed2d6a8
commit
b233faa057
@@ -1034,7 +1034,6 @@ src/tasks/task-flow-registry.ts
|
||||
src/tasks/task-registry.maintenance.ts
|
||||
src/tasks/task-registry.store.test.ts
|
||||
src/tasks/task-registry.test.ts
|
||||
src/tasks/task-registry.ts
|
||||
src/trajectory/export.test.ts
|
||||
src/trajectory/export.ts
|
||||
src/tui/embedded-backend.test.ts
|
||||
|
||||
@@ -20,7 +20,7 @@ const RAW_TASK_MUTATORS = [
|
||||
|
||||
const RAW_TASK_MUTATOR_ALLOWED_CALLERS = new Set([
|
||||
"tasks/task-executor.ts",
|
||||
"tasks/task-registry.ts",
|
||||
"tasks/task-registry-record-api.ts",
|
||||
"tasks/task-registry.maintenance.ts",
|
||||
]);
|
||||
|
||||
|
||||
@@ -0,0 +1,283 @@
|
||||
import type { OpenClawConfig } from "../config/types.openclaw.js";
|
||||
import { formatErrorMessage } from "../infra/errors.js";
|
||||
import { isBackgroundExecTask } from "./background-exec-task-contract.js";
|
||||
import { SUBAGENT_KILL_TASK_ERROR } from "./detached-task-runtime-contract.js";
|
||||
import { isChildlessNativeSubagentTask } from "./native-subagent-task.js";
|
||||
import { isProvisionalSubagentKillTask } from "./task-cancellation-state.js";
|
||||
import { isTerminalTaskStatus } from "./task-executor-policy.js";
|
||||
import { ensureLinkedTaskFlowRegistryReady } from "./task-registry-common.js";
|
||||
import { maybeDeliverTaskTerminalUpdate } from "./task-registry-delivery.js";
|
||||
import { updateTask } from "./task-registry-mutation.js";
|
||||
import { finalizeTaskRunByRunId, updateTaskStateByRunId } from "./task-registry-record-api.js";
|
||||
import { cloneTaskRecord } from "./task-registry-records.js";
|
||||
import {
|
||||
ensureTaskRegistryReady,
|
||||
getTasksByRunScope,
|
||||
loadTaskRegistryControlRuntime,
|
||||
tasks,
|
||||
} from "./task-registry-state.js";
|
||||
import type { TaskRecord } from "./task-registry.types.js";
|
||||
|
||||
function ensureTaskCancellationReady(task: TaskRecord): void {
|
||||
const runId = task.runId?.trim();
|
||||
const linkedTasks =
|
||||
runId && (task.runtime === "acp" || task.runtime === "subagent")
|
||||
? getTasksByRunScope({
|
||||
runId,
|
||||
runtime: task.runtime,
|
||||
sessionKey: task.childSessionKey,
|
||||
})
|
||||
: [task];
|
||||
for (const linkedTask of linkedTasks.length > 0 ? linkedTasks : [task]) {
|
||||
ensureLinkedTaskFlowRegistryReady(linkedTask);
|
||||
}
|
||||
}
|
||||
|
||||
export async function cancelTaskById(params: {
|
||||
cfg: OpenClawConfig;
|
||||
taskId: string;
|
||||
reason?: string;
|
||||
}): Promise<{ found: boolean; cancelled: boolean; reason?: string; task?: TaskRecord }> {
|
||||
ensureTaskRegistryReady();
|
||||
const task = tasks.get(params.taskId.trim());
|
||||
if (!task) {
|
||||
return { found: false, cancelled: false, reason: "Task not found." };
|
||||
}
|
||||
const requestedReason = params.reason?.trim();
|
||||
const cancellationError =
|
||||
requestedReason && requestedReason !== SUBAGENT_KILL_TASK_ERROR
|
||||
? requestedReason
|
||||
: "Cancelled by operator.";
|
||||
let isProvisionalSubagentKill =
|
||||
task.runtime === "subagent" &&
|
||||
task.status === "cancelled" &&
|
||||
task.error === SUBAGENT_KILL_TASK_ERROR;
|
||||
if (
|
||||
!isProvisionalSubagentKill &&
|
||||
(task.status === "succeeded" ||
|
||||
task.status === "failed" ||
|
||||
task.status === "timed_out" ||
|
||||
task.status === "lost" ||
|
||||
task.status === "cancelled")
|
||||
) {
|
||||
return {
|
||||
found: true,
|
||||
cancelled: false,
|
||||
reason: "Task is already terminal.",
|
||||
task: cloneTaskRecord(task),
|
||||
};
|
||||
}
|
||||
const childSessionKey = task.childSessionKey?.trim();
|
||||
try {
|
||||
ensureTaskCancellationReady(task);
|
||||
// A direct kill is only a provisional terminal projection. Re-read the
|
||||
// owning subagent run before promotion so its canonical completion can win.
|
||||
if (isBackgroundExecTask(task)) {
|
||||
const processSessionId = task.sourceId?.trim();
|
||||
const { cancelBackgroundExecSession } = await loadTaskRegistryControlRuntime();
|
||||
if (!processSessionId || !cancelBackgroundExecSession?.(processSessionId)) {
|
||||
return {
|
||||
found: true,
|
||||
cancelled: false,
|
||||
reason: "Background command has no active cancellation handle.",
|
||||
task: cloneTaskRecord(task),
|
||||
};
|
||||
}
|
||||
} else if (task.runtime !== "cli") {
|
||||
if (task.runtime === "cron") {
|
||||
const { cancelActiveCronTaskRun } = await loadTaskRegistryControlRuntime();
|
||||
if (
|
||||
!cancelActiveCronTaskRun({
|
||||
runId: task.runId,
|
||||
reason: params.reason?.trim() || "Cancelled by operator.",
|
||||
})
|
||||
) {
|
||||
if (childSessionKey) {
|
||||
return {
|
||||
found: true,
|
||||
cancelled: false,
|
||||
reason: "Cron task has no active cancellation handle.",
|
||||
task: cloneTaskRecord(task),
|
||||
};
|
||||
}
|
||||
// Childless cron rows are stale legacy ledger records; with no live
|
||||
// runner handle and no child session to cancel, clear the task row.
|
||||
}
|
||||
} else if (!childSessionKey) {
|
||||
if (!isChildlessNativeSubagentTask(task)) {
|
||||
return {
|
||||
found: true,
|
||||
cancelled: false,
|
||||
reason: "Task has no cancellable child session.",
|
||||
task: cloneTaskRecord(task),
|
||||
};
|
||||
}
|
||||
}
|
||||
if (task.runtime === "cron") {
|
||||
// The live cron service owns the abort signal; registry finalization below
|
||||
// keeps CLI/Gateway callers aligned while the run unwinds.
|
||||
} else if (!childSessionKey) {
|
||||
// Codex native subagents are mirrored from the Codex app server and do
|
||||
// not have OpenClaw child sessions to terminate. Cancellation clears
|
||||
// the stale task-registry record only.
|
||||
} else if (task.runtime === "acp") {
|
||||
const { getAcpSessionManager } = await loadTaskRegistryControlRuntime();
|
||||
await getAcpSessionManager().cancelSession({
|
||||
cfg: params.cfg,
|
||||
sessionKey: childSessionKey,
|
||||
reason: params.reason?.trim() || "task-cancel",
|
||||
});
|
||||
} else if (task.runtime === "subagent") {
|
||||
const { killSubagentRunAdmin } = await loadTaskRegistryControlRuntime();
|
||||
const result = await killSubagentRunAdmin({
|
||||
cfg: params.cfg,
|
||||
sessionKey: childSessionKey,
|
||||
});
|
||||
const current = tasks.get(task.taskId);
|
||||
if (current?.status === "cancelled" && current.error === SUBAGENT_KILL_TASK_ERROR) {
|
||||
isProvisionalSubagentKill = true;
|
||||
}
|
||||
if (current?.status === "succeeded") {
|
||||
return {
|
||||
found: true,
|
||||
cancelled: false,
|
||||
reason: "Subagent completed while cancellation was in progress.",
|
||||
task: cloneTaskRecord(current),
|
||||
};
|
||||
}
|
||||
if (current && isTerminalTaskStatus(current.status) && current.status !== "cancelled") {
|
||||
return {
|
||||
found: true,
|
||||
cancelled: false,
|
||||
reason: `Subagent became ${current.status} while cancellation was in progress.`,
|
||||
task: cloneTaskRecord(current),
|
||||
};
|
||||
}
|
||||
if (current?.status === "cancelled" && !isProvisionalSubagentKill) {
|
||||
return {
|
||||
found: true,
|
||||
cancelled: false,
|
||||
reason: "Subagent was cancelled while cancellation was in progress.",
|
||||
task: cloneTaskRecord(current),
|
||||
};
|
||||
}
|
||||
if (result.found && result.targetState?.state === "terminal") {
|
||||
// A subagent run becomes terminal before its task projection settles.
|
||||
// Reconcile the original task scope: steer/orphan recovery may have
|
||||
// replaced the registry run ID without remapping durable task rows.
|
||||
const taskRunId = task.runId?.trim() || result.runId;
|
||||
const reconciledTasks = finalizeTaskRunByRunId({
|
||||
runId: taskRunId,
|
||||
runtime: "subagent",
|
||||
sessionKey: childSessionKey,
|
||||
...result.targetState.task,
|
||||
});
|
||||
const reconciled = reconciledTasks.find((candidate) => candidate.taskId === task.taskId);
|
||||
if (!reconciled) {
|
||||
return {
|
||||
found: true,
|
||||
cancelled: false,
|
||||
reason: "Subagent became terminal, but task state reconciliation failed to persist.",
|
||||
task: cloneTaskRecord(tasks.get(task.taskId) ?? task),
|
||||
};
|
||||
}
|
||||
if (
|
||||
result.targetState.task.status === "cancelled" &&
|
||||
result.targetState.task.error === SUBAGENT_KILL_TASK_ERROR
|
||||
) {
|
||||
isProvisionalSubagentKill = true;
|
||||
} else {
|
||||
const reason =
|
||||
result.targetState.task.status === "succeeded"
|
||||
? "Subagent completed while cancellation was in progress."
|
||||
: `Subagent became ${result.targetState.task.status} while cancellation was in progress.`;
|
||||
return {
|
||||
found: true,
|
||||
cancelled: false,
|
||||
reason,
|
||||
task: cloneTaskRecord(reconciled),
|
||||
};
|
||||
}
|
||||
}
|
||||
if (result.found && result.targetState?.state === "finalizing") {
|
||||
return {
|
||||
found: true,
|
||||
cancelled: false,
|
||||
reason: "Subagent completion is still being finalized.",
|
||||
task: cloneTaskRecord(current ?? task),
|
||||
};
|
||||
}
|
||||
if ((!result.found || !result.killed) && !isProvisionalSubagentKill) {
|
||||
return {
|
||||
found: true,
|
||||
cancelled: false,
|
||||
reason: result.found ? "Subagent was not running." : "Subagent task not found.",
|
||||
task: cloneTaskRecord(current ?? task),
|
||||
};
|
||||
}
|
||||
} else {
|
||||
return {
|
||||
found: true,
|
||||
cancelled: false,
|
||||
reason: "Task runtime does not support cancellation yet.",
|
||||
task: cloneTaskRecord(task),
|
||||
};
|
||||
}
|
||||
}
|
||||
const eventAt = Date.now();
|
||||
const current = tasks.get(task.taskId) ?? task;
|
||||
const endedAt = isProvisionalSubagentKill ? (current.endedAt ?? eventAt) : eventAt;
|
||||
const updated =
|
||||
(task.runtime === "acp" || task.runtime === "subagent") && task.runId?.trim()
|
||||
? (updateTaskStateByRunId({
|
||||
runId: task.runId,
|
||||
runtime: task.runtime,
|
||||
sessionKey: childSessionKey,
|
||||
status: "cancelled",
|
||||
endedAt,
|
||||
lastEventAt: eventAt,
|
||||
error: cancellationError,
|
||||
}).find((record) => record.taskId === task.taskId) ?? null)
|
||||
: updateTask(task.taskId, {
|
||||
status: "cancelled",
|
||||
endedAt,
|
||||
lastEventAt: eventAt,
|
||||
error: cancellationError,
|
||||
});
|
||||
if (!updated) {
|
||||
return {
|
||||
found: true,
|
||||
cancelled: false,
|
||||
reason: "Task persistence failed.",
|
||||
task: cloneTaskRecord(task),
|
||||
};
|
||||
}
|
||||
if (updated) {
|
||||
void maybeDeliverTaskTerminalUpdate(updated.taskId);
|
||||
}
|
||||
return {
|
||||
found: true,
|
||||
cancelled: true,
|
||||
task: updated ?? cloneTaskRecord(task),
|
||||
};
|
||||
} catch (error) {
|
||||
return {
|
||||
found: true,
|
||||
cancelled: false,
|
||||
reason: formatErrorMessage(error),
|
||||
task: cloneTaskRecord(task),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
export function assertTaskCancellationReadyById(taskId: string): TaskRecord | null {
|
||||
ensureTaskRegistryReady();
|
||||
const task = tasks.get(taskId.trim());
|
||||
if (!task) {
|
||||
return null;
|
||||
}
|
||||
if (!isTerminalTaskStatus(task.status) || isProvisionalSubagentKillTask(task)) {
|
||||
ensureTaskCancellationReady(task);
|
||||
}
|
||||
return cloneTaskRecord(task);
|
||||
}
|
||||
@@ -0,0 +1,334 @@
|
||||
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
|
||||
import {
|
||||
buildAgentRunTerminalOutcome,
|
||||
type AgentRunTerminalOutcome,
|
||||
} from "../agents/agent-run-terminal-outcome.js";
|
||||
import { SUBAGENT_KILL_TASK_ERROR } from "./detached-task-runtime-contract.js";
|
||||
import { isTerminalTaskStatus } from "./task-executor-policy.js";
|
||||
import type { TaskFlowRecord } from "./task-flow-registry.types.js";
|
||||
import { ensureTaskFlowRegistryReady, getTaskFlowById } from "./task-flow-runtime-internal.js";
|
||||
import type {
|
||||
TaskDeliveryState,
|
||||
TaskDeliveryStatus,
|
||||
TaskEventKind,
|
||||
TaskEventRecord,
|
||||
TaskNotifyPolicy,
|
||||
TaskRecord,
|
||||
TaskRuntime,
|
||||
TaskScopeKind,
|
||||
TaskStatus,
|
||||
TaskTerminalOutcome,
|
||||
} from "./task-registry.types.js";
|
||||
|
||||
export type TaskDeliveryOwner = {
|
||||
sessionKey?: string;
|
||||
requesterOrigin?: TaskDeliveryState["requesterOrigin"];
|
||||
flowId?: string;
|
||||
};
|
||||
|
||||
type ParentFlowLinkErrorCode =
|
||||
| "scope_kind_not_session"
|
||||
| "parent_flow_not_found"
|
||||
| "owner_key_mismatch"
|
||||
| "cancel_requested"
|
||||
| "terminal";
|
||||
|
||||
class ParentFlowLinkError extends Error {
|
||||
constructor(
|
||||
public readonly code: ParentFlowLinkErrorCode,
|
||||
message: string,
|
||||
public readonly details?: {
|
||||
flowId?: string;
|
||||
status?: TaskFlowRecord["status"];
|
||||
},
|
||||
) {
|
||||
super(message);
|
||||
this.name = "ParentFlowLinkError";
|
||||
}
|
||||
}
|
||||
|
||||
export function isParentFlowLinkError(error: unknown): error is ParentFlowLinkError {
|
||||
return error instanceof ParentFlowLinkError;
|
||||
}
|
||||
|
||||
export function isActiveTaskStatus(status: TaskStatus): boolean {
|
||||
return status === "queued" || status === "running";
|
||||
}
|
||||
|
||||
export function isTerminalFlowStatus(status: TaskFlowRecord["status"]): boolean {
|
||||
return (
|
||||
status === "succeeded" || status === "failed" || status === "cancelled" || status === "lost"
|
||||
);
|
||||
}
|
||||
|
||||
export function assertTaskOwner(params: { ownerKey: string; scopeKind: TaskScopeKind }) {
|
||||
const ownerKey = params.ownerKey.trim();
|
||||
if (!ownerKey && params.scopeKind !== "system") {
|
||||
throw new Error("Task ownerKey is required.");
|
||||
}
|
||||
}
|
||||
|
||||
export function assertParentFlowLinkAllowed(params: {
|
||||
ownerKey: string;
|
||||
scopeKind: TaskScopeKind;
|
||||
parentFlowId?: string;
|
||||
}) {
|
||||
const flowId = params.parentFlowId?.trim();
|
||||
if (!flowId) {
|
||||
return;
|
||||
}
|
||||
if (params.scopeKind !== "session") {
|
||||
throw new ParentFlowLinkError(
|
||||
"scope_kind_not_session",
|
||||
"Only session-scoped tasks can link to flows.",
|
||||
{ flowId },
|
||||
);
|
||||
}
|
||||
const flow = getTaskFlowById(flowId);
|
||||
if (!flow) {
|
||||
throw new ParentFlowLinkError("parent_flow_not_found", `Parent flow not found: ${flowId}`, {
|
||||
flowId,
|
||||
});
|
||||
}
|
||||
if (normalizeOptionalString(flow.ownerKey) !== normalizeOptionalString(params.ownerKey)) {
|
||||
throw new ParentFlowLinkError(
|
||||
"owner_key_mismatch",
|
||||
"Task ownerKey must match parent flow ownerKey.",
|
||||
{ flowId },
|
||||
);
|
||||
}
|
||||
if (flow.cancelRequestedAt != null) {
|
||||
throw new ParentFlowLinkError(
|
||||
"cancel_requested",
|
||||
"Parent flow cancellation has already been requested.",
|
||||
{ flowId, status: flow.status },
|
||||
);
|
||||
}
|
||||
if (isTerminalFlowStatus(flow.status)) {
|
||||
throw new ParentFlowLinkError("terminal", `Parent flow is already ${flow.status}.`, {
|
||||
flowId,
|
||||
status: flow.status,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export function ensureLinkedTaskFlowRegistryReady(task: Pick<TaskRecord, "parentFlowId">): void {
|
||||
if (task.parentFlowId?.trim()) {
|
||||
ensureTaskFlowRegistryReady();
|
||||
}
|
||||
}
|
||||
|
||||
export function ensureDeliveryStatus(params: {
|
||||
ownerKey: string;
|
||||
scopeKind: TaskScopeKind;
|
||||
}): TaskDeliveryStatus {
|
||||
if (params.scopeKind === "system") {
|
||||
return "not_applicable";
|
||||
}
|
||||
return params.ownerKey.trim() ? "pending" : "parent_missing";
|
||||
}
|
||||
|
||||
export function ensureNotifyPolicy(params: {
|
||||
notifyPolicy?: TaskNotifyPolicy;
|
||||
deliveryStatus?: TaskDeliveryStatus;
|
||||
ownerKey: string;
|
||||
scopeKind: TaskScopeKind;
|
||||
}): TaskNotifyPolicy {
|
||||
if (params.notifyPolicy) {
|
||||
return params.notifyPolicy;
|
||||
}
|
||||
const deliveryStatus =
|
||||
params.deliveryStatus ??
|
||||
ensureDeliveryStatus({
|
||||
ownerKey: params.ownerKey,
|
||||
scopeKind: params.scopeKind,
|
||||
});
|
||||
return deliveryStatus === "not_applicable" ? "silent" : "done_only";
|
||||
}
|
||||
|
||||
export function resolveTaskScopeKind(params: {
|
||||
scopeKind?: TaskScopeKind;
|
||||
requesterSessionKey: string;
|
||||
}): TaskScopeKind {
|
||||
if (params.scopeKind) {
|
||||
return params.scopeKind;
|
||||
}
|
||||
return params.requesterSessionKey.trim() ? "session" : "system";
|
||||
}
|
||||
|
||||
export function resolveTaskRequesterSessionKey(params: {
|
||||
requesterSessionKey?: string;
|
||||
ownerKey?: string;
|
||||
scopeKind?: TaskScopeKind;
|
||||
}): string {
|
||||
const requesterSessionKey = params.requesterSessionKey?.trim();
|
||||
if (requesterSessionKey) {
|
||||
return requesterSessionKey;
|
||||
}
|
||||
if (params.scopeKind === "system") {
|
||||
return "";
|
||||
}
|
||||
return params.ownerKey?.trim() ?? "";
|
||||
}
|
||||
|
||||
export function resolveTaskOwnerKey(params: {
|
||||
requesterSessionKey: string;
|
||||
ownerKey?: string;
|
||||
}): string {
|
||||
return params.ownerKey?.trim() || params.requesterSessionKey.trim();
|
||||
}
|
||||
|
||||
export function normalizeTaskSummary(value: string | null | undefined): string | undefined {
|
||||
const normalized = value?.replace(/\s+/g, " ").trim();
|
||||
return normalized || undefined;
|
||||
}
|
||||
|
||||
export function normalizeTaskStatus(value: TaskStatus | null | undefined): TaskStatus {
|
||||
return value === "running" ||
|
||||
value === "queued" ||
|
||||
value === "succeeded" ||
|
||||
value === "failed" ||
|
||||
value === "timed_out" ||
|
||||
value === "cancelled" ||
|
||||
value === "lost"
|
||||
? value
|
||||
: "queued";
|
||||
}
|
||||
|
||||
function normalizeTaskTerminalOutcome(
|
||||
value: TaskTerminalOutcome | null | undefined,
|
||||
): TaskTerminalOutcome | undefined {
|
||||
return value === "succeeded" || value === "blocked" ? value : undefined;
|
||||
}
|
||||
|
||||
export function shouldApplyRunScopedStatusUpdate(params: {
|
||||
currentStatus: TaskStatus;
|
||||
currentRuntime: TaskRuntime;
|
||||
currentChildSessionKey?: string;
|
||||
currentError?: string;
|
||||
currentEndedAt?: number;
|
||||
nextStatus: TaskStatus;
|
||||
nextError?: string;
|
||||
nextEndedAt?: number;
|
||||
}): boolean {
|
||||
if (
|
||||
params.currentRuntime === "subagent" &&
|
||||
params.nextStatus === "cancelled" &&
|
||||
params.nextError === SUBAGENT_KILL_TASK_ERROR &&
|
||||
isTerminalTaskStatus(params.currentStatus) &&
|
||||
!(params.currentStatus === "cancelled" && params.currentError === SUBAGENT_KILL_TASK_ERROR)
|
||||
) {
|
||||
// The kill marker is provisional. It may refresh only its own tombstone;
|
||||
// canonical completion or operator cancellation already won this race.
|
||||
return false;
|
||||
}
|
||||
if (params.currentStatus === params.nextStatus) {
|
||||
return true;
|
||||
}
|
||||
if (!isTerminalTaskStatus(params.currentStatus)) {
|
||||
return true;
|
||||
}
|
||||
if (!isTerminalTaskStatus(params.nextStatus)) {
|
||||
return false;
|
||||
}
|
||||
// Direct subagent termination is provisional. An operator cancellation is
|
||||
// sticky only against outcomes that completed at or after cancellation.
|
||||
if (
|
||||
params.currentStatus === "cancelled" &&
|
||||
(params.nextStatus === "succeeded" ||
|
||||
params.nextStatus === "failed" ||
|
||||
params.nextStatus === "timed_out")
|
||||
) {
|
||||
const canonicalOutcomePredatesCancellation =
|
||||
params.currentRuntime === "subagent" &&
|
||||
params.currentEndedAt !== undefined &&
|
||||
params.nextEndedAt !== undefined &&
|
||||
params.nextEndedAt < params.currentEndedAt;
|
||||
return (
|
||||
canonicalOutcomePredatesCancellation ||
|
||||
(params.currentRuntime === "subagent" &&
|
||||
Boolean(params.currentChildSessionKey?.trim()) &&
|
||||
params.currentError === SUBAGENT_KILL_TASK_ERROR)
|
||||
);
|
||||
}
|
||||
return params.currentStatus === "succeeded" && params.nextStatus !== "lost";
|
||||
}
|
||||
|
||||
export function resolveTaskTerminalOutcome(params: {
|
||||
status: TaskStatus;
|
||||
terminalOutcome?: TaskTerminalOutcome | null;
|
||||
}): TaskTerminalOutcome | undefined {
|
||||
const normalized = normalizeTaskTerminalOutcome(params.terminalOutcome);
|
||||
if (normalized) {
|
||||
return normalized;
|
||||
}
|
||||
return params.status === "succeeded" ? "succeeded" : undefined;
|
||||
}
|
||||
|
||||
export function mapAgentRunTerminalOutcomeToTaskStatus(
|
||||
outcome: AgentRunTerminalOutcome,
|
||||
): Extract<TaskStatus, "succeeded" | "failed" | "timed_out" | "cancelled"> {
|
||||
switch (outcome.reason) {
|
||||
case "completed":
|
||||
return "succeeded";
|
||||
case "hard_timeout":
|
||||
case "timed_out":
|
||||
return "timed_out";
|
||||
case "cancelled":
|
||||
case "aborted":
|
||||
return "cancelled";
|
||||
case "blocked":
|
||||
case "abandoned":
|
||||
case "failed":
|
||||
return "failed";
|
||||
default:
|
||||
return outcome.reason satisfies never;
|
||||
}
|
||||
}
|
||||
|
||||
export function resolveTaskLifecycleTerminalError(params: {
|
||||
runtime: TaskRuntime;
|
||||
status: TaskStatus;
|
||||
error?: string;
|
||||
}): string | undefined {
|
||||
// A runner abort can race either an accepted task cancellation or a real
|
||||
// completion. Keep it provisional until the task-control owner decides.
|
||||
return params.runtime === "subagent" && params.status === "cancelled"
|
||||
? SUBAGENT_KILL_TASK_ERROR
|
||||
: params.error;
|
||||
}
|
||||
|
||||
export function buildTaskLifecycleTerminalOutcome(params: {
|
||||
phase: "end" | "error";
|
||||
data?: Record<string, unknown>;
|
||||
startedAt?: number;
|
||||
endedAt?: number;
|
||||
}): AgentRunTerminalOutcome {
|
||||
const status =
|
||||
params.phase === "error" ? "error" : params.data?.aborted === true ? "timeout" : "ok";
|
||||
// Lifecycle events carry runner/provider terminal facts. Keep the precedence
|
||||
// centralized so task projections match agent.wait and gateway snapshots.
|
||||
return buildAgentRunTerminalOutcome({
|
||||
status,
|
||||
error: params.data?.error,
|
||||
stopReason: params.data?.stopReason,
|
||||
livenessState: params.data?.livenessState,
|
||||
timeoutPhase: params.data?.timeoutPhase,
|
||||
providerStarted: params.data?.providerStarted,
|
||||
startedAt: params.startedAt,
|
||||
endedAt: params.endedAt,
|
||||
});
|
||||
}
|
||||
|
||||
export function appendTaskEvent(event: {
|
||||
at: number;
|
||||
kind: TaskEventKind;
|
||||
summary?: string | null;
|
||||
}): TaskEventRecord {
|
||||
const summary = normalizeTaskSummary(event.summary);
|
||||
return {
|
||||
at: event.at,
|
||||
kind: event.kind,
|
||||
...(summary ? { summary } : {}),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,194 @@
|
||||
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
|
||||
import { normalizeAgentId, parseAgentSessionKey } from "../routing/session-key.js";
|
||||
import { normalizeDeliveryContext } from "../utils/delivery-context.shared.js";
|
||||
import {
|
||||
assertParentFlowLinkAllowed,
|
||||
ensureLinkedTaskFlowRegistryReady,
|
||||
ensureNotifyPolicy,
|
||||
} from "./task-registry-common.js";
|
||||
import { updateTask, upsertTaskDeliveryState } from "./task-registry-mutation.js";
|
||||
import { cloneTaskRecord } from "./task-registry-records.js";
|
||||
import {
|
||||
getTasksByRunId,
|
||||
pickPreferredRunIdTask,
|
||||
taskDeliveryStates,
|
||||
} from "./task-registry-state.js";
|
||||
import type {
|
||||
JsonValue,
|
||||
TaskDeliveryState,
|
||||
TaskDeliveryStatus,
|
||||
TaskNotifyPolicy,
|
||||
TaskRecord,
|
||||
TaskRuntime,
|
||||
TaskScopeKind,
|
||||
} from "./task-registry.types.js";
|
||||
|
||||
export function findExistingTaskForCreate(params: {
|
||||
runtime: TaskRuntime;
|
||||
ownerKey: string;
|
||||
scopeKind: TaskScopeKind;
|
||||
childSessionKey?: string;
|
||||
parentFlowId?: string;
|
||||
runId?: string;
|
||||
label?: string;
|
||||
task: string;
|
||||
}): TaskRecord | undefined {
|
||||
const runId = params.runId?.trim();
|
||||
const runScopeMatches = runId
|
||||
? getTasksByRunId(runId).filter((task) => {
|
||||
if (
|
||||
task.runtime !== params.runtime ||
|
||||
task.scopeKind !== params.scopeKind ||
|
||||
(normalizeOptionalString(task.ownerKey) ?? "") !==
|
||||
(normalizeOptionalString(params.ownerKey) ?? "") ||
|
||||
(normalizeOptionalString(task.childSessionKey) ?? "") !==
|
||||
(normalizeOptionalString(params.childSessionKey) ?? "")
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
if (params.runtime === "acp") {
|
||||
// ACP one-task flow ids can be derived after creation; they must not
|
||||
// split one logical ACP run into duplicate task rows.
|
||||
return true;
|
||||
}
|
||||
return (
|
||||
(normalizeOptionalString(task.parentFlowId) ?? "") ===
|
||||
(normalizeOptionalString(params.parentFlowId) ?? "")
|
||||
);
|
||||
})
|
||||
: [];
|
||||
const exact = runId
|
||||
? runScopeMatches.find(
|
||||
(task) =>
|
||||
(normalizeOptionalString(task.label) ?? "") ===
|
||||
(normalizeOptionalString(params.label) ?? "") &&
|
||||
(normalizeOptionalString(task.task) ?? "") ===
|
||||
(normalizeOptionalString(params.task) ?? ""),
|
||||
)
|
||||
: undefined;
|
||||
if (exact) {
|
||||
return exact;
|
||||
}
|
||||
if (!runId || params.runtime !== "acp") {
|
||||
return undefined;
|
||||
}
|
||||
if (runScopeMatches.length === 0) {
|
||||
return undefined;
|
||||
}
|
||||
return pickPreferredRunIdTask(runScopeMatches);
|
||||
}
|
||||
|
||||
export function mergeExistingTaskForCreate(
|
||||
existing: TaskRecord,
|
||||
params: {
|
||||
taskKind?: string;
|
||||
requesterOrigin?: TaskDeliveryState["requesterOrigin"];
|
||||
sourceId?: string;
|
||||
parentFlowId?: string;
|
||||
parentTaskId?: string;
|
||||
agentId?: string;
|
||||
requesterAgentId?: string;
|
||||
label?: string;
|
||||
task: string;
|
||||
preferMetadata?: boolean;
|
||||
deliveryStatus?: TaskDeliveryStatus;
|
||||
notifyPolicy?: TaskNotifyPolicy;
|
||||
detail?: JsonValue;
|
||||
},
|
||||
): TaskRecord | null {
|
||||
ensureLinkedTaskFlowRegistryReady(existing);
|
||||
const patch: Partial<TaskRecord> = {};
|
||||
const requesterOrigin = normalizeDeliveryContext(params.requesterOrigin);
|
||||
const currentDeliveryState = taskDeliveryStates.get(existing.taskId);
|
||||
if (requesterOrigin && !currentDeliveryState?.requesterOrigin) {
|
||||
const deliveryState = upsertTaskDeliveryState({
|
||||
taskId: existing.taskId,
|
||||
requesterOrigin,
|
||||
lastNotifiedEventAt: currentDeliveryState?.lastNotifiedEventAt,
|
||||
});
|
||||
if (!deliveryState.requesterOrigin) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
if (params.sourceId?.trim() && !existing.sourceId?.trim()) {
|
||||
patch.sourceId = params.sourceId.trim();
|
||||
}
|
||||
if (params.taskKind?.trim() && !existing.taskKind?.trim()) {
|
||||
patch.taskKind = params.taskKind.trim();
|
||||
}
|
||||
if (params.parentFlowId?.trim() && !existing.parentFlowId?.trim()) {
|
||||
assertParentFlowLinkAllowed({
|
||||
ownerKey: existing.ownerKey,
|
||||
scopeKind: existing.scopeKind,
|
||||
parentFlowId: params.parentFlowId,
|
||||
});
|
||||
patch.parentFlowId = params.parentFlowId.trim();
|
||||
}
|
||||
if (params.parentTaskId?.trim() && !existing.parentTaskId?.trim()) {
|
||||
patch.parentTaskId = params.parentTaskId.trim();
|
||||
}
|
||||
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) {
|
||||
patch.label = nextLabel;
|
||||
}
|
||||
const nextTask = params.task.trim();
|
||||
if (nextTask && (normalizeOptionalString(existing.task) ?? "") !== nextTask) {
|
||||
patch.task = nextTask;
|
||||
}
|
||||
} else if (nextLabel && !existing.label?.trim()) {
|
||||
patch.label = nextLabel;
|
||||
}
|
||||
if (params.deliveryStatus === "pending" && existing.deliveryStatus !== "delivered") {
|
||||
patch.deliveryStatus = "pending";
|
||||
}
|
||||
const notifyPolicy = ensureNotifyPolicy({
|
||||
notifyPolicy: params.notifyPolicy,
|
||||
deliveryStatus: params.deliveryStatus,
|
||||
ownerKey: existing.ownerKey,
|
||||
scopeKind: existing.scopeKind,
|
||||
});
|
||||
if (notifyPolicy !== existing.notifyPolicy && existing.notifyPolicy === "silent") {
|
||||
patch.notifyPolicy = notifyPolicy;
|
||||
}
|
||||
if (params.detail !== undefined) {
|
||||
patch.detail = params.detail;
|
||||
}
|
||||
if (Object.keys(patch).length === 0) {
|
||||
return cloneTaskRecord(existing);
|
||||
}
|
||||
return updateTask(existing.taskId, patch);
|
||||
}
|
||||
|
||||
export function resolveTaskAgentId(params: {
|
||||
explicitAgentId?: string;
|
||||
childSessionKey?: string;
|
||||
ownerKey: string;
|
||||
requesterSessionKey: string;
|
||||
}): string | undefined {
|
||||
return (
|
||||
normalizeOptionalString(params.explicitAgentId) ??
|
||||
parseAgentSessionKey(params.childSessionKey)?.agentId ??
|
||||
parseAgentSessionKey(params.ownerKey)?.agentId ??
|
||||
parseAgentSessionKey(params.requesterSessionKey)?.agentId
|
||||
);
|
||||
}
|
||||
|
||||
export 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
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,450 @@
|
||||
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
|
||||
import { shouldRouteCompletionThroughRequesterSession } from "../auto-reply/reply/completion-delivery-policy.js";
|
||||
import { requestHeartbeat } from "../infra/heartbeat-wake.js";
|
||||
import { enqueueSystemEvent } from "../infra/system-events.js";
|
||||
import {
|
||||
isGatewayRestartDraining,
|
||||
runWithGatewayIndependentRootWorkAdmission,
|
||||
} from "../process/gateway-work-admission.js";
|
||||
import { parseAgentSessionKey } from "../routing/session-key.js";
|
||||
import { normalizeDeliveryContext } from "../utils/delivery-context.shared.js";
|
||||
import { isDeliverableMessageChannel } from "../utils/message-channel.js";
|
||||
import {
|
||||
formatTaskBlockedFollowupMessage,
|
||||
formatTaskStateChangeMessage,
|
||||
formatTaskTerminalMessage,
|
||||
shouldAutoDeliverTaskStateChange,
|
||||
shouldAutoDeliverTaskTerminalUpdate,
|
||||
shouldSuppressDuplicateTerminalDelivery,
|
||||
shouldUseParentReviewTaskTerminalMessage,
|
||||
} from "./task-executor-policy.js";
|
||||
import { getTaskFlowById } from "./task-flow-runtime-internal.js";
|
||||
import type { TaskDeliveryOwner } from "./task-registry-common.js";
|
||||
import {
|
||||
getTaskDeliveryState,
|
||||
updateTask,
|
||||
upsertTaskDeliveryState,
|
||||
} from "./task-registry-mutation.js";
|
||||
import { cloneTaskRecord } from "./task-registry-records.js";
|
||||
import {
|
||||
ensureTaskRegistryReady,
|
||||
getPeerTasksForDelivery,
|
||||
loadTaskRegistryDeliveryRuntime,
|
||||
log,
|
||||
pickPreferredRunIdTask,
|
||||
taskDeliveryStates,
|
||||
tasks,
|
||||
tasksWithPendingDelivery,
|
||||
} from "./task-registry-state.js";
|
||||
import type {
|
||||
TaskDeliveryState,
|
||||
TaskDeliveryStatus,
|
||||
TaskEventRecord,
|
||||
TaskRecord,
|
||||
} from "./task-registry.types.js";
|
||||
|
||||
function taskTerminalDeliveryIdempotencyKey(task: TaskRecord): string {
|
||||
const outcome = task.status === "succeeded" ? (task.terminalOutcome ?? "default") : "default";
|
||||
return `task-terminal:${task.taskId}:${task.status}:${outcome}`;
|
||||
}
|
||||
|
||||
function resolveTaskStateChangeIdempotencyKey(params: {
|
||||
task: TaskRecord;
|
||||
latestEvent: TaskEventRecord;
|
||||
owner: TaskDeliveryOwner;
|
||||
}): string {
|
||||
if (params.owner.flowId) {
|
||||
return `flow-event:${params.owner.flowId}:${params.task.taskId}:${params.latestEvent.at}:${params.latestEvent.kind}`;
|
||||
}
|
||||
return `task-event:${params.task.taskId}:${params.latestEvent.at}:${params.latestEvent.kind}`;
|
||||
}
|
||||
|
||||
function resolveTaskTerminalIdempotencyKey(task: TaskRecord): string {
|
||||
const owner = resolveTaskDeliveryOwner(task);
|
||||
if (owner.flowId) {
|
||||
const outcome = task.status === "succeeded" ? (task.terminalOutcome ?? "default") : "default";
|
||||
return `flow-terminal:${owner.flowId}:${task.taskId}:${task.status}:${outcome}`;
|
||||
}
|
||||
return taskTerminalDeliveryIdempotencyKey(task);
|
||||
}
|
||||
|
||||
function getLinkedFlowForDelivery(task: TaskRecord) {
|
||||
const flowId = task.parentFlowId?.trim();
|
||||
if (!flowId || task.scopeKind !== "session") {
|
||||
return undefined;
|
||||
}
|
||||
const flow = getTaskFlowById(flowId);
|
||||
if (!flow) {
|
||||
return undefined;
|
||||
}
|
||||
if (normalizeOptionalString(flow.ownerKey) !== normalizeOptionalString(task.ownerKey)) {
|
||||
return undefined;
|
||||
}
|
||||
return flow;
|
||||
}
|
||||
|
||||
function resolveTaskDeliveryOwner(task: TaskRecord): TaskDeliveryOwner {
|
||||
const flow = getLinkedFlowForDelivery(task);
|
||||
if (flow) {
|
||||
return {
|
||||
sessionKey: flow.ownerKey.trim(),
|
||||
requesterOrigin: normalizeDeliveryContext(
|
||||
flow.requesterOrigin ?? taskDeliveryStates.get(task.taskId)?.requesterOrigin,
|
||||
),
|
||||
flowId: flow.flowId,
|
||||
};
|
||||
}
|
||||
if (task.scopeKind !== "session") {
|
||||
return {};
|
||||
}
|
||||
return {
|
||||
sessionKey: task.ownerKey.trim(),
|
||||
requesterOrigin: normalizeDeliveryContext(taskDeliveryStates.get(task.taskId)?.requesterOrigin),
|
||||
};
|
||||
}
|
||||
|
||||
function canDeliverTaskToRequesterOrigin(task: TaskRecord): boolean {
|
||||
const owner = resolveTaskDeliveryOwner(task);
|
||||
if (shouldRouteCompletionThroughRequesterSession(owner.sessionKey)) {
|
||||
return false;
|
||||
}
|
||||
return canDeliverToRequesterOrigin(owner.requesterOrigin);
|
||||
}
|
||||
|
||||
function canDeliverToRequesterOrigin(origin: TaskDeliveryState["requesterOrigin"]): boolean {
|
||||
const channel = origin?.channel?.trim();
|
||||
const to = origin?.to?.trim();
|
||||
return Boolean(channel && to && isDeliverableMessageChannel(channel));
|
||||
}
|
||||
|
||||
function canDeliverParentReviewTaskToBoundDiscordThread(task: TaskRecord): boolean {
|
||||
if (!shouldUseParentReviewTaskTerminalMessage(task)) {
|
||||
return false;
|
||||
}
|
||||
const owner = resolveTaskDeliveryOwner(task);
|
||||
const origin = owner.requesterOrigin;
|
||||
const channel = origin?.channel?.trim().toLowerCase();
|
||||
const to = origin?.to?.trim().toLowerCase();
|
||||
const threadId = String(origin?.threadId ?? "").trim();
|
||||
// This is a narrow transport exception for explicitly bound Discord threads,
|
||||
// not a general parent-review direct-delivery relaxation.
|
||||
return Boolean(
|
||||
channel === "discord" &&
|
||||
to?.startsWith("channel:") &&
|
||||
threadId &&
|
||||
canDeliverToRequesterOrigin(origin),
|
||||
);
|
||||
}
|
||||
|
||||
function resolveMissingOwnerDeliveryStatus(task: TaskRecord): TaskDeliveryStatus {
|
||||
return task.scopeKind === "system" ? "not_applicable" : "parent_missing";
|
||||
}
|
||||
|
||||
function queueTaskSystemEvent(task: TaskRecord, text: string) {
|
||||
const owner = resolveTaskDeliveryOwner(task);
|
||||
const ownerKey = owner.sessionKey?.trim();
|
||||
if (!ownerKey) {
|
||||
return false;
|
||||
}
|
||||
enqueueSystemEvent(text, {
|
||||
sessionKey: ownerKey,
|
||||
contextKey: `task:${task.taskId}`,
|
||||
deliveryContext: owner.requesterOrigin,
|
||||
});
|
||||
requestHeartbeat({
|
||||
source: "background-task",
|
||||
intent: "immediate",
|
||||
reason: "background-task",
|
||||
sessionKey: ownerKey,
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
function queueBlockedTaskFollowup(task: TaskRecord) {
|
||||
const followupText = formatTaskBlockedFollowupMessage(task);
|
||||
if (!followupText) {
|
||||
return false;
|
||||
}
|
||||
const owner = resolveTaskDeliveryOwner(task);
|
||||
const ownerKey = owner.sessionKey?.trim();
|
||||
if (!ownerKey) {
|
||||
return false;
|
||||
}
|
||||
enqueueSystemEvent(followupText, {
|
||||
sessionKey: ownerKey,
|
||||
contextKey: `task:${task.taskId}:blocked-followup`,
|
||||
deliveryContext: owner.requesterOrigin,
|
||||
});
|
||||
requestHeartbeat({
|
||||
source: "background-task-blocked",
|
||||
intent: "immediate",
|
||||
reason: "background-task-blocked",
|
||||
sessionKey: ownerKey,
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
export async function maybeDeliverTaskTerminalUpdate(taskId: string): Promise<TaskRecord | null> {
|
||||
return await runTaskDeliveryWithIndependentAdmission(taskId, async () =>
|
||||
maybeDeliverTaskTerminalUpdateUnderAdmission(taskId),
|
||||
);
|
||||
}
|
||||
|
||||
async function runTaskDeliveryWithIndependentAdmission(
|
||||
taskId: string,
|
||||
deliver: () => Promise<TaskRecord | null>,
|
||||
): Promise<TaskRecord | null> {
|
||||
ensureTaskRegistryReady();
|
||||
let admitted = false;
|
||||
try {
|
||||
return await runWithGatewayIndependentRootWorkAdmission(async () => {
|
||||
admitted = true;
|
||||
return await deliver();
|
||||
});
|
||||
} catch (error) {
|
||||
// Late lifecycle callbacks must not leak a rejected detached promise after
|
||||
// restart closes admission. An already-admitted delivery still reports its
|
||||
// own failures instead of hiding them behind a concurrent restart.
|
||||
if (!admitted && isGatewayRestartDraining()) {
|
||||
ensureTaskRegistryReady();
|
||||
const current = tasks.get(taskId);
|
||||
return current ? cloneTaskRecord(current) : null;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function maybeDeliverTaskTerminalUpdateUnderAdmission(
|
||||
taskId: string,
|
||||
): Promise<TaskRecord | null> {
|
||||
ensureTaskRegistryReady();
|
||||
const current = tasks.get(taskId);
|
||||
if (!current || !shouldAutoDeliverTaskTerminalUpdate(current)) {
|
||||
return current ? cloneTaskRecord(current) : null;
|
||||
}
|
||||
if (tasksWithPendingDelivery.has(taskId)) {
|
||||
return cloneTaskRecord(current);
|
||||
}
|
||||
tasksWithPendingDelivery.add(taskId);
|
||||
try {
|
||||
const latest = tasks.get(taskId);
|
||||
if (!latest || !shouldAutoDeliverTaskTerminalUpdate(latest)) {
|
||||
return latest ? cloneTaskRecord(latest) : null;
|
||||
}
|
||||
const peers = latest.runId ? getPeerTasksForDelivery(latest) : [];
|
||||
const isSubagentCancellation = latest.runtime === "subagent" && latest.status === "cancelled";
|
||||
const preferred = pickPreferredRunIdTask(
|
||||
isSubagentCancellation
|
||||
? peers.filter((candidate) => shouldAutoDeliverTaskTerminalUpdate(candidate))
|
||||
: peers,
|
||||
);
|
||||
const peerDeliveryCovered =
|
||||
isSubagentCancellation &&
|
||||
peers.some(
|
||||
(candidate) =>
|
||||
candidate.taskId !== latest.taskId &&
|
||||
(candidate.deliveryStatus === "delivered" ||
|
||||
candidate.deliveryStatus === "session_queued"),
|
||||
);
|
||||
if (
|
||||
shouldSuppressDuplicateTerminalDelivery({
|
||||
task: latest,
|
||||
preferredTaskId: preferred?.taskId,
|
||||
peerDeliveryCovered,
|
||||
})
|
||||
) {
|
||||
return updateTask(taskId, {
|
||||
deliveryStatus: "not_applicable",
|
||||
lastEventAt: Date.now(),
|
||||
});
|
||||
}
|
||||
const owner = resolveTaskDeliveryOwner(latest);
|
||||
const ownerSessionKey = owner.sessionKey?.trim();
|
||||
if (!ownerSessionKey) {
|
||||
return updateTask(taskId, {
|
||||
deliveryStatus: resolveMissingOwnerDeliveryStatus(latest),
|
||||
lastEventAt: Date.now(),
|
||||
});
|
||||
}
|
||||
const shouldRouteParentReview = shouldUseParentReviewTaskTerminalMessage(latest);
|
||||
const shouldDeliverParentReviewDirect = canDeliverParentReviewTaskToBoundDiscordThread(latest);
|
||||
const canDeliverDirect =
|
||||
canDeliverTaskToRequesterOrigin(latest) || shouldDeliverParentReviewDirect;
|
||||
const directEventText = formatTaskTerminalMessage(latest);
|
||||
const sessionEventText = formatTaskTerminalMessage(
|
||||
latest,
|
||||
shouldRouteParentReview ? { surface: "parent_session" } : undefined,
|
||||
);
|
||||
if ((shouldRouteParentReview && !shouldDeliverParentReviewDirect) || !canDeliverDirect) {
|
||||
try {
|
||||
queueTaskSystemEvent(latest, sessionEventText);
|
||||
if (latest.terminalOutcome === "blocked") {
|
||||
queueBlockedTaskFollowup(latest);
|
||||
}
|
||||
return updateTask(taskId, {
|
||||
deliveryStatus:
|
||||
shouldRouteParentReview && canDeliverDirect ? "pending" : "session_queued",
|
||||
lastEventAt: Date.now(),
|
||||
});
|
||||
} catch (error) {
|
||||
log.warn("Failed to queue background task session delivery", {
|
||||
taskId,
|
||||
ownerKey: latest.ownerKey,
|
||||
error,
|
||||
});
|
||||
return updateTask(taskId, {
|
||||
deliveryStatus: "failed",
|
||||
lastEventAt: Date.now(),
|
||||
});
|
||||
}
|
||||
}
|
||||
try {
|
||||
const { sendMessage } = await loadTaskRegistryDeliveryRuntime();
|
||||
const beforeSend = tasks.get(taskId);
|
||||
if (!beforeSend || !shouldAutoDeliverTaskTerminalUpdate(beforeSend)) {
|
||||
return beforeSend ? cloneTaskRecord(beforeSend) : null;
|
||||
}
|
||||
const requesterAgentId = parseAgentSessionKey(ownerSessionKey)?.agentId;
|
||||
const idempotencyKey = resolveTaskTerminalIdempotencyKey(latest);
|
||||
await sendMessage({
|
||||
channel: owner.requesterOrigin?.channel,
|
||||
to: owner.requesterOrigin?.to ?? "",
|
||||
accountId: owner.requesterOrigin?.accountId,
|
||||
threadId: owner.requesterOrigin?.threadId,
|
||||
content: shouldDeliverParentReviewDirect ? sessionEventText : directEventText,
|
||||
agentId: requesterAgentId,
|
||||
idempotencyKey,
|
||||
mirror: {
|
||||
sessionKey: ownerSessionKey,
|
||||
agentId: requesterAgentId,
|
||||
idempotencyKey,
|
||||
},
|
||||
});
|
||||
const afterSend = tasks.get(taskId);
|
||||
if (!afterSend || !shouldAutoDeliverTaskTerminalUpdate(afterSend)) {
|
||||
return afterSend ? cloneTaskRecord(afterSend) : null;
|
||||
}
|
||||
if (afterSend.terminalOutcome === "blocked") {
|
||||
queueBlockedTaskFollowup(afterSend);
|
||||
}
|
||||
return updateTask(taskId, {
|
||||
deliveryStatus: "delivered",
|
||||
lastEventAt: Date.now(),
|
||||
});
|
||||
} catch (error) {
|
||||
log.warn("Failed to deliver background task update", {
|
||||
taskId,
|
||||
ownerKey: ownerSessionKey,
|
||||
requesterOrigin: owner.requesterOrigin,
|
||||
error,
|
||||
});
|
||||
const beforeFallback = tasks.get(taskId);
|
||||
if (!beforeFallback || !shouldAutoDeliverTaskTerminalUpdate(beforeFallback)) {
|
||||
return beforeFallback ? cloneTaskRecord(beforeFallback) : null;
|
||||
}
|
||||
try {
|
||||
queueTaskSystemEvent(beforeFallback, sessionEventText);
|
||||
if (beforeFallback.terminalOutcome === "blocked") {
|
||||
queueBlockedTaskFollowup(beforeFallback);
|
||||
}
|
||||
} catch (fallbackError) {
|
||||
log.warn("Failed to queue background task fallback event", {
|
||||
taskId,
|
||||
ownerKey: latest.ownerKey,
|
||||
error: fallbackError,
|
||||
});
|
||||
}
|
||||
return updateTask(taskId, {
|
||||
deliveryStatus: "failed",
|
||||
lastEventAt: Date.now(),
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
tasksWithPendingDelivery.delete(taskId);
|
||||
}
|
||||
}
|
||||
|
||||
export async function maybeDeliverTaskStateChangeUpdate(
|
||||
taskId: string,
|
||||
latestEvent?: TaskEventRecord,
|
||||
): Promise<TaskRecord | null> {
|
||||
return await runTaskDeliveryWithIndependentAdmission(taskId, async () =>
|
||||
maybeDeliverTaskStateChangeUpdateUnderAdmission(taskId, latestEvent),
|
||||
);
|
||||
}
|
||||
|
||||
async function maybeDeliverTaskStateChangeUpdateUnderAdmission(
|
||||
taskId: string,
|
||||
latestEvent?: TaskEventRecord,
|
||||
): Promise<TaskRecord | null> {
|
||||
ensureTaskRegistryReady();
|
||||
const current = tasks.get(taskId);
|
||||
if (!current || !shouldAutoDeliverTaskStateChange(current)) {
|
||||
return current ? cloneTaskRecord(current) : null;
|
||||
}
|
||||
const deliveryState = getTaskDeliveryState(taskId);
|
||||
if (!latestEvent || (deliveryState?.lastNotifiedEventAt ?? 0) >= latestEvent.at) {
|
||||
return cloneTaskRecord(current);
|
||||
}
|
||||
const eventText = formatTaskStateChangeMessage(current, latestEvent);
|
||||
if (!eventText) {
|
||||
return cloneTaskRecord(current);
|
||||
}
|
||||
try {
|
||||
const owner = resolveTaskDeliveryOwner(current);
|
||||
const ownerSessionKey = owner.sessionKey?.trim();
|
||||
if (!ownerSessionKey) {
|
||||
return updateTask(taskId, {
|
||||
deliveryStatus: resolveMissingOwnerDeliveryStatus(current),
|
||||
lastEventAt: Date.now(),
|
||||
});
|
||||
}
|
||||
if (!canDeliverTaskToRequesterOrigin(current)) {
|
||||
queueTaskSystemEvent(current, eventText);
|
||||
upsertTaskDeliveryState({
|
||||
taskId,
|
||||
requesterOrigin: deliveryState?.requesterOrigin,
|
||||
lastNotifiedEventAt: latestEvent.at,
|
||||
});
|
||||
return updateTask(taskId, {
|
||||
lastEventAt: Date.now(),
|
||||
});
|
||||
}
|
||||
const { sendMessage } = await loadTaskRegistryDeliveryRuntime();
|
||||
const requesterAgentId = parseAgentSessionKey(ownerSessionKey)?.agentId;
|
||||
const idempotencyKey = resolveTaskStateChangeIdempotencyKey({
|
||||
task: current,
|
||||
latestEvent,
|
||||
owner,
|
||||
});
|
||||
await sendMessage({
|
||||
channel: owner.requesterOrigin?.channel,
|
||||
to: owner.requesterOrigin?.to ?? "",
|
||||
accountId: owner.requesterOrigin?.accountId,
|
||||
threadId: owner.requesterOrigin?.threadId,
|
||||
content: eventText,
|
||||
agentId: requesterAgentId,
|
||||
idempotencyKey,
|
||||
mirror: {
|
||||
sessionKey: ownerSessionKey,
|
||||
agentId: requesterAgentId,
|
||||
idempotencyKey,
|
||||
},
|
||||
});
|
||||
upsertTaskDeliveryState({
|
||||
taskId,
|
||||
requesterOrigin: deliveryState?.requesterOrigin,
|
||||
lastNotifiedEventAt: latestEvent.at,
|
||||
});
|
||||
return updateTask(taskId, {
|
||||
lastEventAt: Date.now(),
|
||||
});
|
||||
} catch (error) {
|
||||
log.warn("Failed to deliver background task state change", {
|
||||
taskId,
|
||||
ownerKey: current.ownerKey,
|
||||
error,
|
||||
});
|
||||
return cloneTaskRecord(current);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
import { onAgentEvent } from "../infra/agent-events.js";
|
||||
import { isTerminalTaskStatus } from "./task-executor-policy.js";
|
||||
import {
|
||||
appendTaskEvent,
|
||||
buildTaskLifecycleTerminalOutcome,
|
||||
mapAgentRunTerminalOutcomeToTaskStatus,
|
||||
resolveTaskLifecycleTerminalError,
|
||||
} from "./task-registry-common.js";
|
||||
import {
|
||||
maybeDeliverTaskStateChangeUpdate,
|
||||
maybeDeliverTaskTerminalUpdate,
|
||||
} from "./task-registry-delivery.js";
|
||||
import { updateTask } from "./task-registry-mutation.js";
|
||||
import {
|
||||
claimTaskRegistryListenerStart,
|
||||
getTasksByRunScope,
|
||||
restoreTaskRegistryOnce,
|
||||
setTaskRegistryListenerStarter,
|
||||
setTaskRegistryListenerStop,
|
||||
} from "./task-registry-state.js";
|
||||
import type { TaskRecord } from "./task-registry.types.js";
|
||||
|
||||
function ensureListener() {
|
||||
if (!claimTaskRegistryListenerStart()) {
|
||||
return;
|
||||
}
|
||||
const stop = onAgentEvent((evt) => {
|
||||
restoreTaskRegistryOnce();
|
||||
const scopedTasks = getTasksByRunScope({
|
||||
runId: evt.runId,
|
||||
sessionKey: evt.sessionKey,
|
||||
});
|
||||
if (scopedTasks.length === 0) {
|
||||
return;
|
||||
}
|
||||
const now = evt.ts || Date.now();
|
||||
for (const current of scopedTasks) {
|
||||
if (isTerminalTaskStatus(current.status)) {
|
||||
continue;
|
||||
}
|
||||
const patch: Partial<TaskRecord> = {
|
||||
lastEventAt: now,
|
||||
};
|
||||
if (evt.stream === "lifecycle") {
|
||||
const phase = typeof evt.data?.phase === "string" ? evt.data.phase : undefined;
|
||||
const eventStartedAt = evt.data?.startedAt;
|
||||
const startedAt =
|
||||
typeof eventStartedAt === "number" && Number.isFinite(eventStartedAt)
|
||||
? eventStartedAt
|
||||
: current.startedAt;
|
||||
const endedAt = typeof evt.data?.endedAt === "number" ? evt.data.endedAt : undefined;
|
||||
if (startedAt !== undefined) {
|
||||
patch.startedAt = startedAt;
|
||||
}
|
||||
if (phase === "start") {
|
||||
patch.status = "running";
|
||||
} else if (phase === "end") {
|
||||
const terminal = buildTaskLifecycleTerminalOutcome({
|
||||
phase,
|
||||
data: evt.data,
|
||||
startedAt,
|
||||
endedAt: endedAt ?? now,
|
||||
});
|
||||
patch.status = mapAgentRunTerminalOutcomeToTaskStatus(terminal);
|
||||
patch.endedAt = terminal.endedAt ?? now;
|
||||
const error = resolveTaskLifecycleTerminalError({
|
||||
runtime: current.runtime,
|
||||
status: patch.status,
|
||||
error: terminal.error,
|
||||
});
|
||||
if (error) {
|
||||
patch.error = error;
|
||||
}
|
||||
} else if (phase === "error") {
|
||||
const terminal = buildTaskLifecycleTerminalOutcome({
|
||||
phase,
|
||||
data: evt.data,
|
||||
startedAt,
|
||||
endedAt: endedAt ?? now,
|
||||
});
|
||||
patch.status = mapAgentRunTerminalOutcomeToTaskStatus(terminal);
|
||||
patch.endedAt = terminal.endedAt ?? now;
|
||||
patch.error =
|
||||
resolveTaskLifecycleTerminalError({
|
||||
runtime: current.runtime,
|
||||
status: patch.status,
|
||||
error: terminal.error,
|
||||
}) ?? current.error;
|
||||
}
|
||||
} else if (evt.stream === "error") {
|
||||
patch.error = typeof evt.data?.error === "string" ? evt.data.error : current.error;
|
||||
} else if (evt.stream === "tool" && evt.data?.phase === "start") {
|
||||
// Tool starts are the activity signal surfaced in task summaries; ends
|
||||
// and outputs only refresh lastEventAt.
|
||||
const toolName = typeof evt.data.name === "string" ? evt.data.name.trim() : "";
|
||||
if (toolName) {
|
||||
patch.toolUseCount = (current.toolUseCount ?? 0) + 1;
|
||||
patch.lastToolName = toolName;
|
||||
}
|
||||
}
|
||||
const stateChangeEvent =
|
||||
patch.status && patch.status !== current.status
|
||||
? appendTaskEvent({
|
||||
at: now,
|
||||
kind: patch.status,
|
||||
summary:
|
||||
patch.status === "failed"
|
||||
? (patch.error ?? current.error)
|
||||
: patch.status === "succeeded"
|
||||
? current.terminalSummary
|
||||
: undefined,
|
||||
})
|
||||
: undefined;
|
||||
const updated = updateTask(current.taskId, patch);
|
||||
if (updated) {
|
||||
void maybeDeliverTaskStateChangeUpdate(current.taskId, stateChangeEvent);
|
||||
void maybeDeliverTaskTerminalUpdate(current.taskId);
|
||||
}
|
||||
}
|
||||
});
|
||||
setTaskRegistryListenerStop(stop);
|
||||
}
|
||||
|
||||
setTaskRegistryListenerStarter(ensureListener);
|
||||
@@ -0,0 +1,243 @@
|
||||
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
|
||||
import { runWithGatewayIndependentRootWorkAdmission } from "../process/gateway-work-admission.js";
|
||||
import { normalizeDeliveryContext } from "../utils/delivery-context.shared.js";
|
||||
import { isTaskFlowCancellationPending } from "./task-cancellation-state.js";
|
||||
import { isTerminalTaskStatus } from "./task-executor-policy.js";
|
||||
import {
|
||||
getTaskFlowById,
|
||||
syncFlowFromTaskResult,
|
||||
updateFlowRecordByIdExpectedRevision,
|
||||
} from "./task-flow-runtime-internal.js";
|
||||
import { ensureLinkedTaskFlowRegistryReady, isTerminalFlowStatus } from "./task-registry-common.js";
|
||||
import { findLatestTaskForFlowId, listTasksForFlowId } from "./task-registry-query.js";
|
||||
import {
|
||||
cloneTaskDeliveryState,
|
||||
cloneTaskRecord,
|
||||
normalizeTaskTimestamps,
|
||||
} from "./task-registry-records.js";
|
||||
import {
|
||||
addOwnerKeyIndex,
|
||||
addParentFlowIdIndex,
|
||||
addRelatedSessionKeyIndex,
|
||||
deleteOwnerKeyIndex,
|
||||
deleteParentFlowIdIndex,
|
||||
deleteRelatedSessionKeyIndex,
|
||||
emitTaskRegistryObserverEvent,
|
||||
log,
|
||||
rebuildRunIdIndex,
|
||||
taskDeliveryStates,
|
||||
taskFlowSyncRetryTimers,
|
||||
tasks,
|
||||
TASK_FLOW_SYNC_RETRY_DELAYS_MS,
|
||||
tryPersistTaskDeliveryStateUpsert,
|
||||
tryPersistTaskUpsert,
|
||||
} from "./task-registry-state.js";
|
||||
import type { TaskDeliveryState, TaskRecord } from "./task-registry.types.js";
|
||||
import { resolveTaskCleanupAfter } from "./task-retention.js";
|
||||
|
||||
function syncManagedFlowCancellationFromTask(task: TaskRecord): void {
|
||||
const flowId = task.parentFlowId?.trim();
|
||||
if (!flowId) {
|
||||
return;
|
||||
}
|
||||
let flow = getTaskFlowById(flowId);
|
||||
if (
|
||||
!flow ||
|
||||
flow.syncMode !== "managed" ||
|
||||
flow.cancelRequestedAt == null ||
|
||||
isTerminalFlowStatus(flow.status)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
if (listTasksForFlowId(flowId).some(isTaskFlowCancellationPending)) {
|
||||
return;
|
||||
}
|
||||
const endedAt = task.endedAt ?? task.lastEventAt ?? Date.now();
|
||||
for (let attempt = 0; attempt < 2; attempt += 1) {
|
||||
const result = updateFlowRecordByIdExpectedRevision({
|
||||
flowId,
|
||||
expectedRevision: flow.revision,
|
||||
patch: {
|
||||
status: "cancelled",
|
||||
blockedTaskId: null,
|
||||
blockedSummary: null,
|
||||
waitJson: null,
|
||||
endedAt,
|
||||
updatedAt: endedAt,
|
||||
},
|
||||
});
|
||||
if (result.applied || result.reason === "not_found") {
|
||||
return;
|
||||
}
|
||||
flow = result.current;
|
||||
if (
|
||||
!flow ||
|
||||
flow.syncMode !== "managed" ||
|
||||
flow.cancelRequestedAt == null ||
|
||||
isTerminalFlowStatus(flow.status)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
if (listTasksForFlowId(flowId).some(isTaskFlowCancellationPending)) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function scheduleTaskFlowSyncRetry(task: TaskRecord, operation: string, attempt = 0): void {
|
||||
const taskId = task.taskId.trim();
|
||||
if (!taskId || taskFlowSyncRetryTimers.has(taskId)) {
|
||||
return;
|
||||
}
|
||||
const delayMs = TASK_FLOW_SYNC_RETRY_DELAYS_MS[attempt];
|
||||
if (delayMs == null) {
|
||||
log.warn("Exhausted parent flow sync retries from task", {
|
||||
operation,
|
||||
taskId,
|
||||
flowId: task.parentFlowId,
|
||||
});
|
||||
return;
|
||||
}
|
||||
const retryTimer = setTimeout(() => {
|
||||
taskFlowSyncRetryTimers.delete(taskId);
|
||||
// A terminal task no longer blocks suspension, but its durable parent-flow
|
||||
// projection still mutates state. Keep every delayed attempt visible and
|
||||
// prevent it from crossing a prepared host snapshot boundary.
|
||||
void runWithGatewayIndependentRootWorkAdmission(async () => {
|
||||
const current = tasks.get(taskId);
|
||||
if (!current) {
|
||||
return;
|
||||
}
|
||||
const flowId = current.parentFlowId?.trim();
|
||||
if (!flowId || findLatestTaskForFlowId(flowId)?.taskId !== taskId) {
|
||||
return;
|
||||
}
|
||||
const result = syncFlowFromTaskResult(current);
|
||||
if (!result.ok) {
|
||||
log.warn("Failed to retry parent flow sync from task", {
|
||||
operation,
|
||||
taskId,
|
||||
flowId: current.parentFlowId,
|
||||
reason: result.reason,
|
||||
});
|
||||
scheduleTaskFlowSyncRetry(current, operation, attempt + 1);
|
||||
}
|
||||
}).catch((error: unknown) => {
|
||||
log.warn("Failed to admit parent flow sync retry from task", {
|
||||
operation,
|
||||
taskId,
|
||||
flowId: task.parentFlowId,
|
||||
error,
|
||||
});
|
||||
});
|
||||
}, delayMs);
|
||||
retryTimer.unref?.();
|
||||
taskFlowSyncRetryTimers.set(taskId, retryTimer);
|
||||
}
|
||||
|
||||
export function syncFlowFromTaskAfterTaskMutation(task: TaskRecord, operation: string): void {
|
||||
const result = syncFlowFromTaskResult(task);
|
||||
if (result.ok) {
|
||||
return;
|
||||
}
|
||||
log.warn("Failed to sync parent flow from task mutation", {
|
||||
operation,
|
||||
taskId: task.taskId,
|
||||
flowId: task.parentFlowId,
|
||||
reason: result.reason,
|
||||
});
|
||||
scheduleTaskFlowSyncRetry(task, operation);
|
||||
}
|
||||
|
||||
export function updateTask(taskId: string, patch: Partial<TaskRecord>): TaskRecord | null {
|
||||
const current = tasks.get(taskId);
|
||||
if (!current) {
|
||||
return null;
|
||||
}
|
||||
const next = normalizeTaskTimestamps({
|
||||
...current,
|
||||
...patch,
|
||||
...(patch.detail !== undefined ? { detail: structuredClone(patch.detail) } : {}),
|
||||
});
|
||||
if (Object.hasOwn(patch, "error") && patch.error === undefined) {
|
||||
delete next.error;
|
||||
}
|
||||
if (Object.hasOwn(patch, "childSessionKey") && patch.childSessionKey === undefined) {
|
||||
delete next.childSessionKey;
|
||||
}
|
||||
if (isTerminalTaskStatus(next.status) && typeof next.cleanupAfter !== "number") {
|
||||
const createdAt = next.createdAt ?? Date.now();
|
||||
const cleanupAfter = resolveTaskCleanupAfter({ ...next, createdAt });
|
||||
Object.assign(next, cleanupAfter === undefined ? {} : { cleanupAfter });
|
||||
}
|
||||
const sessionIndexChanged =
|
||||
normalizeOptionalString(current.ownerKey) !== normalizeOptionalString(next.ownerKey) ||
|
||||
normalizeOptionalString(current.childSessionKey) !==
|
||||
normalizeOptionalString(next.childSessionKey);
|
||||
const parentFlowIndexChanged = current.parentFlowId?.trim() !== next.parentFlowId?.trim();
|
||||
ensureLinkedTaskFlowRegistryReady(current);
|
||||
ensureLinkedTaskFlowRegistryReady(next);
|
||||
// Persist before mutating memory. If the store rejects the write, keep the
|
||||
// in-memory mirror at the durable value and report that no mutation applied.
|
||||
if (!tryPersistTaskUpsert(next, "update")) {
|
||||
return null;
|
||||
}
|
||||
tasks.set(taskId, next);
|
||||
if (patch.runId && patch.runId !== current.runId) {
|
||||
rebuildRunIdIndex();
|
||||
}
|
||||
if (sessionIndexChanged) {
|
||||
deleteOwnerKeyIndex(taskId, current);
|
||||
addOwnerKeyIndex(taskId, next);
|
||||
deleteRelatedSessionKeyIndex(taskId, current);
|
||||
addRelatedSessionKeyIndex(taskId, next);
|
||||
}
|
||||
if (parentFlowIndexChanged) {
|
||||
deleteParentFlowIdIndex(taskId, current);
|
||||
addParentFlowIdIndex(taskId, next);
|
||||
}
|
||||
syncFlowFromTaskAfterTaskMutation(next, "update");
|
||||
try {
|
||||
syncManagedFlowCancellationFromTask(next);
|
||||
} catch (error) {
|
||||
log.warn("Failed to finalize managed flow cancellation from task update", {
|
||||
taskId,
|
||||
flowId: next.parentFlowId,
|
||||
error,
|
||||
});
|
||||
}
|
||||
emitTaskRegistryObserverEvent(() => ({
|
||||
kind: "upserted",
|
||||
task: cloneTaskRecord(next),
|
||||
previous: cloneTaskRecord(current),
|
||||
}));
|
||||
return cloneTaskRecord(next);
|
||||
}
|
||||
|
||||
export function upsertTaskDeliveryState(state: TaskDeliveryState): TaskDeliveryState {
|
||||
const current = taskDeliveryStates.get(state.taskId);
|
||||
const next: TaskDeliveryState = {
|
||||
taskId: state.taskId,
|
||||
...(state.requesterOrigin
|
||||
? { requesterOrigin: normalizeDeliveryContext(state.requesterOrigin) }
|
||||
: {}),
|
||||
...(state.lastNotifiedEventAt != null
|
||||
? { lastNotifiedEventAt: state.lastNotifiedEventAt }
|
||||
: {}),
|
||||
};
|
||||
if (!next.requesterOrigin && typeof next.lastNotifiedEventAt !== "number" && !current) {
|
||||
return cloneTaskDeliveryState({ taskId: state.taskId });
|
||||
}
|
||||
if (!tryPersistTaskDeliveryStateUpsert(next)) {
|
||||
return current
|
||||
? cloneTaskDeliveryState(current)
|
||||
: cloneTaskDeliveryState({ taskId: state.taskId });
|
||||
}
|
||||
taskDeliveryStates.set(state.taskId, next);
|
||||
return cloneTaskDeliveryState(next);
|
||||
}
|
||||
|
||||
export function getTaskDeliveryState(taskId: string): TaskDeliveryState | undefined {
|
||||
const state = taskDeliveryStates.get(taskId);
|
||||
return state ? cloneTaskDeliveryState(state) : undefined;
|
||||
}
|
||||
@@ -0,0 +1,275 @@
|
||||
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
|
||||
import { isActiveTaskStatus, ensureLinkedTaskFlowRegistryReady } from "./task-registry-common.js";
|
||||
import type { TaskRegistryControlRuntime } from "./task-registry-control.types.js";
|
||||
import { cloneTaskRecord, normalizeTaskTimestamps } from "./task-registry-records.js";
|
||||
import {
|
||||
TASK_REGISTRY_CONTROL_RUNTIME_OVERRIDE_KEY,
|
||||
TASK_REGISTRY_DELIVERY_RUNTIME_OVERRIDE_KEY,
|
||||
clearTaskRegistryMemory,
|
||||
compareTasksNewestFirst,
|
||||
controlRuntimeLoader,
|
||||
deleteOwnerKeyIndex,
|
||||
deleteParentFlowIdIndex,
|
||||
deleteRelatedSessionKeyIndex,
|
||||
deliveryRuntimeLoader,
|
||||
emitTaskRegistryObserverEvent,
|
||||
ensureTaskRegistryReady,
|
||||
getTasksByRunId,
|
||||
log,
|
||||
persistTaskRegistry,
|
||||
pickPreferredRunIdTask,
|
||||
rebuildRunIdIndex,
|
||||
resetTaskRegistryListenerState,
|
||||
resetTaskRegistryRestoreState,
|
||||
snapshotTaskRecords,
|
||||
taskDeliveryStates,
|
||||
taskIdsByOwnerKey,
|
||||
taskIdsByParentFlowId,
|
||||
taskIdsByRelatedSessionKey,
|
||||
tasks,
|
||||
tryPersistTaskDelete,
|
||||
type TaskRegistryDeliveryRuntime,
|
||||
type TaskRegistryGlobalWithRuntimeOverrides,
|
||||
} from "./task-registry-state.js";
|
||||
import { getTaskRegistryStore, resetTaskRegistryRuntimeForTests } from "./task-registry.store.js";
|
||||
import type { TaskRecord } from "./task-registry.types.js";
|
||||
|
||||
export function listTaskRecordsUnsorted(): TaskRecord[] {
|
||||
ensureTaskRegistryReady();
|
||||
return snapshotTaskRecords(tasks);
|
||||
}
|
||||
|
||||
export function listTaskRecords(): TaskRecord[] {
|
||||
ensureTaskRegistryReady();
|
||||
return [...tasks.values()]
|
||||
.map((task, insertionIndex) => Object.assign({}, cloneTaskRecord(task), { insertionIndex }))
|
||||
.toSorted(compareTasksNewestFirst)
|
||||
.map(({ insertionIndex: _, ...task }) => task);
|
||||
}
|
||||
|
||||
export function hasActiveTaskForChildSessionKey(params: {
|
||||
sessionKey: string;
|
||||
excludeTaskId?: string;
|
||||
}): boolean {
|
||||
ensureTaskRegistryReady();
|
||||
const sessionKey = normalizeOptionalString(params.sessionKey);
|
||||
if (!sessionKey) {
|
||||
return false;
|
||||
}
|
||||
const ids = taskIdsByRelatedSessionKey.get(sessionKey);
|
||||
if (!ids) {
|
||||
return false;
|
||||
}
|
||||
for (const taskId of ids) {
|
||||
if (taskId === params.excludeTaskId) {
|
||||
continue;
|
||||
}
|
||||
const task = tasks.get(taskId);
|
||||
if (
|
||||
task &&
|
||||
isActiveTaskStatus(task.status) &&
|
||||
normalizeOptionalString(task.childSessionKey) === sessionKey
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
export function getTaskById(taskId: string): TaskRecord | undefined {
|
||||
ensureTaskRegistryReady();
|
||||
const task = tasks.get(taskId.trim());
|
||||
return task ? cloneTaskRecord(task) : undefined;
|
||||
}
|
||||
|
||||
export function findTaskByRunId(runId: string): TaskRecord | undefined {
|
||||
ensureTaskRegistryReady();
|
||||
const task = pickPreferredRunIdTask(getTasksByRunId(runId));
|
||||
return task ? cloneTaskRecord(task) : undefined;
|
||||
}
|
||||
|
||||
function listTasksFromIndex(index: Map<string, Set<string>>, key: string): TaskRecord[] {
|
||||
const ids = index.get(key);
|
||||
if (!ids || ids.size === 0) {
|
||||
return [];
|
||||
}
|
||||
return [...ids]
|
||||
.map((taskId, insertionIndex) => {
|
||||
const task = tasks.get(taskId);
|
||||
return task ? Object.assign({}, cloneTaskRecord(task), { insertionIndex }) : null;
|
||||
})
|
||||
.filter(
|
||||
(
|
||||
task,
|
||||
): task is TaskRecord & {
|
||||
insertionIndex: number;
|
||||
} => Boolean(task),
|
||||
)
|
||||
.toSorted(compareTasksNewestFirst)
|
||||
.map(({ insertionIndex: _, ...task }) => task);
|
||||
}
|
||||
|
||||
export function listTasksForSessionKey(sessionKey: string): TaskRecord[] {
|
||||
ensureTaskRegistryReady();
|
||||
const key = normalizeOptionalString(sessionKey);
|
||||
if (!key) {
|
||||
return [];
|
||||
}
|
||||
return listTasksFromIndex(taskIdsByRelatedSessionKey, key);
|
||||
}
|
||||
|
||||
export function listTasksForAgentId(agentId: string): TaskRecord[] {
|
||||
ensureTaskRegistryReady();
|
||||
const lookup = agentId.trim();
|
||||
if (!lookup) {
|
||||
return [];
|
||||
}
|
||||
return snapshotTaskRecords(tasks)
|
||||
.filter((task) => task.agentId?.trim() === lookup)
|
||||
.toSorted(compareTasksNewestFirst);
|
||||
}
|
||||
|
||||
export function findLatestTaskForFlowId(flowId: string): TaskRecord | undefined {
|
||||
const task = listTasksForFlowId(flowId)[0];
|
||||
return task ? cloneTaskRecord(task) : undefined;
|
||||
}
|
||||
|
||||
export function listTasksForOwnerKey(ownerKey: string): TaskRecord[] {
|
||||
ensureTaskRegistryReady();
|
||||
const key = normalizeOptionalString(ownerKey);
|
||||
if (!key) {
|
||||
return [];
|
||||
}
|
||||
return listTasksFromIndex(taskIdsByOwnerKey, key);
|
||||
}
|
||||
|
||||
export function listFreshTasksForOwnerKey(ownerKey: string): TaskRecord[] {
|
||||
ensureTaskRegistryReady();
|
||||
const key = normalizeOptionalString(ownerKey);
|
||||
if (!key) {
|
||||
return [];
|
||||
}
|
||||
const store = getTaskRegistryStore();
|
||||
if (store.listTasksForOwnerKey) {
|
||||
try {
|
||||
const merged = new Map<string, TaskRecord>();
|
||||
for (const task of store.listTasksForOwnerKey(key)) {
|
||||
merged.set(task.taskId, cloneTaskRecord(normalizeTaskTimestamps(task)));
|
||||
}
|
||||
return [...merged.values()]
|
||||
.map((task, insertionIndex) => Object.assign({}, task, { insertionIndex }))
|
||||
.toSorted(compareTasksNewestFirst)
|
||||
.map(({ insertionIndex: _, ...task }) => task);
|
||||
} catch (error) {
|
||||
log.warn("Failed to read fresh owner task registry records", {
|
||||
ownerKey: key,
|
||||
error,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return listTasksFromIndex(taskIdsByOwnerKey, key);
|
||||
}
|
||||
|
||||
export function listTasksForFlowId(flowId: string): TaskRecord[] {
|
||||
ensureTaskRegistryReady();
|
||||
const key = flowId.trim();
|
||||
if (!key) {
|
||||
return [];
|
||||
}
|
||||
return listTasksFromIndex(taskIdsByParentFlowId, key);
|
||||
}
|
||||
|
||||
function findLatestTaskForRelatedSessionKey(sessionKey: string): TaskRecord | undefined {
|
||||
const task = listTasksForRelatedSessionKey(sessionKey)[0];
|
||||
return task ? cloneTaskRecord(task) : undefined;
|
||||
}
|
||||
|
||||
export function listTasksForRelatedSessionKey(sessionKey: string): TaskRecord[] {
|
||||
ensureTaskRegistryReady();
|
||||
const key = normalizeOptionalString(sessionKey);
|
||||
if (!key) {
|
||||
return [];
|
||||
}
|
||||
return listTasksFromIndex(taskIdsByRelatedSessionKey, key);
|
||||
}
|
||||
|
||||
export function resolveTaskForLookupToken(token: string): TaskRecord | undefined {
|
||||
const lookup = token.trim();
|
||||
if (!lookup) {
|
||||
return undefined;
|
||||
}
|
||||
return (
|
||||
getTaskById(lookup) ?? findTaskByRunId(lookup) ?? findLatestTaskForRelatedSessionKey(lookup)
|
||||
);
|
||||
}
|
||||
|
||||
export function deleteTaskRecordById(taskId: string): boolean {
|
||||
ensureTaskRegistryReady();
|
||||
const current = tasks.get(taskId);
|
||||
if (!current) {
|
||||
return false;
|
||||
}
|
||||
ensureLinkedTaskFlowRegistryReady(current);
|
||||
// Persist the delete before mutating memory, as a single atomic store
|
||||
// operation. If persistence fails, leave the in-memory record intact and
|
||||
// report that no delete was applied.
|
||||
if (!tryPersistTaskDelete(taskId)) {
|
||||
return false;
|
||||
}
|
||||
deleteOwnerKeyIndex(taskId, current);
|
||||
deleteParentFlowIdIndex(taskId, current);
|
||||
deleteRelatedSessionKeyIndex(taskId, current);
|
||||
tasks.delete(taskId);
|
||||
taskDeliveryStates.delete(taskId);
|
||||
rebuildRunIdIndex();
|
||||
emitTaskRegistryObserverEvent(() => ({
|
||||
kind: "deleted",
|
||||
taskId: current.taskId,
|
||||
previous: cloneTaskRecord(current),
|
||||
}));
|
||||
return true;
|
||||
}
|
||||
|
||||
export function resetTaskRegistryForTests(opts?: { persist?: boolean }) {
|
||||
clearTaskRegistryMemory();
|
||||
resetTaskRegistryRestoreState();
|
||||
resetTaskRegistryRuntimeForTests();
|
||||
resetTaskRegistryListenerState();
|
||||
deliveryRuntimeLoader.clear();
|
||||
controlRuntimeLoader.clear();
|
||||
if (opts?.persist !== false) {
|
||||
persistTaskRegistry();
|
||||
}
|
||||
// Always close the sqlite handle so Windows temp-dir cleanup can remove the
|
||||
// state directory even when a test intentionally skips persisting the reset.
|
||||
getTaskRegistryStore().close?.();
|
||||
}
|
||||
|
||||
export function resetTaskRegistryDeliveryRuntimeForTests() {
|
||||
(globalThis as TaskRegistryGlobalWithRuntimeOverrides)[
|
||||
TASK_REGISTRY_DELIVERY_RUNTIME_OVERRIDE_KEY
|
||||
] = null;
|
||||
deliveryRuntimeLoader.clear();
|
||||
}
|
||||
|
||||
export function setTaskRegistryDeliveryRuntimeForTests(runtime: TaskRegistryDeliveryRuntime): void {
|
||||
(globalThis as TaskRegistryGlobalWithRuntimeOverrides)[
|
||||
TASK_REGISTRY_DELIVERY_RUNTIME_OVERRIDE_KEY
|
||||
] = runtime;
|
||||
deliveryRuntimeLoader.clear();
|
||||
}
|
||||
|
||||
export function resetTaskRegistryControlRuntimeForTests() {
|
||||
(globalThis as TaskRegistryGlobalWithRuntimeOverrides)[
|
||||
TASK_REGISTRY_CONTROL_RUNTIME_OVERRIDE_KEY
|
||||
] = null;
|
||||
controlRuntimeLoader.clear();
|
||||
}
|
||||
|
||||
export function setTaskRegistryControlRuntimeForTests(runtime: TaskRegistryControlRuntime): void {
|
||||
(globalThis as TaskRegistryGlobalWithRuntimeOverrides)[
|
||||
TASK_REGISTRY_CONTROL_RUNTIME_OVERRIDE_KEY
|
||||
] = runtime;
|
||||
controlRuntimeLoader.clear();
|
||||
}
|
||||
@@ -0,0 +1,562 @@
|
||||
import crypto from "node:crypto";
|
||||
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
|
||||
import { normalizeDeliveryContext } from "../utils/delivery-context.shared.js";
|
||||
import { isTerminalTaskStatus } from "./task-executor-policy.js";
|
||||
import {
|
||||
appendTaskEvent,
|
||||
assertParentFlowLinkAllowed,
|
||||
assertTaskOwner,
|
||||
ensureDeliveryStatus,
|
||||
ensureNotifyPolicy,
|
||||
normalizeTaskStatus,
|
||||
normalizeTaskSummary,
|
||||
resolveTaskOwnerKey,
|
||||
resolveTaskRequesterSessionKey,
|
||||
resolveTaskScopeKind,
|
||||
resolveTaskTerminalOutcome,
|
||||
shouldApplyRunScopedStatusUpdate,
|
||||
} from "./task-registry-common.js";
|
||||
import {
|
||||
findExistingTaskForCreate,
|
||||
mergeExistingTaskForCreate,
|
||||
resolveTaskAgentId,
|
||||
resolveTaskRequesterAgentId,
|
||||
} from "./task-registry-create-helpers.js";
|
||||
import {
|
||||
maybeDeliverTaskStateChangeUpdate,
|
||||
maybeDeliverTaskTerminalUpdate,
|
||||
} from "./task-registry-delivery.js";
|
||||
import { syncFlowFromTaskAfterTaskMutation, updateTask } from "./task-registry-mutation.js";
|
||||
import { cloneTaskRecord, normalizeTaskTimestamps } from "./task-registry-records.js";
|
||||
import {
|
||||
addOwnerKeyIndex,
|
||||
addParentFlowIdIndex,
|
||||
addRelatedSessionKeyIndex,
|
||||
addRunIdIndex,
|
||||
emitTaskRegistryObserverEvent,
|
||||
ensureTaskRegistryReady,
|
||||
getTasksByRunScope,
|
||||
taskDeliveryStates,
|
||||
tasks,
|
||||
tryPersistTaskUpsert,
|
||||
} from "./task-registry-state.js";
|
||||
import type {
|
||||
JsonValue,
|
||||
TaskDeliveryState,
|
||||
TaskDeliveryStatus,
|
||||
TaskNotifyPolicy,
|
||||
TaskRecord,
|
||||
TaskRuntime,
|
||||
TaskScopeKind,
|
||||
TaskStatus,
|
||||
TaskTerminalOutcome,
|
||||
} from "./task-registry.types.js";
|
||||
import { resolveTaskCleanupAfter } from "./task-retention.js";
|
||||
|
||||
export function setTaskCleanupAfterById(params: {
|
||||
taskId: string;
|
||||
cleanupAfter: number;
|
||||
}): TaskRecord | null {
|
||||
ensureTaskRegistryReady();
|
||||
return updateTask(params.taskId, {
|
||||
cleanupAfter: params.cleanupAfter,
|
||||
});
|
||||
}
|
||||
|
||||
export function markTaskTerminalById(params: {
|
||||
taskId: string;
|
||||
status: Extract<TaskStatus, "succeeded" | "failed" | "timed_out" | "cancelled">;
|
||||
childSessionKey?: string | null;
|
||||
endedAt: number;
|
||||
lastEventAt?: number;
|
||||
error?: string;
|
||||
terminalSummary?: string | null;
|
||||
preserveTerminalSummary?: boolean;
|
||||
terminalOutcome?: TaskTerminalOutcome | null;
|
||||
detail?: JsonValue;
|
||||
}): TaskRecord | null {
|
||||
ensureTaskRegistryReady();
|
||||
const patch: Partial<TaskRecord> = {
|
||||
status: params.status,
|
||||
...(params.childSessionKey !== undefined
|
||||
? { childSessionKey: params.childSessionKey?.trim() || undefined }
|
||||
: {}),
|
||||
endedAt: params.endedAt,
|
||||
lastEventAt: params.lastEventAt ?? params.endedAt,
|
||||
...(params.terminalSummary !== undefined
|
||||
? {
|
||||
terminalSummary: params.preserveTerminalSummary
|
||||
? (params.terminalSummary ?? undefined)
|
||||
: normalizeTaskSummary(params.terminalSummary),
|
||||
}
|
||||
: {}),
|
||||
...(params.terminalOutcome !== undefined
|
||||
? {
|
||||
terminalOutcome: resolveTaskTerminalOutcome({
|
||||
status: params.status,
|
||||
terminalOutcome: params.terminalOutcome,
|
||||
}),
|
||||
}
|
||||
: {}),
|
||||
...(params.detail !== undefined ? { detail: structuredClone(params.detail) } : {}),
|
||||
};
|
||||
if (Object.hasOwn(params, "error")) {
|
||||
patch.error = params.error;
|
||||
}
|
||||
return updateTask(params.taskId, patch);
|
||||
}
|
||||
|
||||
export function markTaskLostById(params: {
|
||||
taskId: string;
|
||||
endedAt: number;
|
||||
lastEventAt?: number;
|
||||
error?: string;
|
||||
cleanupAfter?: number;
|
||||
}): TaskRecord | null {
|
||||
ensureTaskRegistryReady();
|
||||
return updateTask(params.taskId, {
|
||||
status: "lost",
|
||||
endedAt: params.endedAt,
|
||||
lastEventAt: params.lastEventAt ?? params.endedAt,
|
||||
...(params.error !== undefined ? { error: params.error } : {}),
|
||||
...(params.cleanupAfter !== undefined ? { cleanupAfter: params.cleanupAfter } : {}),
|
||||
});
|
||||
}
|
||||
|
||||
function updateTasksByRunId(params: {
|
||||
runId: string;
|
||||
patch: Partial<TaskRecord>;
|
||||
runtime?: TaskRuntime;
|
||||
sessionKey?: string;
|
||||
}): TaskRecord[] {
|
||||
const matches = getTasksByRunScope(params);
|
||||
if (matches.length === 0) {
|
||||
return [];
|
||||
}
|
||||
const updated: TaskRecord[] = [];
|
||||
for (const match of matches) {
|
||||
const task = updateTask(match.taskId, params.patch);
|
||||
if (task) {
|
||||
updated.push(task);
|
||||
}
|
||||
}
|
||||
return updated;
|
||||
}
|
||||
|
||||
export function createTaskRecord(params: {
|
||||
runtime: TaskRuntime;
|
||||
taskKind?: string;
|
||||
sourceId?: string;
|
||||
requesterSessionKey?: string;
|
||||
ownerKey?: string;
|
||||
scopeKind?: TaskScopeKind;
|
||||
requesterOrigin?: TaskDeliveryState["requesterOrigin"];
|
||||
childSessionKey?: string;
|
||||
parentFlowId?: string;
|
||||
parentTaskId?: string;
|
||||
agentId?: string;
|
||||
requesterAgentId?: string;
|
||||
runId?: string;
|
||||
label?: string;
|
||||
task: string;
|
||||
preferMetadata?: boolean;
|
||||
status?: TaskStatus;
|
||||
deliveryStatus?: TaskDeliveryStatus;
|
||||
notifyPolicy?: TaskNotifyPolicy;
|
||||
startedAt?: number;
|
||||
lastEventAt?: number;
|
||||
cleanupAfter?: number;
|
||||
progressSummary?: string | null;
|
||||
terminalSummary?: string | null;
|
||||
terminalOutcome?: TaskTerminalOutcome | null;
|
||||
detail?: JsonValue;
|
||||
}): TaskRecord | null {
|
||||
ensureTaskRegistryReady();
|
||||
const requesterSessionKey = resolveTaskRequesterSessionKey(params);
|
||||
const scopeKind = resolveTaskScopeKind({
|
||||
scopeKind: params.scopeKind,
|
||||
requesterSessionKey,
|
||||
});
|
||||
const ownerKey = resolveTaskOwnerKey({
|
||||
requesterSessionKey,
|
||||
ownerKey: params.ownerKey,
|
||||
});
|
||||
const agentId = resolveTaskAgentId({
|
||||
explicitAgentId: params.agentId,
|
||||
childSessionKey: params.childSessionKey,
|
||||
ownerKey,
|
||||
requesterSessionKey,
|
||||
});
|
||||
const requesterAgentId = resolveTaskRequesterAgentId({
|
||||
explicitRequesterAgentId: params.requesterAgentId,
|
||||
ownerKey,
|
||||
requesterSessionKey,
|
||||
});
|
||||
assertTaskOwner({
|
||||
ownerKey,
|
||||
scopeKind,
|
||||
});
|
||||
assertParentFlowLinkAllowed({
|
||||
ownerKey,
|
||||
scopeKind,
|
||||
parentFlowId: params.parentFlowId,
|
||||
});
|
||||
const existing = findExistingTaskForCreate({
|
||||
runtime: params.runtime,
|
||||
ownerKey,
|
||||
scopeKind,
|
||||
childSessionKey: params.childSessionKey,
|
||||
parentFlowId: params.parentFlowId,
|
||||
runId: params.runId,
|
||||
label: params.label,
|
||||
task: params.task,
|
||||
});
|
||||
if (existing) {
|
||||
return mergeExistingTaskForCreate(existing, { ...params, agentId });
|
||||
}
|
||||
const now = Date.now();
|
||||
const taskId = crypto.randomUUID();
|
||||
const status = normalizeTaskStatus(params.status);
|
||||
const deliveryStatus =
|
||||
params.deliveryStatus ??
|
||||
ensureDeliveryStatus({
|
||||
ownerKey,
|
||||
scopeKind,
|
||||
});
|
||||
const notifyPolicy = ensureNotifyPolicy({
|
||||
notifyPolicy: params.notifyPolicy,
|
||||
deliveryStatus,
|
||||
ownerKey,
|
||||
scopeKind,
|
||||
});
|
||||
const lastEventAt = params.lastEventAt ?? params.startedAt ?? now;
|
||||
const record: TaskRecord = normalizeTaskTimestamps({
|
||||
taskId,
|
||||
runtime: params.runtime,
|
||||
taskKind: normalizeOptionalString(params.taskKind),
|
||||
sourceId: normalizeOptionalString(params.sourceId),
|
||||
requesterSessionKey,
|
||||
ownerKey,
|
||||
scopeKind,
|
||||
childSessionKey: params.childSessionKey,
|
||||
parentFlowId: normalizeOptionalString(params.parentFlowId),
|
||||
parentTaskId: normalizeOptionalString(params.parentTaskId),
|
||||
agentId,
|
||||
requesterAgentId,
|
||||
runId: normalizeOptionalString(params.runId),
|
||||
label: normalizeOptionalString(params.label),
|
||||
task: params.task,
|
||||
status,
|
||||
deliveryStatus,
|
||||
notifyPolicy,
|
||||
createdAt: now,
|
||||
startedAt: params.startedAt,
|
||||
lastEventAt,
|
||||
cleanupAfter: params.cleanupAfter,
|
||||
progressSummary: normalizeTaskSummary(params.progressSummary),
|
||||
terminalSummary: normalizeTaskSummary(params.terminalSummary),
|
||||
terminalOutcome: resolveTaskTerminalOutcome({
|
||||
status,
|
||||
terminalOutcome: params.terminalOutcome,
|
||||
}),
|
||||
...(params.detail !== undefined ? { detail: structuredClone(params.detail) } : {}),
|
||||
});
|
||||
if (isTerminalTaskStatus(record.status) && typeof record.cleanupAfter !== "number") {
|
||||
const cleanupAfter = resolveTaskCleanupAfter(record);
|
||||
Object.assign(record, cleanupAfter === undefined ? {} : { cleanupAfter });
|
||||
}
|
||||
const requesterOrigin = normalizeDeliveryContext(params.requesterOrigin);
|
||||
const deliveryState = requesterOrigin
|
||||
? {
|
||||
taskId,
|
||||
requesterOrigin,
|
||||
}
|
||||
: undefined;
|
||||
if (!tryPersistTaskUpsert(record, "create", deliveryState)) {
|
||||
return null;
|
||||
}
|
||||
tasks.set(taskId, record);
|
||||
if (requesterOrigin) {
|
||||
taskDeliveryStates.set(taskId, deliveryState!);
|
||||
}
|
||||
addRunIdIndex(taskId, record.runId);
|
||||
addOwnerKeyIndex(taskId, record);
|
||||
addParentFlowIdIndex(taskId, record);
|
||||
addRelatedSessionKeyIndex(taskId, record);
|
||||
syncFlowFromTaskAfterTaskMutation(record, "create");
|
||||
emitTaskRegistryObserverEvent(() => ({
|
||||
kind: "upserted",
|
||||
task: cloneTaskRecord(record),
|
||||
}));
|
||||
if (isTerminalTaskStatus(record.status)) {
|
||||
void maybeDeliverTaskTerminalUpdate(taskId);
|
||||
}
|
||||
return cloneTaskRecord(record);
|
||||
}
|
||||
|
||||
export function updateTaskStateByRunId(params: {
|
||||
runId: string;
|
||||
runtime?: TaskRuntime;
|
||||
sessionKey?: string;
|
||||
childSessionKey?: string | null;
|
||||
status?: TaskStatus;
|
||||
startedAt?: number;
|
||||
endedAt?: number;
|
||||
lastEventAt?: number;
|
||||
error?: string;
|
||||
clearError?: boolean;
|
||||
progressSummary?: string | null;
|
||||
terminalSummary?: string | null;
|
||||
preserveTerminalSummary?: boolean;
|
||||
terminalOutcome?: TaskTerminalOutcome | null;
|
||||
detail?: JsonValue;
|
||||
eventSummary?: string | null;
|
||||
suppressDelivery?: boolean;
|
||||
}) {
|
||||
ensureTaskRegistryReady();
|
||||
const matches = getTasksByRunScope(params);
|
||||
if (matches.length === 0) {
|
||||
return [];
|
||||
}
|
||||
const updated: TaskRecord[] = [];
|
||||
for (const current of matches) {
|
||||
const patch: Partial<TaskRecord> = {};
|
||||
const nextStatus = params.status ? normalizeTaskStatus(params.status) : current.status;
|
||||
if (
|
||||
params.status &&
|
||||
!shouldApplyRunScopedStatusUpdate({
|
||||
currentStatus: current.status,
|
||||
currentRuntime: current.runtime,
|
||||
currentChildSessionKey: current.childSessionKey,
|
||||
currentError: current.error,
|
||||
currentEndedAt: current.endedAt,
|
||||
nextStatus,
|
||||
nextError: params.error,
|
||||
nextEndedAt: params.endedAt,
|
||||
})
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
const eventAt = params.lastEventAt ?? params.endedAt ?? Date.now();
|
||||
if (params.status) {
|
||||
patch.status = normalizeTaskStatus(params.status);
|
||||
}
|
||||
if (params.startedAt != null) {
|
||||
patch.startedAt = params.startedAt;
|
||||
}
|
||||
if (params.endedAt != null) {
|
||||
patch.endedAt = params.endedAt;
|
||||
}
|
||||
if (params.lastEventAt != null) {
|
||||
patch.lastEventAt = params.lastEventAt;
|
||||
}
|
||||
if (params.childSessionKey !== undefined) {
|
||||
patch.childSessionKey = params.childSessionKey?.trim() || undefined;
|
||||
}
|
||||
if (params.clearError) {
|
||||
patch.error = undefined;
|
||||
} else if (
|
||||
current.status === "cancelled" &&
|
||||
nextStatus !== "cancelled" &&
|
||||
params.error === undefined
|
||||
) {
|
||||
patch.error = undefined;
|
||||
} else if (params.error !== undefined) {
|
||||
patch.error = params.error;
|
||||
}
|
||||
if (params.progressSummary !== undefined) {
|
||||
patch.progressSummary = normalizeTaskSummary(params.progressSummary);
|
||||
}
|
||||
if (params.terminalSummary !== undefined) {
|
||||
patch.terminalSummary = params.preserveTerminalSummary
|
||||
? (params.terminalSummary ?? undefined)
|
||||
: normalizeTaskSummary(params.terminalSummary);
|
||||
}
|
||||
if (params.terminalOutcome !== undefined) {
|
||||
patch.terminalOutcome = resolveTaskTerminalOutcome({
|
||||
status: nextStatus,
|
||||
terminalOutcome: params.terminalOutcome,
|
||||
});
|
||||
}
|
||||
if (params.detail !== undefined) {
|
||||
patch.detail = params.detail;
|
||||
}
|
||||
if (params.suppressDelivery) {
|
||||
// Teardown suppression must survive redundant lifecycle finalizers that
|
||||
// arrive after queues are cleared, or they can repopulate the stopped session.
|
||||
patch.deliveryStatus = "not_applicable";
|
||||
}
|
||||
const eventSummary =
|
||||
normalizeTaskSummary(params.eventSummary) ??
|
||||
(nextStatus === "failed"
|
||||
? normalizeTaskSummary(params.error ?? current.error)
|
||||
: nextStatus === "succeeded"
|
||||
? normalizeTaskSummary(params.terminalSummary ?? current.terminalSummary)
|
||||
: undefined);
|
||||
const shouldAppendEvent =
|
||||
(params.status && params.status !== current.status) ||
|
||||
Boolean(normalizeTaskSummary(params.eventSummary));
|
||||
const nextEvent = shouldAppendEvent
|
||||
? appendTaskEvent({
|
||||
at: eventAt,
|
||||
kind:
|
||||
params.status && normalizeTaskStatus(params.status) !== current.status
|
||||
? normalizeTaskStatus(params.status)
|
||||
: "progress",
|
||||
summary: eventSummary,
|
||||
})
|
||||
: undefined;
|
||||
const task = updateTask(current.taskId, patch);
|
||||
if (task) {
|
||||
updated.push(task);
|
||||
if (!params.suppressDelivery) {
|
||||
void maybeDeliverTaskStateChangeUpdate(task.taskId, nextEvent);
|
||||
void maybeDeliverTaskTerminalUpdate(task.taskId);
|
||||
}
|
||||
}
|
||||
}
|
||||
return updated;
|
||||
}
|
||||
|
||||
function updateTaskDeliveryByRunId(params: {
|
||||
runId: string;
|
||||
runtime?: TaskRuntime;
|
||||
sessionKey?: string;
|
||||
deliveryStatus: TaskDeliveryStatus;
|
||||
error?: string;
|
||||
}) {
|
||||
ensureTaskRegistryReady();
|
||||
const patch: Partial<TaskRecord> = {
|
||||
deliveryStatus: params.deliveryStatus,
|
||||
};
|
||||
if (params.error !== undefined) {
|
||||
patch.error = params.error;
|
||||
}
|
||||
return updateTasksByRunId({
|
||||
runId: params.runId,
|
||||
runtime: params.runtime,
|
||||
sessionKey: params.sessionKey,
|
||||
patch,
|
||||
});
|
||||
}
|
||||
|
||||
export function markTaskRunningByRunId(params: {
|
||||
runId: string;
|
||||
runtime?: TaskRuntime;
|
||||
sessionKey?: string;
|
||||
startedAt?: number;
|
||||
lastEventAt?: number;
|
||||
progressSummary?: string | null;
|
||||
eventSummary?: string | null;
|
||||
}) {
|
||||
return updateTaskStateByRunId({
|
||||
runId: params.runId,
|
||||
runtime: params.runtime,
|
||||
sessionKey: params.sessionKey,
|
||||
status: "running",
|
||||
startedAt: params.startedAt,
|
||||
lastEventAt: params.lastEventAt,
|
||||
progressSummary: params.progressSummary,
|
||||
eventSummary: params.eventSummary,
|
||||
});
|
||||
}
|
||||
|
||||
export function recordTaskProgressByRunId(params: {
|
||||
runId: string;
|
||||
runtime?: TaskRuntime;
|
||||
sessionKey?: string;
|
||||
lastEventAt?: number;
|
||||
progressSummary?: string | null;
|
||||
eventSummary?: string | null;
|
||||
}) {
|
||||
return updateTaskStateByRunId({
|
||||
runId: params.runId,
|
||||
runtime: params.runtime,
|
||||
sessionKey: params.sessionKey,
|
||||
lastEventAt: params.lastEventAt,
|
||||
progressSummary: params.progressSummary,
|
||||
eventSummary: params.eventSummary,
|
||||
});
|
||||
}
|
||||
|
||||
export function finalizeTaskRunByRunId(params: {
|
||||
runId: string;
|
||||
runtime?: TaskRuntime;
|
||||
sessionKey?: string;
|
||||
childSessionKey?: string | null;
|
||||
status: Extract<TaskStatus, "succeeded" | "failed" | "timed_out" | "cancelled">;
|
||||
startedAt?: number;
|
||||
endedAt: number;
|
||||
lastEventAt?: number;
|
||||
error?: string;
|
||||
clearError?: boolean;
|
||||
progressSummary?: string | null;
|
||||
terminalSummary?: string | null;
|
||||
preserveTerminalSummary?: boolean;
|
||||
terminalOutcome?: TaskTerminalOutcome | null;
|
||||
detail?: JsonValue;
|
||||
suppressDelivery?: boolean;
|
||||
}) {
|
||||
return updateTaskStateByRunId({
|
||||
runId: params.runId,
|
||||
runtime: params.runtime,
|
||||
sessionKey: params.sessionKey,
|
||||
childSessionKey: params.childSessionKey,
|
||||
status: params.status,
|
||||
startedAt: params.startedAt,
|
||||
endedAt: params.endedAt,
|
||||
lastEventAt: params.lastEventAt,
|
||||
error: params.error,
|
||||
clearError: params.clearError,
|
||||
progressSummary: params.progressSummary,
|
||||
terminalSummary: params.terminalSummary,
|
||||
preserveTerminalSummary: params.preserveTerminalSummary,
|
||||
terminalOutcome: params.terminalOutcome,
|
||||
detail: params.detail,
|
||||
suppressDelivery: params.suppressDelivery,
|
||||
});
|
||||
}
|
||||
|
||||
export function setTaskRunDeliveryStatusByRunId(params: {
|
||||
runId: string;
|
||||
runtime?: TaskRuntime;
|
||||
sessionKey?: string;
|
||||
deliveryStatus: TaskDeliveryStatus;
|
||||
error?: string;
|
||||
}) {
|
||||
return updateTaskDeliveryByRunId(params);
|
||||
}
|
||||
|
||||
export function updateTaskNotifyPolicyById(params: {
|
||||
taskId: string;
|
||||
notifyPolicy: TaskNotifyPolicy;
|
||||
}): TaskRecord | null {
|
||||
ensureTaskRegistryReady();
|
||||
return updateTask(params.taskId, {
|
||||
notifyPolicy: params.notifyPolicy,
|
||||
lastEventAt: Date.now(),
|
||||
});
|
||||
}
|
||||
|
||||
export function linkTaskToFlowById(params: { taskId: string; flowId: string }): TaskRecord | null {
|
||||
ensureTaskRegistryReady();
|
||||
const flowId = params.flowId.trim();
|
||||
if (!flowId) {
|
||||
return null;
|
||||
}
|
||||
const current = tasks.get(params.taskId);
|
||||
if (!current) {
|
||||
return null;
|
||||
}
|
||||
if (current.parentFlowId?.trim()) {
|
||||
return cloneTaskRecord(current);
|
||||
}
|
||||
assertParentFlowLinkAllowed({
|
||||
ownerKey: current.ownerKey,
|
||||
scopeKind: current.scopeKind,
|
||||
parentFlowId: flowId,
|
||||
});
|
||||
return updateTask(params.taskId, {
|
||||
parentFlowId: flowId,
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,558 @@
|
||||
import { createRequire } from "node:module";
|
||||
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
|
||||
import { uniqueStrings } from "@openclaw/normalization-core/string-normalization";
|
||||
import { formatErrorMessage } from "../infra/errors.js";
|
||||
import { createSubsystemLogger } from "../logging/subsystem.js";
|
||||
import { createLazyPromiseLoader } from "../shared/lazy-runtime.js";
|
||||
import type { TaskRegistryControlRuntime } from "./task-registry-control.types.js";
|
||||
import {
|
||||
cloneTaskDeliveryState,
|
||||
cloneTaskRecord,
|
||||
normalizeTaskTimestamps,
|
||||
} from "./task-registry-records.js";
|
||||
import { getTaskRegistryProcessState } from "./task-registry.process-state.js";
|
||||
import {
|
||||
getTaskRegistryObservers,
|
||||
getTaskRegistryStore,
|
||||
type TaskRegistryObserverEvent,
|
||||
} from "./task-registry.store.js";
|
||||
import type { TaskDeliveryState, TaskRecord, TaskRuntime } from "./task-registry.types.js";
|
||||
|
||||
export const log = createSubsystemLogger("tasks/registry");
|
||||
export const TASK_FLOW_SYNC_RETRY_DELAYS_MS = [1_000, 5_000, 25_000, 120_000, 600_000] as const;
|
||||
|
||||
const taskRegistryProcessState = getTaskRegistryProcessState();
|
||||
export const tasks = taskRegistryProcessState.tasks;
|
||||
export const taskDeliveryStates = taskRegistryProcessState.taskDeliveryStates;
|
||||
const taskIdsByRunId = taskRegistryProcessState.taskIdsByRunId;
|
||||
export const taskIdsByOwnerKey = taskRegistryProcessState.taskIdsByOwnerKey;
|
||||
export const taskIdsByParentFlowId = taskRegistryProcessState.taskIdsByParentFlowId;
|
||||
export const taskIdsByRelatedSessionKey = taskRegistryProcessState.taskIdsByRelatedSessionKey;
|
||||
export const tasksWithPendingDelivery = taskRegistryProcessState.tasksWithPendingDelivery;
|
||||
let listenerStarted = false;
|
||||
let listenerStop: (() => void) | null = null;
|
||||
type TaskRegistryRestoreState =
|
||||
| { status: "uninitialized" }
|
||||
| { status: "restoring" }
|
||||
| { status: "ready" }
|
||||
| { status: "failed"; error: Error };
|
||||
let taskRegistryRestoreState: TaskRegistryRestoreState = { status: "uninitialized" };
|
||||
export const taskFlowSyncRetryTimers = new Map<string, ReturnType<typeof setTimeout>>();
|
||||
export type TaskRegistryDeliveryRuntime = Pick<
|
||||
typeof import("./task-registry-delivery-runtime.js"),
|
||||
"sendMessage"
|
||||
>;
|
||||
export const TASK_REGISTRY_DELIVERY_RUNTIME_OVERRIDE_KEY = Symbol.for(
|
||||
"openclaw.taskRegistry.deliveryRuntimeOverride",
|
||||
);
|
||||
export const TASK_REGISTRY_CONTROL_RUNTIME_OVERRIDE_KEY = Symbol.for(
|
||||
"openclaw.taskRegistry.controlRuntimeOverride",
|
||||
);
|
||||
const require = createRequire(import.meta.url);
|
||||
const TASK_REGISTRY_CONTROL_RUNTIME_CANDIDATES = [
|
||||
"./task-registry-control.runtime.js",
|
||||
"./task-registry-control.runtime.ts",
|
||||
] as const;
|
||||
export type TaskRegistryGlobalWithRuntimeOverrides = typeof globalThis & {
|
||||
[TASK_REGISTRY_DELIVERY_RUNTIME_OVERRIDE_KEY]?: TaskRegistryDeliveryRuntime | null;
|
||||
[TASK_REGISTRY_CONTROL_RUNTIME_OVERRIDE_KEY]?: TaskRegistryControlRuntime | null;
|
||||
};
|
||||
export const deliveryRuntimeLoader = createLazyPromiseLoader(
|
||||
() => import("./task-registry-delivery-runtime.js"),
|
||||
{ cacheRejections: true },
|
||||
);
|
||||
export const controlRuntimeLoader = createLazyPromiseLoader(
|
||||
() =>
|
||||
Promise.resolve().then(() => {
|
||||
for (const candidate of TASK_REGISTRY_CONTROL_RUNTIME_CANDIDATES) {
|
||||
try {
|
||||
return require(candidate) as TaskRegistryControlRuntime;
|
||||
} catch {
|
||||
// Try runtime/source candidates in order.
|
||||
}
|
||||
}
|
||||
throw new Error("Failed to load task registry control runtime.");
|
||||
}),
|
||||
{ cacheRejections: true },
|
||||
);
|
||||
|
||||
let listenerStarter: () => void = () => {};
|
||||
|
||||
export function setTaskRegistryListenerStarter(starter: () => void): void {
|
||||
listenerStarter = starter;
|
||||
}
|
||||
|
||||
export function claimTaskRegistryListenerStart(): boolean {
|
||||
if (listenerStarted) {
|
||||
return false;
|
||||
}
|
||||
listenerStarted = true;
|
||||
return true;
|
||||
}
|
||||
|
||||
export function setTaskRegistryListenerStop(stop: (() => void) | null): void {
|
||||
listenerStop = stop;
|
||||
}
|
||||
|
||||
export function resetTaskRegistryListenerState(): void {
|
||||
listenerStop?.();
|
||||
listenerStop = null;
|
||||
listenerStarted = false;
|
||||
}
|
||||
|
||||
function clearTaskFlowSyncRetries(): void {
|
||||
for (const timer of taskFlowSyncRetryTimers.values()) {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
taskFlowSyncRetryTimers.clear();
|
||||
}
|
||||
|
||||
export function snapshotTaskRecords(source: ReadonlyMap<string, TaskRecord>): TaskRecord[] {
|
||||
return [...source.values()].map((record) => cloneTaskRecord(record));
|
||||
}
|
||||
|
||||
export function emitTaskRegistryObserverEvent(createEvent: () => TaskRegistryObserverEvent): void {
|
||||
const observers = getTaskRegistryObservers();
|
||||
if (!observers?.onEvent) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
observers.onEvent(createEvent());
|
||||
} catch (error) {
|
||||
log.warn("Task registry observer failed", {
|
||||
event: "task-registry",
|
||||
error,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export function persistTaskRegistry(): boolean {
|
||||
try {
|
||||
getTaskRegistryStore().saveSnapshot({
|
||||
tasks,
|
||||
deliveryStates: taskDeliveryStates,
|
||||
});
|
||||
return true;
|
||||
} catch (error) {
|
||||
log.warn("Failed to persist task registry snapshot", { error });
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function persistTaskUpsert(task: TaskRecord, pendingDeliveryState?: TaskDeliveryState): void {
|
||||
const store = getTaskRegistryStore();
|
||||
const deliveryState = pendingDeliveryState ?? taskDeliveryStates.get(task.taskId);
|
||||
if (store.upsertTaskWithDeliveryState) {
|
||||
store.upsertTaskWithDeliveryState({
|
||||
task,
|
||||
...(deliveryState ? { deliveryState } : {}),
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (!deliveryState && store.upsertTask) {
|
||||
store.upsertTask(task);
|
||||
return;
|
||||
}
|
||||
// Snapshot fallback: project the pending upsert so the snapshot is correct
|
||||
// even though we persist before mutating memory. Delivery state must stay in
|
||||
// the same write as its task; split upserts can leave a durable half-create.
|
||||
store.saveSnapshot({
|
||||
tasks: new Map(tasks).set(task.taskId, task),
|
||||
deliveryStates: deliveryState
|
||||
? new Map(taskDeliveryStates).set(task.taskId, deliveryState)
|
||||
: taskDeliveryStates,
|
||||
});
|
||||
}
|
||||
|
||||
export function tryPersistTaskUpsert(
|
||||
task: TaskRecord,
|
||||
operation: string,
|
||||
pendingDeliveryState?: TaskDeliveryState,
|
||||
): boolean {
|
||||
try {
|
||||
persistTaskUpsert(task, pendingDeliveryState);
|
||||
return true;
|
||||
} catch (error) {
|
||||
log.warn("Failed to persist task registry upsert", {
|
||||
operation,
|
||||
taskId: task.taskId,
|
||||
runId: task.runId,
|
||||
error,
|
||||
});
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function persistTaskDelete(taskId: string) {
|
||||
const store = getTaskRegistryStore();
|
||||
if (store.deleteTaskWithDeliveryState) {
|
||||
// Composite delete removes the task row and its delivery state in a single
|
||||
// transaction. This is the only atomic "remove both records" store
|
||||
// primitive, and the one the default sqlite store uses.
|
||||
store.deleteTaskWithDeliveryState(taskId);
|
||||
return;
|
||||
}
|
||||
// No atomic composite delete is available: persist the removal of BOTH the
|
||||
// task and its delivery state in one projected snapshot. saveSnapshot is a
|
||||
// required store method and writes atomically. Using the separate deleteTask
|
||||
// / deleteDeliveryState methods instead would either leave the delivery-state
|
||||
// row behind (a task-only delete) or, if both were called, reintroduce a
|
||||
// two-write divergence window when the second delete threw before the
|
||||
// in-memory mutation. Projecting both deletions into a single snapshot keeps
|
||||
// the persisted store consistent under the persist-before-in-memory ordering.
|
||||
const projectedTasks = new Map(tasks);
|
||||
projectedTasks.delete(taskId);
|
||||
const projectedDeliveryStates = new Map(taskDeliveryStates);
|
||||
projectedDeliveryStates.delete(taskId);
|
||||
store.saveSnapshot({
|
||||
tasks: projectedTasks,
|
||||
deliveryStates: projectedDeliveryStates,
|
||||
});
|
||||
}
|
||||
|
||||
export function tryPersistTaskDelete(taskId: string): boolean {
|
||||
try {
|
||||
persistTaskDelete(taskId);
|
||||
return true;
|
||||
} catch (error) {
|
||||
log.warn("Failed to persist task registry delete", {
|
||||
taskId,
|
||||
error,
|
||||
});
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function persistTaskDeliveryStateUpsert(state: TaskDeliveryState) {
|
||||
const store = getTaskRegistryStore();
|
||||
if (store.upsertDeliveryState) {
|
||||
store.upsertDeliveryState(state);
|
||||
return;
|
||||
}
|
||||
const projectedDeliveryStates = new Map(taskDeliveryStates);
|
||||
projectedDeliveryStates.set(state.taskId, cloneTaskDeliveryState(state));
|
||||
store.saveSnapshot({
|
||||
tasks,
|
||||
deliveryStates: projectedDeliveryStates,
|
||||
});
|
||||
}
|
||||
|
||||
export function tryPersistTaskDeliveryStateUpsert(state: TaskDeliveryState): boolean {
|
||||
try {
|
||||
persistTaskDeliveryStateUpsert(state);
|
||||
return true;
|
||||
} catch (error) {
|
||||
log.warn("Failed to persist task delivery state", {
|
||||
taskId: state.taskId,
|
||||
error,
|
||||
});
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
export function clearTaskRegistryMemory(): void {
|
||||
clearTaskFlowSyncRetries();
|
||||
tasks.clear();
|
||||
taskDeliveryStates.clear();
|
||||
taskIdsByRunId.clear();
|
||||
taskIdsByOwnerKey.clear();
|
||||
taskIdsByParentFlowId.clear();
|
||||
taskIdsByRelatedSessionKey.clear();
|
||||
tasksWithPendingDelivery.clear();
|
||||
}
|
||||
|
||||
export function loadTaskRegistryDeliveryRuntime() {
|
||||
const deliveryRuntimeOverride = (globalThis as TaskRegistryGlobalWithRuntimeOverrides)[
|
||||
TASK_REGISTRY_DELIVERY_RUNTIME_OVERRIDE_KEY
|
||||
];
|
||||
if (deliveryRuntimeOverride) {
|
||||
return Promise.resolve(deliveryRuntimeOverride);
|
||||
}
|
||||
return deliveryRuntimeLoader.load();
|
||||
}
|
||||
|
||||
export function loadTaskRegistryControlRuntime() {
|
||||
const controlRuntimeOverride = (globalThis as TaskRegistryGlobalWithRuntimeOverrides)[
|
||||
TASK_REGISTRY_CONTROL_RUNTIME_OVERRIDE_KEY
|
||||
];
|
||||
if (controlRuntimeOverride) {
|
||||
return Promise.resolve(controlRuntimeOverride);
|
||||
}
|
||||
// Registry reads happen far more often than task cancellation, so keep the ACP/subagent
|
||||
// control graph off the default import path until a cancellation flow actually needs it.
|
||||
return controlRuntimeLoader.load();
|
||||
}
|
||||
|
||||
export function addRunIdIndex(taskId: string, runId?: string) {
|
||||
const trimmed = runId?.trim();
|
||||
if (!trimmed) {
|
||||
return;
|
||||
}
|
||||
let ids = taskIdsByRunId.get(trimmed);
|
||||
if (!ids) {
|
||||
ids = new Set<string>();
|
||||
taskIdsByRunId.set(trimmed, ids);
|
||||
}
|
||||
ids.add(taskId);
|
||||
}
|
||||
|
||||
function addIndexedKey(index: Map<string, Set<string>>, key: string, taskId: string) {
|
||||
let ids = index.get(key);
|
||||
if (!ids) {
|
||||
ids = new Set<string>();
|
||||
index.set(key, ids);
|
||||
}
|
||||
ids.add(taskId);
|
||||
}
|
||||
|
||||
function deleteIndexedKey(index: Map<string, Set<string>>, key: string, taskId: string) {
|
||||
const ids = index.get(key);
|
||||
if (!ids) {
|
||||
return;
|
||||
}
|
||||
ids.delete(taskId);
|
||||
if (ids.size === 0) {
|
||||
index.delete(key);
|
||||
}
|
||||
}
|
||||
|
||||
function getTaskRelatedSessionIndexKeys(task: Pick<TaskRecord, "ownerKey" | "childSessionKey">) {
|
||||
return uniqueStrings(
|
||||
[normalizeOptionalString(task.ownerKey), normalizeOptionalString(task.childSessionKey)].filter(
|
||||
Boolean,
|
||||
) as string[],
|
||||
);
|
||||
}
|
||||
|
||||
export function addOwnerKeyIndex(taskId: string, task: Pick<TaskRecord, "ownerKey">) {
|
||||
const key = normalizeOptionalString(task.ownerKey);
|
||||
if (!key) {
|
||||
return;
|
||||
}
|
||||
addIndexedKey(taskIdsByOwnerKey, key, taskId);
|
||||
}
|
||||
|
||||
export function deleteOwnerKeyIndex(taskId: string, task: Pick<TaskRecord, "ownerKey">) {
|
||||
const key = normalizeOptionalString(task.ownerKey);
|
||||
if (!key) {
|
||||
return;
|
||||
}
|
||||
deleteIndexedKey(taskIdsByOwnerKey, key, taskId);
|
||||
}
|
||||
|
||||
export function addParentFlowIdIndex(taskId: string, task: Pick<TaskRecord, "parentFlowId">) {
|
||||
const key = task.parentFlowId?.trim();
|
||||
if (!key) {
|
||||
return;
|
||||
}
|
||||
addIndexedKey(taskIdsByParentFlowId, key, taskId);
|
||||
}
|
||||
|
||||
export function deleteParentFlowIdIndex(taskId: string, task: Pick<TaskRecord, "parentFlowId">) {
|
||||
const key = task.parentFlowId?.trim();
|
||||
if (!key) {
|
||||
return;
|
||||
}
|
||||
deleteIndexedKey(taskIdsByParentFlowId, key, taskId);
|
||||
}
|
||||
|
||||
export function addRelatedSessionKeyIndex(
|
||||
taskId: string,
|
||||
task: Pick<TaskRecord, "ownerKey" | "childSessionKey">,
|
||||
) {
|
||||
for (const sessionKey of getTaskRelatedSessionIndexKeys(task)) {
|
||||
addIndexedKey(taskIdsByRelatedSessionKey, sessionKey, taskId);
|
||||
}
|
||||
}
|
||||
|
||||
export function deleteRelatedSessionKeyIndex(
|
||||
taskId: string,
|
||||
task: Pick<TaskRecord, "ownerKey" | "childSessionKey">,
|
||||
) {
|
||||
for (const sessionKey of getTaskRelatedSessionIndexKeys(task)) {
|
||||
deleteIndexedKey(taskIdsByRelatedSessionKey, sessionKey, taskId);
|
||||
}
|
||||
}
|
||||
|
||||
export function rebuildRunIdIndex() {
|
||||
taskIdsByRunId.clear();
|
||||
for (const [taskId, task] of tasks.entries()) {
|
||||
addRunIdIndex(taskId, task.runId);
|
||||
}
|
||||
}
|
||||
|
||||
function rebuildOwnerKeyIndex() {
|
||||
taskIdsByOwnerKey.clear();
|
||||
for (const [taskId, task] of tasks.entries()) {
|
||||
addOwnerKeyIndex(taskId, task);
|
||||
}
|
||||
}
|
||||
|
||||
function rebuildParentFlowIdIndex() {
|
||||
taskIdsByParentFlowId.clear();
|
||||
for (const [taskId, task] of tasks.entries()) {
|
||||
addParentFlowIdIndex(taskId, task);
|
||||
}
|
||||
}
|
||||
|
||||
function rebuildRelatedSessionKeyIndex() {
|
||||
taskIdsByRelatedSessionKey.clear();
|
||||
for (const [taskId, task] of tasks.entries()) {
|
||||
addRelatedSessionKeyIndex(taskId, task);
|
||||
}
|
||||
}
|
||||
|
||||
export function getTasksByRunId(runId: string): TaskRecord[] {
|
||||
const ids = taskIdsByRunId.get(runId.trim());
|
||||
if (!ids || ids.size === 0) {
|
||||
return [];
|
||||
}
|
||||
return [...ids]
|
||||
.map((taskId) => tasks.get(taskId))
|
||||
.filter((task): task is TaskRecord => Boolean(task));
|
||||
}
|
||||
|
||||
function taskRunScopeKey(
|
||||
task: Pick<TaskRecord, "runtime" | "scopeKind" | "ownerKey" | "childSessionKey">,
|
||||
): string {
|
||||
return [
|
||||
task.runtime,
|
||||
task.scopeKind,
|
||||
normalizeOptionalString(task.ownerKey) ?? "",
|
||||
normalizeOptionalString(task.childSessionKey) ?? "",
|
||||
].join("\u0000");
|
||||
}
|
||||
|
||||
export function getTasksByRunScope(params: {
|
||||
runId: string;
|
||||
runtime?: TaskRuntime;
|
||||
sessionKey?: string;
|
||||
}): TaskRecord[] {
|
||||
const matches = getTasksByRunId(params.runId).filter(
|
||||
(task) => !params.runtime || task.runtime === params.runtime,
|
||||
);
|
||||
const sessionKey = normalizeOptionalString(params.sessionKey);
|
||||
if (sessionKey) {
|
||||
const childMatches = matches.filter(
|
||||
(task) => normalizeOptionalString(task.childSessionKey) === sessionKey,
|
||||
);
|
||||
if (childMatches.length > 0) {
|
||||
return childMatches;
|
||||
}
|
||||
const ownerMatches = matches.filter(
|
||||
(task) =>
|
||||
task.scopeKind === "session" && normalizeOptionalString(task.ownerKey) === sessionKey,
|
||||
);
|
||||
return ownerMatches;
|
||||
}
|
||||
const scopeKeys = new Set(matches.map((task) => taskRunScopeKey(task)));
|
||||
return scopeKeys.size <= 1 ? matches : [];
|
||||
}
|
||||
|
||||
export function getPeerTasksForDelivery(task: TaskRecord): TaskRecord[] {
|
||||
if (!task.runId?.trim()) {
|
||||
return [];
|
||||
}
|
||||
return getTasksByRunId(task.runId).filter(
|
||||
(candidate) =>
|
||||
candidate.runtime === task.runtime &&
|
||||
candidate.scopeKind === task.scopeKind &&
|
||||
(normalizeOptionalString(candidate.ownerKey) ?? "") ===
|
||||
(normalizeOptionalString(task.ownerKey) ?? "") &&
|
||||
(normalizeOptionalString(candidate.childSessionKey) ?? "") ===
|
||||
(normalizeOptionalString(task.childSessionKey) ?? ""),
|
||||
);
|
||||
}
|
||||
|
||||
function taskLookupPriority(task: TaskRecord): number {
|
||||
const runtimePriority = task.runtime === "cli" ? 1 : 0;
|
||||
return runtimePriority;
|
||||
}
|
||||
|
||||
export function pickPreferredRunIdTask(matches: TaskRecord[]): TaskRecord | undefined {
|
||||
return [...matches].toSorted((left, right) => {
|
||||
const priorityDiff = taskLookupPriority(left) - taskLookupPriority(right);
|
||||
if (priorityDiff !== 0) {
|
||||
return priorityDiff;
|
||||
}
|
||||
return left.createdAt - right.createdAt;
|
||||
})[0];
|
||||
}
|
||||
|
||||
export function compareTasksNewestFirst(
|
||||
left: Pick<TaskRecord, "createdAt"> & { insertionIndex?: number },
|
||||
right: Pick<TaskRecord, "createdAt"> & { insertionIndex?: number },
|
||||
): number {
|
||||
const createdAtDiff = right.createdAt - left.createdAt;
|
||||
if (createdAtDiff !== 0) {
|
||||
return createdAtDiff;
|
||||
}
|
||||
return (right.insertionIndex ?? 0) - (left.insertionIndex ?? 0);
|
||||
}
|
||||
|
||||
export function restoreTaskRegistryOnce() {
|
||||
switch (taskRegistryRestoreState.status) {
|
||||
case "ready":
|
||||
return;
|
||||
case "failed":
|
||||
throw taskRegistryRestoreState.error;
|
||||
case "restoring":
|
||||
throw new Error("Task registry restore is already in progress.");
|
||||
case "uninitialized":
|
||||
break;
|
||||
}
|
||||
taskRegistryRestoreState = { status: "restoring" };
|
||||
try {
|
||||
const restored = getTaskRegistryStore().loadSnapshot();
|
||||
const restoredTasks = new Map<string, TaskRecord>();
|
||||
for (const [taskId, task] of restored.tasks.entries()) {
|
||||
restoredTasks.set(taskId, normalizeTaskTimestamps(task));
|
||||
}
|
||||
const restoredDeliveryStates = new Map(restored.deliveryStates);
|
||||
|
||||
clearTaskRegistryMemory();
|
||||
for (const [taskId, task] of restoredTasks.entries()) {
|
||||
tasks.set(taskId, task);
|
||||
}
|
||||
for (const [taskId, state] of restoredDeliveryStates.entries()) {
|
||||
taskDeliveryStates.set(taskId, state);
|
||||
}
|
||||
rebuildRunIdIndex();
|
||||
rebuildOwnerKeyIndex();
|
||||
rebuildParentFlowIdIndex();
|
||||
rebuildRelatedSessionKeyIndex();
|
||||
taskRegistryRestoreState = { status: "ready" };
|
||||
if (restoredTasks.size > 0 || restoredDeliveryStates.size > 0) {
|
||||
emitTaskRegistryObserverEvent(() => ({
|
||||
kind: "restored",
|
||||
tasks: snapshotTaskRecords(tasks),
|
||||
}));
|
||||
}
|
||||
} catch (error) {
|
||||
clearTaskRegistryMemory();
|
||||
const message = formatErrorMessage(error);
|
||||
const restoreError = new Error(`Task registry restore failed: ${message}`, { cause: error });
|
||||
taskRegistryRestoreState = { status: "failed", error: restoreError };
|
||||
// Compact console logs omit structured metadata, so keep the rejected value visible there too.
|
||||
log.warn("Failed to restore task registry", {
|
||||
error: message,
|
||||
consoleMessage: `Failed to restore task registry: ${message}`,
|
||||
});
|
||||
throw restoreError;
|
||||
}
|
||||
}
|
||||
|
||||
export function ensureTaskRegistryReady(): void {
|
||||
restoreTaskRegistryOnce();
|
||||
listenerStarter();
|
||||
}
|
||||
|
||||
export function reloadTaskRegistryFromStore(): void {
|
||||
clearTaskRegistryMemory();
|
||||
taskRegistryRestoreState = { status: "uninitialized" };
|
||||
ensureTaskRegistryReady();
|
||||
}
|
||||
|
||||
export function resetTaskRegistryRestoreState(): void {
|
||||
taskRegistryRestoreState = { status: "uninitialized" };
|
||||
}
|
||||
+41
-2798
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user