|
| 1 | +import { promises as fs } from "node:fs"; |
| 2 | +import path from "node:path"; |
| 3 | +import { |
| 4 | + resolveAgentWorkspaceDir, |
| 5 | + resolveDefaultAgentId, |
| 6 | +} from "openclaw/plugin-sdk/memory-core-host-engine-foundation"; |
| 7 | +import type { OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry"; |
| 8 | +import { |
| 9 | + addTaskRegistryEventListener, |
| 10 | + type TaskRecord, |
| 11 | + type TaskRegistryObserverEvent, |
| 12 | + type TaskStatus, |
| 13 | +} from "openclaw/plugin-sdk/task-events"; |
| 14 | + |
| 15 | +const TASK_OUTCOME_LOG_RELATIVE_PATH = path.join("memory", ".dreams", "task-outcomes.jsonl"); |
| 16 | + |
| 17 | +const TERMINAL_STATUSES = new Set<TaskStatus>([ |
| 18 | + "succeeded", |
| 19 | + "failed", |
| 20 | + "timed_out", |
| 21 | + "cancelled", |
| 22 | + "lost", |
| 23 | +]); |
| 24 | + |
| 25 | +type TaskOutcomeRecord = { |
| 26 | + type: "task.outcome"; |
| 27 | + timestamp: string; |
| 28 | + taskId: string; |
| 29 | + status: TaskStatus; |
| 30 | + runtime: TaskRecord["runtime"]; |
| 31 | + agentId?: string; |
| 32 | + taskKind?: string; |
| 33 | + label?: string; |
| 34 | + summary: string; |
| 35 | + durationMs?: number; |
| 36 | + error?: string; |
| 37 | +}; |
| 38 | + |
| 39 | +const MAX_LABEL_LENGTH = 240; |
| 40 | +const MAX_ERROR_LENGTH = 480; |
| 41 | + |
| 42 | +function shouldRecord( |
| 43 | + event: TaskRegistryObserverEvent, |
| 44 | +): { task: TaskRecord; previous?: TaskRecord } | null { |
| 45 | + if (event.kind !== "upserted") { |
| 46 | + return null; |
| 47 | + } |
| 48 | + if (!TERMINAL_STATUSES.has(event.task.status)) { |
| 49 | + return null; |
| 50 | + } |
| 51 | + // Only record state transitions; if the task was already terminal we have |
| 52 | + // already logged the outcome on the original event. |
| 53 | + if (event.previous && TERMINAL_STATUSES.has(event.previous.status)) { |
| 54 | + return null; |
| 55 | + } |
| 56 | + return { task: event.task, previous: event.previous }; |
| 57 | +} |
| 58 | + |
| 59 | +function truncate(value: string | undefined, max: number): string | undefined { |
| 60 | + if (!value) { |
| 61 | + return undefined; |
| 62 | + } |
| 63 | + return value.length > max ? `${value.slice(0, max - 1)}…` : value; |
| 64 | +} |
| 65 | + |
| 66 | +function summarizeOutcome(task: TaskRecord): string { |
| 67 | + const label = |
| 68 | + truncate(task.label?.trim(), MAX_LABEL_LENGTH) ?? |
| 69 | + truncate(task.task.split("\n", 1)[0]?.trim(), MAX_LABEL_LENGTH) ?? |
| 70 | + task.taskId; |
| 71 | + if (task.status === "succeeded") { |
| 72 | + const trailing = task.terminalSummary ? ` — ${truncate(task.terminalSummary, 240)}` : ""; |
| 73 | + return `Task succeeded: ${label}${trailing}`; |
| 74 | + } |
| 75 | + const error = task.error ? ` — ${truncate(task.error, MAX_ERROR_LENGTH)}` : ""; |
| 76 | + return `Task ${task.status}: ${label}${error}`; |
| 77 | +} |
| 78 | + |
| 79 | +function buildOutcomeRecord(task: TaskRecord, nowMs: number): TaskOutcomeRecord { |
| 80 | + return { |
| 81 | + type: "task.outcome", |
| 82 | + timestamp: new Date(nowMs).toISOString(), |
| 83 | + taskId: task.taskId, |
| 84 | + status: task.status, |
| 85 | + runtime: task.runtime, |
| 86 | + ...(task.agentId ? { agentId: task.agentId } : {}), |
| 87 | + ...(task.taskKind ? { taskKind: task.taskKind } : {}), |
| 88 | + ...(task.label ? { label: truncate(task.label, MAX_LABEL_LENGTH) } : {}), |
| 89 | + summary: summarizeOutcome(task), |
| 90 | + ...(task.startedAt && task.endedAt |
| 91 | + ? { durationMs: Math.max(0, task.endedAt - task.startedAt) } |
| 92 | + : {}), |
| 93 | + ...(task.error ? { error: truncate(task.error, MAX_ERROR_LENGTH) } : {}), |
| 94 | + }; |
| 95 | +} |
| 96 | + |
| 97 | +async function appendTaskOutcomeRecord( |
| 98 | + workspaceDir: string, |
| 99 | + record: TaskOutcomeRecord, |
| 100 | +): Promise<void> { |
| 101 | + const target = path.join(workspaceDir, TASK_OUTCOME_LOG_RELATIVE_PATH); |
| 102 | + await fs.mkdir(path.dirname(target), { recursive: true }); |
| 103 | + await fs.appendFile(target, `${JSON.stringify(record)}\n`, "utf8"); |
| 104 | +} |
| 105 | + |
| 106 | +/** |
| 107 | + * Record terminal task outcomes (succeeded, failed, timed_out, cancelled, lost) |
| 108 | + * to the agent's workspace as JSONL events. Downstream dreaming can read these |
| 109 | + * to grow MEMORY entries about what worked and what didn't, closing the |
| 110 | + * task → memory feedback loop. |
| 111 | + * |
| 112 | + * Returns an unsubscribe function so the listener can be cleaned up in tests |
| 113 | + * or during plugin reloads. |
| 114 | + */ |
| 115 | +export function registerTaskOutcomeRecorder(api: OpenClawPluginApi): () => void { |
| 116 | + const log = api.logger; |
| 117 | + const unsubscribe = addTaskRegistryEventListener((event) => { |
| 118 | + const recordable = shouldRecord(event); |
| 119 | + if (!recordable) { |
| 120 | + return; |
| 121 | + } |
| 122 | + const { task } = recordable; |
| 123 | + const agentId = task.agentId ?? resolveDefaultAgentId(api.config); |
| 124 | + if (!agentId) { |
| 125 | + return; |
| 126 | + } |
| 127 | + let workspaceDir: string; |
| 128 | + try { |
| 129 | + workspaceDir = resolveAgentWorkspaceDir(api.config, agentId); |
| 130 | + } catch (error) { |
| 131 | + log.debug?.( |
| 132 | + `task-outcome-recorder: workspace resolution failed for agent ${agentId}: ${String(error)}`, |
| 133 | + ); |
| 134 | + return; |
| 135 | + } |
| 136 | + if (!workspaceDir) { |
| 137 | + return; |
| 138 | + } |
| 139 | + const record = buildOutcomeRecord(task, Date.now()); |
| 140 | + void appendTaskOutcomeRecord(workspaceDir, record).catch((error: unknown) => { |
| 141 | + log.debug?.(`task-outcome-recorder: append failed for ${task.taskId}: ${String(error)}`); |
| 142 | + }); |
| 143 | + }); |
| 144 | + return unsubscribe; |
| 145 | +} |
| 146 | + |
| 147 | +export const __testing = { |
| 148 | + buildOutcomeRecord, |
| 149 | + shouldRecord, |
| 150 | + summarizeOutcome, |
| 151 | + TASK_OUTCOME_LOG_RELATIVE_PATH, |
| 152 | +}; |
0 commit comments