From 7d2189cd572a015c956758ea996d70d60bb31c72 Mon Sep 17 00:00:00 2001 From: yejunbo <692979649@qq.com> Date: Sat, 29 Aug 2026 02:05:29 +0800 Subject: [PATCH 1/3] fix(workflows): bound settled run retention --- extensions/workflows/artifacts.ts | 20 + extensions/workflows/index.ts | 127 +++- extensions/workflows/model.ts | 17 + extensions/workflows/retention.ts | 597 +++++++++++++++++++ tests/extensions/workflows/retention.test.ts | 188 ++++++ 5 files changed, 918 insertions(+), 31 deletions(-) create mode 100644 extensions/workflows/retention.ts create mode 100644 tests/extensions/workflows/retention.test.ts diff --git a/extensions/workflows/artifacts.ts b/extensions/workflows/artifacts.ts index 5f14e1d5..62131169 100644 --- a/extensions/workflows/artifacts.ts +++ b/extensions/workflows/artifacts.ts @@ -10,6 +10,7 @@ import { import { refreshWorkflowGraph, type TranscriptEntry, + type WorkflowDelivery, type WorkflowDetails, } from "./model.ts"; import { @@ -187,6 +188,25 @@ export function persistWorkflowJson( ); } +/** + * Update only the durable delivery receipt. Completion delivery may retain a + * compact memory projection after the run has been evicted, so rewriting the + * whole details object here would risk replacing exact result artifacts with + * that projection. + */ +export function persistWorkflowDeliveryState( + runDir: string, + delivery: WorkflowDelivery, +) { + const file = path.join(runDir, "workflow.json"); + const raw: unknown = JSON.parse(fs.readFileSync(file, "utf8")); + if (!raw || typeof raw !== "object" || Array.isArray(raw)) { + throw new Error(`Invalid persisted workflow details: ${file}`); + } + const next = { ...(raw as Record), delivery }; + writeFileAtomic(file, JSON.stringify(next)); +} + /** Coalesce live checkpoints while keeping final persistence synchronous. */ export function createWorkflowPersistence( runDir: string, diff --git a/extensions/workflows/index.ts b/extensions/workflows/index.ts index 03a4656c..85ec0730 100644 --- a/extensions/workflows/index.ts +++ b/extensions/workflows/index.ts @@ -85,6 +85,7 @@ import { createWorkflowPersistence, loadJournal, persistWorkflowAgentResult, + persistWorkflowDeliveryState, persistWorkflowJson, } from "./artifacts.ts"; import { RunController } from "./controller.ts"; @@ -185,6 +186,11 @@ import { type WorkflowAgentSessionFactory, type WorkflowModel, } from "./runner.ts"; +import { + createWorkflowSettledRunRetention, + projectWorkflowDetails, + type WorkflowSettledRunRetentionOptions, +} from "./retention.ts"; import { runWorkflowSandbox } from "./sandbox.ts"; import { safeStringify, writeFileAtomic } from "./serialization.ts"; import { @@ -757,17 +763,28 @@ function runDetailText( return `Run ${run.runId} — ${run.status}`; } -export default function workflows(pi: ExtensionAPI) { +export interface WorkflowExtensionOptions { + /** Test/configuration seam for the settled session-memory projection. */ + readonly settledRetention?: WorkflowSettledRunRetentionOptions; +} + +const WORKFLOW_DELIVERY_DETAILS_MAX_BYTES = 128 * 1024; + +export default function workflows( + pi: ExtensionAPI, + options: WorkflowExtensionOptions = {}, +) { /** Live background runs, for /workflows and shutdown cleanup. */ const activeRuns = new Map(); const activeDetails = () => new Map( [...activeRuns].map(([runId, run]) => [runId, run.details] as const), ); - const settledRuns = new Map(); - /** Keep current-session settled records live for their ephemeral UI renderer. */ - const dashboardDetails = () => - new Map([...settledRuns, ...activeDetails()]); + const settledRuns = createWorkflowSettledRunRetention( + options.settledRetention, + ); + /** Settled details are loaded from canonical artifacts by the dashboard. */ + const dashboardDetails = () => activeDetails(); const registerStableToolFamily = () => patchOwnedTools(pi, "workflows", { enable: OPENPI_TOOL_SURFACE.workflows.entry, @@ -785,22 +802,41 @@ export default function workflows(pi: ExtensionAPI) { ): WorkflowCompletionEnvelope => { const deliveryId = details.delivery?.id; if (!deliveryId) throw new Error("Workflow delivery identity is missing"); + const projection = projectWorkflowDetails( + details, + WORKFLOW_DELIVERY_DETAILS_MAX_BYTES, + ); + if (!projection) { + throw new Error( + `Workflow ${details.runId} cannot create a bounded completion projection`, + ); + } return { deliveryId, runId: details.runId, - details, + details: projection, }; }; const resultDelivery = createWorkflowResultDelivery({ isIdle: () => lastContext?.isIdle() ?? false, - persist: (details) => - persistWorkflowJson( + persist: (details) => { + if (!details.delivery) + throw new Error("Workflow delivery identity is missing"); + persistWorkflowDeliveryState( path.join(getAgentDir(), "workflows", details.runId), - details, - ), + details.delivery, + ); + }, deliver: async (envelopes, wake) => { + const hydrated = envelopes.map((envelope) => ({ + ...envelope, + details: + readPersistedWorkflowDetails(envelope.runId, { + hydrateArtifacts: true, + }) ?? envelope.details, + })); const content = buildProjectedWorkflowCompletionBatch( - envelopes.map((envelope) => ({ + hydrated.map((envelope) => ({ deliveryId: envelope.deliveryId, details: envelope.details, runDir: path.join(getAgentDir(), "workflows", envelope.runId), @@ -812,8 +848,8 @@ export default function workflows(pi: ExtensionAPI) { customType: "workflow-result", content, display: true, - ...(envelopes.length === 1 - ? { details: compactToolDetails(envelopes[0]!.details) } + ...(hydrated.length === 1 + ? { details: compactToolDetails(hydrated[0]!.details) } : {}), }, wake @@ -859,7 +895,10 @@ export default function workflows(pi: ExtensionAPI) { const running = newestEntry( [...activeRuns].map(([runId, run]) => [runId, run.details] as const), ); - return running ?? newestEntry(settledRuns); + return ( + running ?? + newestEntry(settledRuns.entriesArray()) + ); }; const updateWorkflowWidget = () => { @@ -915,7 +954,7 @@ export default function workflows(pi: ExtensionAPI) { }; const recordSettledRun = (details: WorkflowDetails) => { - settledRuns.set(details.runId, details); + settledRuns.set(details); if (details.status === "completed") completedRuns += 1; else failedRuns += 1; }; @@ -985,7 +1024,7 @@ export default function workflows(pi: ExtensionAPI) { turnStartedAt = 0; completedRuns = 0; failedRuns = 0; - settledRuns.clear(); + settledRuns.resetSession(); installWorkflowNavigation(ctx); updateIndicator(); @@ -2257,24 +2296,25 @@ export default function workflows(pi: ExtensionAPI) { const active = activeRuns.get(resolution.runId); if (active) return { ok: true, details: active.details } as const; - const settled = settledRuns.get(resolution.runId); - if (settled) return { ok: true, details: settled } as const; - const details = readPersistedWorkflowDetails(resolution.runId, { hydrateArtifacts: true, }); - if (!details) { + if (details) { + // A run absent from activeRuns cannot still be running this session; a + // persisted "running" is a run that was hard-killed or missed the + // shutdown settle deadline. return { - ok: false, - error: `Workflow run ${resolution.runId} could not be read.`, + ok: true, + details: recoverStaleWorkflowDetails(details), } as const; } - // A run absent from activeRuns cannot still be running this session; a - // persisted "running" is a run that was hard-killed or missed the - // shutdown settle deadline. + // Keep the bounded projection as a diagnostic fallback when an artifact is + // temporarily unreadable. An explicit id still resolves to a known run. + const settled = settledRuns.get(resolution.runId); + if (settled) return { ok: true, details: settled } as const; return { - ok: true, - details: recoverStaleWorkflowDetails(details), + ok: false, + error: `Workflow run ${resolution.runId} could not be read.`, } as const; }; @@ -2347,32 +2387,57 @@ export default function workflows(pi: ExtensionAPI) { if (!resolution.ok) throw new Error(resolution.error); const details = resolution.details; const runDir = path.join(getAgentDir(), "workflows", details.runId); + const retention = settledRuns.stats; return Promise.resolve({ content: [ { type: "text", text: buildWorkflowStatusSummary(details, runDir) }, ], - details: { runs: [summarize(details)] }, + details: { + runs: [summarize(details)], + retention, + settledRunsEvicted: retention.settledRunsEvicted, + }, }); } const runs = [ ...[...activeRuns.values()].map((run) => run.details), ...settledRuns.values(), ]; + const retention = settledRuns.stats; if (runs.length === 0) { return Promise.resolve({ content: [ - { type: "text", text: "No active or recently finished workflows." }, + { + type: "text", + text: + retention.evictedRuns > 0 + ? `No active or retained workflows. ${retention.evictedRuns} settled run(s) omitted from memory in the current session; canonical artifacts remain available on disk.` + : "No active or recently finished workflows.", + }, ], - details: { runs: [] }, + details: { + runs: [], + retention, + settledRunsEvicted: retention.settledRunsEvicted, + }, }); } const lines = runs.map((d) => { const { done, failed, uncertain } = countStates(d); return `${d.runId}${d.name ? ` "${d.name}"` : ""} — ${statusWord(d.status)} · ${done + failed}/${d.agents.length} agents${failed ? `, ${failed} failed` : ""}${uncertain ? `, ${uncertain} uncertain` : ""}`; }); + if (retention.evictedRuns > 0) { + lines.push( + `Retention (current session): ${retention.retainedRuns} settled projection(s) retained; ${retention.evictedRuns} evicted/omitted (${retention.evictedBytes} UTF-8 bytes). Canonical artifacts remain available on disk.`, + ); + } return Promise.resolve({ content: [{ type: "text", text: lines.join("\n") }], - details: { runs: runs.map(summarize) }, + details: { + runs: runs.map(summarize), + retention, + settledRunsEvicted: retention.settledRunsEvicted, + }, }); }, }); diff --git a/extensions/workflows/model.ts b/extensions/workflows/model.ts index e5bd48ae..7973af05 100644 --- a/extensions/workflows/model.ts +++ b/extensions/workflows/model.ts @@ -143,6 +143,21 @@ export interface WorkflowLogEntry { text: string; } +/** Metadata attached only to the bounded in-memory settled-run projection. */ +export interface WorkflowMemoryProjection { + readonly kind: "settled"; + readonly maxBytes: number; + readonly bytes: number; + readonly truncated: boolean; + readonly omitted: { + readonly agents: number; + readonly logs: number; + readonly transcriptEntries: number; + readonly result: boolean; + readonly graph: boolean; + }; +} + export interface WorkflowDetails { runId: string; /** Pi session that launched this run. */ @@ -172,6 +187,8 @@ export interface WorkflowDetails { /** Read-only lineage projection; never execution or admission authority. */ graph?: WorkflowGraphProjection; error?: string; + /** Present only on the session-memory projection, never canonical history. */ + memoryProjection?: WorkflowMemoryProjection; } export function workflowGraphRecords( diff --git a/extensions/workflows/retention.ts b/extensions/workflows/retention.ts new file mode 100644 index 00000000..3b5bb697 --- /dev/null +++ b/extensions/workflows/retention.ts @@ -0,0 +1,597 @@ +import type { + AgentRecord, + AgentUsage, + WorkflowDetails, + WorkflowMemoryProjection, +} from "./model.ts"; + +/** Defaults apply only to settled session-memory projections. Disk is canonical. */ +export const DEFAULT_WORKFLOW_SETTLED_MAX_RUNS = 32; +export const DEFAULT_WORKFLOW_SETTLED_MAX_BYTES = 2 * 1024 * 1024; + +const MAX_PHASES = 32; +const MAX_LOGS = 8; +const MAX_NAME_BYTES = 512; +const MAX_DESCRIPTION_BYTES = 1_024; +const MAX_LABEL_BYTES = 256; +const MAX_PREVIEW_BYTES = 512; +const MAX_ERROR_BYTES = 1_024; + +export interface WorkflowSettledRunRetentionOptions { + /** Maximum number of settled projections retained in this session. */ + readonly maxRuns?: number; + /** Maximum serialized UTF-8 bytes retained by all projections. */ + readonly maxBytes?: number; +} + +export interface WorkflowRetentionStats { + /** These counters describe only the current process/session epoch. */ + readonly scope: "current-session"; + readonly retainedRuns: number; + readonly retainedBytes: number; + /** Cumulative runs removed by count/byte pressure in this session. */ + readonly evictedRuns: number; + /** Compatibility alias for evictedRuns. */ + readonly settledRunsEvicted: number; + /** Serialized bytes belonging to projections removed by pressure. */ + readonly evictedBytes: number; +} + +function utf8Head(value: string, maxBytes: number) { + if (maxBytes <= 0) return ""; + const bytes = Buffer.from(value, "utf8"); + if (bytes.length <= maxBytes) return value; + let end = maxBytes; + while (end > 0 && (bytes[end] & 0xc0) === 0x80) end--; + return bytes.subarray(0, end).toString("utf8"); +} + +function bounded(value: string | undefined, maxBytes: number) { + return value === undefined ? undefined : utf8Head(value, maxBytes); +} + +function jsonBytes(value: unknown) { + const serialized = JSON.stringify(value); + if (serialized === undefined) + throw new Error("Workflow projection is not serializable"); + return Buffer.byteLength(serialized, "utf8"); +} + +function validateLimit(value: number, name: string) { + if (!Number.isSafeInteger(value) || value < 0) { + throw new Error(`${name} must be a non-negative safe integer`); + } + return value; +} + +function finite(value: number | undefined) { + return typeof value === "number" && Number.isFinite(value) + ? value + : undefined; +} + +function compactUsage(usage: AgentUsage) { + return { + input: finite(usage.input) ?? 0, + output: finite(usage.output) ?? 0, + cacheRead: finite(usage.cacheRead) ?? 0, + cacheWrite: finite(usage.cacheWrite) ?? 0, + cost: finite(usage.cost) ?? 0, + ...(finite(usage.contextTokens) !== undefined + ? { contextTokens: usage.contextTokens } + : {}), + turns: finite(usage.turns) ?? 0, + }; +} + +function compactInvocation(agent: AgentRecord) { + const invocation = agent.invocation; + if (!invocation) return undefined; + return { + identity: { + runId: invocation.identity.runId, + callIndex: invocation.identity.callIndex, + }, + intentState: invocation.intentState, + admissionState: invocation.admissionState, + executionState: invocation.executionState, + ...(invocation.outcome ? { outcome: invocation.outcome } : {}), + requestedAt: invocation.requestedAt, + ...(invocation.claimedAt !== undefined + ? { claimedAt: invocation.claimedAt } + : {}), + ...(invocation.runningAt !== undefined + ? { runningAt: invocation.runningAt } + : {}), + ...(invocation.terminalAt !== undefined + ? { terminalAt: invocation.terminalAt } + : {}), + }; +} + +function compactAgent( + agent: AgentRecord, + display: boolean, +): Record { + const result: Record = { + index: agent.index, + ...(agent.callId ? { callId: agent.callId } : {}), + ...(agent.invocation ? { invocation: compactInvocation(agent) } : {}), + label: bounded(agent.label, MAX_LABEL_BYTES) ?? "agent", + ...(agent.phase ? { phase: bounded(agent.phase, MAX_LABEL_BYTES) } : {}), + state: agent.state, + ...(agent.model ? { model: bounded(agent.model, MAX_LABEL_BYTES) } : {}), + ...(agent.contextWindow !== undefined + ? { contextWindow: agent.contextWindow } + : {}), + startedAt: agent.startedAt, + ...(agent.finishedAt !== undefined ? { finishedAt: agent.finishedAt } : {}), + usage: compactUsage(agent.usage), + transcript: [], + }; + + // These are recovery references, not display text. Truncating them would + // turn an otherwise recoverable artifact into an unusable path. + if (agent.resultArtifact) result.resultArtifact = agent.resultArtifact; + if (agent.resultRef) result.resultRef = agent.resultRef; + + if (!display) return result; + if (agent.preview) result.preview = bounded(agent.preview, MAX_PREVIEW_BYTES); + if (agent.error) result.error = bounded(agent.error, MAX_ERROR_BYTES); + if (agent.replayed) result.replayed = true; + if (agent.operatorKey) result.operatorKey = agent.operatorKey; + if (agent.inputCallIds?.length) + result.inputCallIds = agent.inputCallIds.slice(0, 32); + if (agent.worktreeBranch) result.worktreeBranch = agent.worktreeBranch; + if (agent.worktreePath) result.worktreePath = agent.worktreePath; + if (agent.worktreeHandoffArtifact) + result.worktreeHandoffArtifact = agent.worktreeHandoffArtifact; + if (agent.worktreeCleanup) { + result.worktreeCleanup = { + removed: agent.worktreeCleanup.removed, + branchDeleted: agent.worktreeCleanup.branchDeleted, + branch: agent.worktreeCleanup.branch, + detached: agent.worktreeCleanup.detached, + ...(agent.worktreeCleanup.reason + ? { reason: bounded(agent.worktreeCleanup.reason, MAX_ERROR_BYTES) } + : {}), + }; + } + if (agent.acceptance) { + result.acceptance = { + status: agent.acceptance.status, + errors: agent.acceptance.errors + .slice(0, 4) + .map((error) => bounded(error, MAX_ERROR_BYTES)), + criteria: agent.acceptance.criteria.slice(0, 16).map((criterion) => ({ + id: bounded(criterion.id, MAX_LABEL_BYTES), + status: criterion.status, + evidence: criterion.evidence + .slice(0, 4) + .map((evidence) => bounded(evidence, MAX_PREVIEW_BYTES)), + ...(criterion.note + ? { note: bounded(criterion.note, MAX_ERROR_BYTES) } + : {}), + })), + }; + } + return result; +} + +function omissionMetadata( + details: WorkflowDetails, + agents: readonly Record[], + logs: readonly unknown[], + agentLimit: number, + logLimit: number, +): WorkflowMemoryProjection["omitted"] { + return { + agents: Math.max(0, details.agents.length - agentLimit), + logs: + Math.max(0, (details.logs?.length ?? 0) - logLimit) + + (details.logsDropped ?? 0), + transcriptEntries: details.agents.reduce( + (total, agent) => total + agent.transcript.length, + 0, + ), + result: details.result !== undefined, + graph: details.graph !== undefined, + }; +} + +function makeProjection( + details: WorkflowDetails, + cap: number, + agents: readonly Record[], + logs: readonly unknown[], + agentLimit: number, + logLimit: number, + display: boolean, +) { + const phases = details.phases.slice(0, MAX_PHASES).map((phase) => ({ + title: bounded(phase.title, MAX_LABEL_BYTES) ?? "phase", + ...(phase.detail + ? { detail: bounded(phase.detail, MAX_DESCRIPTION_BYTES) } + : {}), + })); + const omitted = omissionMetadata(details, agents, logs, agentLimit, logLimit); + const candidate: Record = { + runId: details.runId, + ...(details.sessionId ? { sessionId: details.sessionId } : {}), + ...(details.name ? { name: bounded(details.name, MAX_NAME_BYTES) } : {}), + ...(details.description + ? { description: bounded(details.description, MAX_DESCRIPTION_BYTES) } + : {}), + background: details.background, + status: details.status, + startedAt: details.startedAt, + ...(details.finishedAt !== undefined + ? { finishedAt: details.finishedAt } + : {}), + phases, + ...(details.currentPhase + ? { currentPhase: bounded(details.currentPhase, MAX_LABEL_BYTES) } + : {}), + agents, + ...(logs.length > 0 ? { logs } : {}), + ...(details.logsDropped ? { logsDropped: details.logsDropped } : {}), + ...(details.delivery + ? { + delivery: { + id: details.delivery.id, + state: details.delivery.state, + attempts: details.delivery.attempts, + updatedAt: details.delivery.updatedAt, + ...(details.delivery.deliveredAt !== undefined + ? { deliveredAt: details.delivery.deliveredAt } + : {}), + ...(details.delivery.lastError + ? { + lastError: bounded( + details.delivery.lastError, + MAX_ERROR_BYTES, + ), + } + : {}), + }, + } + : {}), + ...(details.result !== undefined || details.resultArtifact + ? { + result: details.resultArtifact + ? "[stored in result.json]" + : "[result omitted from memory]", + ...(details.resultArtifact + ? { resultArtifact: details.resultArtifact } + : {}), + } + : {}), + ...(details.transcriptArtifact + ? { transcriptArtifact: details.transcriptArtifact } + : {}), + ...(details.resumedFrom ? { resumedFrom: details.resumedFrom } : {}), + ...(details.resumeNote + ? { resumeNote: bounded(details.resumeNote, MAX_ERROR_BYTES) } + : {}), + ...(details.error + ? { error: bounded(details.error, MAX_ERROR_BYTES) } + : {}), + ...(details.graph ? { graphOmitted: true } : {}), + }; + + const sourceBytes = jsonBytes(details); + const projectedBytes = jsonBytes(candidate); + const omittedBytes = Math.max(0, sourceBytes - projectedBytes); + const metadata = (bytes: number): WorkflowMemoryProjection => ({ + kind: "settled", + maxBytes: cap, + bytes, + truncated: + omittedBytes > 0 || + omitted.agents > 0 || + omitted.logs > 0 || + omitted.transcriptEntries > 0 || + omitted.result || + omitted.graph, + omitted, + }); + + return { candidate, metadata }; +} + +/** + * Build a detached settled-run projection. It intentionally never copies + * transcripts or arbitrary result values; those remain in the run artifacts. + */ +export function projectWorkflowDetails( + details: WorkflowDetails, + maxBytes = DEFAULT_WORKFLOW_SETTLED_MAX_BYTES, +): WorkflowDetails | undefined { + const cap = validateLimit(maxBytes, "maxBytes"); + if (cap === 0) return undefined; + + let display = true; + let agentLimit = details.agents.length; + let logLimit = Math.min(MAX_LOGS, details.logs?.length ?? 0); + let agents = details.agents.map((agent) => compactAgent(agent, display)); + let logs = (details.logs ?? []).slice(-logLimit).map((log) => ({ + at: log.at, + text: bounded(log.text, MAX_PREVIEW_BYTES) ?? "", + })); + + const removeOptionalFields = () => { + display = false; + agents = details.agents.map((agent) => compactAgent(agent, false)); + logs = []; + logLimit = 0; + }; + + const dropAgent = () => { + if (agents.length === 0) return false; + // Keep the oldest and newest identities preferentially while pressure + // removes middle display rows. Lifecycle facts for retained rows survive. + const index = agents.length > 1 ? Math.floor(agents.length / 2) : 0; + agents = [...agents.slice(0, index), ...agents.slice(index + 1)]; + agentLimit--; + return true; + }; + + const dropRootOptional = (candidate: Record) => { + for (const key of [ + "description", + "resumeNote", + "error", + "currentPhase", + "name", + "sessionId", + "logsDropped", + "logs", + "graphOmitted", + "result", + ]) { + if (key in candidate) { + delete candidate[key]; + return true; + } + } + return false; + }; + + let candidate: Record | undefined; + let projection: WorkflowMemoryProjection | undefined; + for (let pass = 0; pass < details.agents.length + 32; pass++) { + const built = makeProjection( + details, + cap, + agents, + logs, + agentLimit, + logLimit, + display, + ); + candidate = built.candidate; + projection = built.metadata(0); + candidate.memoryProjection = projection; + let bytes = jsonBytes(candidate); + if (bytes <= cap) { + // The byte count is itself part of the projection. Iterate until the + // decimal width of that field stabilizes. + for (let i = 0; i < 4; i++) { + candidate.memoryProjection = built.metadata(bytes); + const next = jsonBytes(candidate); + if (next === bytes) break; + bytes = next; + } + candidate.memoryProjection = built.metadata(bytes); + bytes = jsonBytes(candidate); + if (bytes <= cap) return candidate as unknown as WorkflowDetails; + } + + if (display) { + removeOptionalFields(); + continue; + } + if (dropAgent()) continue; + if (dropRootOptional(candidate)) continue; + + // The exact run id and terminal identity are non-negotiable. A caller + // using an impossibly small test budget gets an explicit non-retained run + // rather than a projection that cannot be addressed or measured safely. + const minimal: Record = { + runId: details.runId, + background: details.background, + status: details.status, + startedAt: details.startedAt, + ...(details.finishedAt !== undefined + ? { finishedAt: details.finishedAt } + : {}), + phases: [], + agents: [], + ...(details.resultArtifact + ? { resultArtifact: details.resultArtifact } + : {}), + ...(details.transcriptArtifact + ? { transcriptArtifact: details.transcriptArtifact } + : {}), + }; + const minimalProjection = makeProjection( + details, + cap, + [], + [], + 0, + 0, + false, + ).metadata(0); + minimal.memoryProjection = minimalProjection; + const minimalBytes = jsonBytes(minimal); + if (minimalBytes <= cap) { + minimal.memoryProjection = { ...minimalProjection, bytes: minimalBytes }; + if (jsonBytes(minimal) <= cap) + return minimal as unknown as WorkflowDetails; + } + return undefined; + } + return undefined; +} + +export function measureWorkflowDetailsBytes(details: WorkflowDetails) { + return jsonBytes(details); +} + +interface RetainedEntry { + readonly details: WorkflowDetails; + readonly bytes: number; +} + +/** Count- and byte-bounded insertion-ordered store for settled projections. */ +export class WorkflowSettledRunRetention { + readonly maxRuns: number; + readonly maxBytes: number; + private readonly entries = new Map(); + private totalBytes = 0; + private totalEvictedRuns = 0; + private totalEvictedBytes = 0; + + constructor(options: WorkflowSettledRunRetentionOptions = {}) { + this.maxRuns = validateLimit( + options.maxRuns ?? DEFAULT_WORKFLOW_SETTLED_MAX_RUNS, + "maxRuns", + ); + this.maxBytes = validateLimit( + options.maxBytes ?? DEFAULT_WORKFLOW_SETTLED_MAX_BYTES, + "maxBytes", + ); + } + + set(details: WorkflowDetails) { + const previous = this.entries.get(details.runId); + if (previous) { + this.entries.delete(details.runId); + this.totalBytes -= previous.bytes; + } + const projection = projectWorkflowDetails(details, this.maxBytes); + if (!projection) { + // Keep a previously valid projection addressable when an unusually small + // configured budget cannot represent a later update. The canonical file + // remains the recovery source either way. + if (previous) { + this.entries.set(details.runId, previous); + this.totalBytes += previous.bytes; + } else { + this.totalEvictedRuns++; + } + return undefined; + } + const bytes = measureWorkflowDetailsBytes(projection); + this.entries.set(details.runId, { details: projection, bytes }); + this.totalBytes += bytes; + while ( + this.entries.size > this.maxRuns || + this.totalBytes > this.maxBytes + ) { + const oldest = this.entries.keys().next().value; + if (oldest === undefined) break; + const entry = this.entries.get(oldest); + this.entries.delete(oldest); + if (!entry) continue; + this.totalBytes -= entry.bytes; + this.totalEvictedRuns++; + this.totalEvictedBytes += entry.bytes; + } + return this.entries.get(details.runId)?.details; + } + + get(runId: string) { + return this.entries.get(runId)?.details; + } + + has(runId: string) { + return this.entries.has(runId); + } + + delete(runId: string) { + const entry = this.entries.get(runId); + if (!entry) return false; + this.entries.delete(runId); + this.totalBytes -= entry.bytes; + return true; + } + + clear() { + this.entries.clear(); + this.totalBytes = 0; + } + + /** Clear retained projections without changing current-session counters. */ + reset() { + this.clear(); + } + + /** Start a new session accounting epoch. */ + resetStats() { + this.totalEvictedRuns = 0; + this.totalEvictedBytes = 0; + } + + /** Clear projections and start a new session accounting epoch. */ + resetSession() { + this.clear(); + this.resetStats(); + } + + keys() { + return this.entries.keys(); + } + + values() { + return [...this.entries.values()].map((entry) => entry.details).values(); + } + + entriesArray() { + return [...this.entries.entries()].map( + ([id, entry]) => [id, entry.details] as const, + ); + } + + get size() { + return this.entries.size; + } + + get retainedBytes() { + return this.totalBytes; + } + + get evictedRuns() { + return this.totalEvictedRuns; + } + + get settledRunsEvicted() { + return this.totalEvictedRuns; + } + + get evictedBytes() { + return this.totalEvictedBytes; + } + + get stats(): WorkflowRetentionStats { + return { + scope: "current-session", + retainedRuns: this.entries.size, + retainedBytes: this.totalBytes, + evictedRuns: this.totalEvictedRuns, + settledRunsEvicted: this.totalEvictedRuns, + evictedBytes: this.totalEvictedBytes, + }; + } + + getStats() { + return this.stats; + } +} + +export function createWorkflowSettledRunRetention( + options: WorkflowSettledRunRetentionOptions = {}, +) { + return new WorkflowSettledRunRetention(options); +} diff --git a/tests/extensions/workflows/retention.test.ts b/tests/extensions/workflows/retention.test.ts new file mode 100644 index 00000000..23ce866e --- /dev/null +++ b/tests/extensions/workflows/retention.test.ts @@ -0,0 +1,188 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import { + persistWorkflowDeliveryState, + persistWorkflowJson, +} from "../../../extensions/workflows/artifacts.ts"; +import { + WorkflowSettledRunRetention, + measureWorkflowDetailsBytes, + projectWorkflowDetails, +} from "../../../extensions/workflows/retention.ts"; +import { + emptyUsage, + type WorkflowDetails, +} from "../../../extensions/workflows/model.ts"; + +function details(runId: string, label = "worker"): WorkflowDetails { + return { + runId, + sessionId: "retention-test", + background: true, + status: "completed", + startedAt: 1, + finishedAt: 2, + phases: [], + agents: [ + { + index: 1, + label, + state: "done", + startedAt: 1, + finishedAt: 2, + preview: "completed", + usage: emptyUsage(), + transcript: [{ role: "assistant", text: "完整结果" }], + }, + ], + }; +} + +test("retention evicts by count and exposes compatible statistics", () => { + const retention = new WorkflowSettledRunRetention({ maxRuns: 2 }); + retention.set(details("wf_1")); + retention.set(details("wf_2")); + retention.set(details("wf_3")); + + assert.equal(retention.size, 2); + assert.equal(retention.get("wf_1"), undefined); + assert.equal(retention.evictedRuns, 1); + assert.equal(retention.settledRunsEvicted, 1); + assert.equal(retention.stats.settledRunsEvicted, retention.stats.evictedRuns); +}); + +test("retention evicts by aggregate UTF-8 bytes and never exceeds the cap", () => { + const first = projectWorkflowDetails(details("wf_字", "甲"), 100_000)!; + const retention = new WorkflowSettledRunRetention({ + maxRuns: 10, + maxBytes: measureWorkflowDetailsBytes(first), + }); + retention.set(details("wf_字", "甲")); + retention.set(details("wf_二", "乙")); + + assert.ok(retention.retainedBytes <= retention.maxBytes); + assert.ok(retention.evictedRuns >= 1); + assert.ok(retention.evictedBytes > 0); + assert.ok( + retention.get("wf_字") === undefined || + retention.get("wf_二") === undefined, + ); +}); + +test("projection byte accounting is UTF-8 safe and preserves exact references", () => { + const source = details("wf_引用", "中文代理"); + source.resultArtifact = "result-这是一个不能被截断的精确引用.json"; + source.transcriptArtifact = "transcripts-这是一个不能被截断的精确引用.json"; + source.agents[0]!.resultRef = "result-ref-这是一个精确引用"; + const projection = projectWorkflowDetails(source, 100_000); + + assert.ok(projection); + assert.equal(projection?.resultArtifact, source.resultArtifact); + assert.equal(projection?.transcriptArtifact, source.transcriptArtifact); + assert.equal(projection?.agents[0]?.resultRef, source.agents[0]?.resultRef); + assert.equal( + Buffer.byteLength(JSON.stringify(projection), "utf8"), + projection?.memoryProjection?.bytes, + ); +}); + +test("replacement does not count a failed projection as a new eviction", () => { + const initial = details("wf_replace"); + const initialProjection = projectWorkflowDetails(initial, 100_000)!; + const retention = new WorkflowSettledRunRetention({ + maxRuns: 1, + maxBytes: measureWorkflowDetailsBytes(initialProjection), + }); + retention.set(initial); + const before = retention.stats; + const original = retention.get("wf_replace"); + assert.ok(original); + + const updated = { + ...initial, + transcriptArtifact: "x".repeat(retention.maxBytes), + }; + assert.equal(retention.set(updated), undefined); + assert.equal(retention.get("wf_replace"), original); + assert.deepEqual(retention.stats, before); +}); + +test("zero limits are valid and invalid limits fail fast", () => { + assert.equal(new WorkflowSettledRunRetention({ maxRuns: 0 }).maxRuns, 0); + const zeroBytes = new WorkflowSettledRunRetention({ maxBytes: 0 }); + assert.equal(zeroBytes.maxBytes, 0); + assert.equal(zeroBytes.set(details("wf_zero")), undefined); + assert.equal(zeroBytes.evictedRuns, 1); + for (const options of [ + { maxRuns: -1 }, + { maxRuns: 1.5 }, + { maxRuns: Number.NaN }, + { maxBytes: Number.POSITIVE_INFINITY }, + { maxBytes: Number.MAX_SAFE_INTEGER + 1 }, + ]) { + assert.throws( + () => new WorkflowSettledRunRetention(options), + /non-negative safe integer/, + ); + } +}); + +test("reset separates retained memory from current-session statistics", () => { + const retention = new WorkflowSettledRunRetention({ maxRuns: 1 }); + retention.set(details("wf_1")); + retention.set(details("wf_2")); + assert.equal(retention.evictedRuns, 1); + + retention.reset(); + assert.equal(retention.size, 0); + assert.equal(retention.evictedRuns, 1); + retention.resetStats(); + assert.equal(retention.evictedRuns, 0); + assert.equal(retention.evictedBytes, 0); +}); + +test("projection and delivery metadata cannot overwrite canonical artifacts", () => { + const directory = mkdtempSync( + join(tmpdir(), "pi-workflow-retention-artifacts-"), + ); + try { + const source = details("wf_canonical"); + source.result = { answer: "完整中文结果", nested: { value: 42 } }; + persistWorkflowJson(directory, source); + const before = readFileSync(join(directory, "workflow.json"), "utf8"); + const projection = projectWorkflowDetails(source, 512)!; + assert.notEqual(projection, source); + persistWorkflowDeliveryState(directory, { + id: "workflow:wf_canonical:terminal", + state: "delivered", + attempts: 1, + updatedAt: 3, + deliveredAt: 3, + }); + const stored = JSON.parse( + readFileSync(join(directory, "workflow.json"), "utf8"), + ) as Record; + assert.equal(stored.result, "[stored in result.json]"); + assert.equal(stored.resultArtifact, "result.json"); + assert.deepEqual(stored.delivery, { + id: "workflow:wf_canonical:terminal", + state: "delivered", + attempts: 1, + updatedAt: 3, + deliveredAt: 3, + }); + assert.deepEqual( + JSON.parse(readFileSync(join(directory, "result.json"), "utf8")), + source.result, + ); + assert.notEqual( + readFileSync(join(directory, "workflow.json"), "utf8"), + before, + ); + } finally { + rmSync(directory, { recursive: true, force: true }); + } +}); From 1c9b3a6b3538496b2cced2caa7975e793af00101 Mon Sep 17 00:00:00 2001 From: yejunbo <692979649@qq.com> Date: Sun, 30 Aug 2026 00:42:36 +0800 Subject: [PATCH 2/3] fix(workflows): retain dashboard fallback projections --- extensions/workflows/dashboard.ts | 20 ++++++-- extensions/workflows/index.ts | 5 +- extensions/workflows/retention.ts | 3 +- tests/extensions/workflows/dashboard.test.ts | 52 ++++++++++++++++++++ tests/extensions/workflows/retention.test.ts | 14 ++++++ 5 files changed, 89 insertions(+), 5 deletions(-) diff --git a/extensions/workflows/dashboard.ts b/extensions/workflows/dashboard.ts index a37152eb..53ce2623 100644 --- a/extensions/workflows/dashboard.ts +++ b/extensions/workflows/dashboard.ts @@ -566,22 +566,30 @@ export function loadRunEntries( referencedRunIds: ReadonlySet, /** Hide runs untouched by the current request; live runs always show. */ startedSince = 0, + /** Bounded settled projections used only if canonical disk state is unreadable. */ + retained: ReadonlyMap = new Map(), ): RunEntry[] { const entries: RunEntry[] = []; - for (const runId of listPersistedRunIds()) { + const runIds = new Set([...listPersistedRunIds(), ...retained.keys()]); + for (const runId of runIds) { const live = active.get(runId); if (live) { entries.push({ runId, details: live, live: true }); continue; } - const details = readPersistedWorkflowDetails(runId, { + const persisted = readPersistedWorkflowDetails(runId, { hydrateArtifacts: true, }); + const retainedDetails = retained.get(runId); + const details = persisted ?? retainedDetails; if (!details) continue; + const fromRetention = persisted === undefined && retainedDetails !== undefined; const touchedAt = Math.max(details.startedAt, details.finishedAt ?? 0); if ( touchedAt < startedSince || - (details.sessionId !== sessionId && !referencedRunIds.has(runId)) + (!fromRetention && + details.sessionId !== sessionId && + !referencedRunIds.has(runId)) ) { continue; } @@ -713,6 +721,7 @@ export class WorkflowDashboard { private theme: Theme; private keybindings: KeybindingsManager; private getActive: () => Map; + private getRetained: () => ReadonlyMap; private sessionId: string; private referencedRunIds: ReadonlySet; private startedSince: number; @@ -730,11 +739,13 @@ export class WorkflowDashboard { close: () => void, initialRunId?: string, onAbort?: (runId: string) => boolean, + getRetained: () => ReadonlyMap = () => new Map(), ) { this.tui = tui; this.theme = theme; this.keybindings = keybindings; this.getActive = getActive; + this.getRetained = getRetained; this.sessionId = sessionId; this.referencedRunIds = referencedRunIds; this.startedSince = startedSince; @@ -805,6 +816,7 @@ export class WorkflowDashboard { this.sessionId, this.referencedRunIds, this.startedSince, + this.getRetained(), ); if (selected) { const index = this.entries.findIndex((e) => e.runId === selected); @@ -1405,6 +1417,7 @@ export async function showWorkflowDashboard( initialRunId?: string, startedSince = 0, onAbort?: (runId: string) => boolean, + getRetained?: () => ReadonlyMap, ) { await ctx.ui.custom( (tui, theme, keybindings, done) => { @@ -1422,6 +1435,7 @@ export async function showWorkflowDashboard( }, initialRunId, onAbort, + getRetained, ); return dashboard; }, diff --git a/extensions/workflows/index.ts b/extensions/workflows/index.ts index f361c0db..738945c0 100644 --- a/extensions/workflows/index.ts +++ b/extensions/workflows/index.ts @@ -795,8 +795,10 @@ export default function workflows( const settledRuns = createWorkflowSettledRunRetention( options.settledRetention, ); - /** Settled details are loaded from canonical artifacts by the dashboard. */ + /** Disk remains canonical; retained projections cover transient read failures. */ const dashboardDetails = () => activeDetails(); + const dashboardRetainedDetails = () => + new Map(settledRuns.entriesArray()); const registerStableToolFamily = () => patchOwnedTools(pi, "workflows", { enable: OPENPI_TOOL_SURFACE.workflows.entry, @@ -993,6 +995,7 @@ export default function workflows( initialRunId, startedSince, stopRun, + dashboardRetainedDetails, ); acknowledgeSettledRuns(); } finally { diff --git a/extensions/workflows/retention.ts b/extensions/workflows/retention.ts index 3b5bb697..7f017f59 100644 --- a/extensions/workflows/retention.ts +++ b/extensions/workflows/retention.ts @@ -4,6 +4,7 @@ import type { WorkflowDetails, WorkflowMemoryProjection, } from "./model.ts"; +import { toSerializable } from "./serialization.ts"; /** Defaults apply only to settled session-memory projections. Disk is canonical. */ export const DEFAULT_WORKFLOW_SETTLED_MAX_RUNS = 32; @@ -51,7 +52,7 @@ function bounded(value: string | undefined, maxBytes: number) { } function jsonBytes(value: unknown) { - const serialized = JSON.stringify(value); + const serialized = JSON.stringify(toSerializable(value)); if (serialized === undefined) throw new Error("Workflow projection is not serializable"); return Buffer.byteLength(serialized, "utf8"); diff --git a/tests/extensions/workflows/dashboard.test.ts b/tests/extensions/workflows/dashboard.test.ts index 9e564fb7..e2b90208 100644 --- a/tests/extensions/workflows/dashboard.test.ts +++ b/tests/extensions/workflows/dashboard.test.ts @@ -61,6 +61,20 @@ function writeRun( ); } +function retainedRun(runId: string, startedAt: number): WorkflowDetails { + return { + runId, + sessionId: SESSION, + name: runId, + background: false, + status: "completed", + startedAt, + finishedAt: startedAt + 1_000, + agents: [], + phases: [], + }; +} + test("persisted nonterminal invocation facts are projected as uncertain", () => { const restored = normalizePersistedWorkflowDetails("wf_dead", { status: "running", @@ -352,6 +366,44 @@ test("the dashboard reports the current request, not the session's history", () ); }); +test("retained projections keep a settled run visible when disk state is unreadable", () => { + const runId = "wf_fa11bac"; + const details = retainedRun(runId, 6_000); + writeRun(runId, details.startedAt, details.finishedAt); + writeFileSync(join(agentDir, "workflows", runId, "workflow.json"), "{"); + + const entry = loadRunEntries( + new Map(), + SESSION, + new Set(), + 0, + new Map([[runId, details]]), + ).find((candidate) => candidate.runId === runId); + + assert.ok(entry); + assert.equal(entry.live, false); + assert.equal(entry.details, details); +}); + +test("retained projections without session metadata keep current-session runs visible", () => { + const runId = "wf_retained_minimal"; + const details = retainedRun(runId, 7_000); + delete details.sessionId; + writeRun(runId, details.startedAt, details.finishedAt); + writeFileSync(join(agentDir, "workflows", runId, "workflow.json"), "{"); + + const entry = loadRunEntries( + new Map(), + SESSION, + new Set(), + 0, + new Map([[runId, details]]), + ).find((candidate) => candidate.runId === runId); + + assert.ok(entry); + assert.equal(entry.live, false); + assert.equal(entry.details, details); +}); test("restored run directories require a generated safe id", () => { writeRun("wf_\u001b]52;c;clipboard\u0007", 9_000); const runIds = loadRunEntries(new Map(), SESSION, new Set()).map( diff --git a/tests/extensions/workflows/retention.test.ts b/tests/extensions/workflows/retention.test.ts index 23ce866e..5a19607d 100644 --- a/tests/extensions/workflows/retention.test.ts +++ b/tests/extensions/workflows/retention.test.ts @@ -89,6 +89,20 @@ test("projection byte accounting is UTF-8 safe and preserves exact references", ); }); +test("projection byte accounting accepts bigint and cyclic workflow values", () => { + const source = details("wf_non_json_values"); + const cyclic: Record = { count: 1n }; + cyclic.self = cyclic; + source.result = cyclic; + + const projection = projectWorkflowDetails(source, 100_000); + + assert.ok(projection); + assert.equal(projection?.result, "[result omitted from memory]"); + assert.doesNotThrow(() => measureWorkflowDetailsBytes(source)); + assert.doesNotThrow(() => new WorkflowSettledRunRetention().set(source)); +}); + test("replacement does not count a failed projection as a new eviction", () => { const initial = details("wf_replace"); const initialProjection = projectWorkflowDetails(initial, 100_000)!; From a9b43c6497782fdcdc11b5a623c839f083f96f8c Mon Sep 17 00:00:00 2001 From: yejunbo <692979649@qq.com> Date: Sun, 30 Aug 2026 01:19:07 +0800 Subject: [PATCH 3/3] test(workflows): cover delivery and retention pressure --- .../extensions/workflows/execute.e2e.test.ts | 83 ++++++++++++++++++- .../workflows/result-delivery.test.ts | 35 ++++++++ 2 files changed, 117 insertions(+), 1 deletion(-) diff --git a/tests/extensions/workflows/execute.e2e.test.ts b/tests/extensions/workflows/execute.e2e.test.ts index 99edb253..848f898c 100644 --- a/tests/extensions/workflows/execute.e2e.test.ts +++ b/tests/extensions/workflows/execute.e2e.test.ts @@ -156,7 +156,9 @@ for (const [runId, status] of [ ); } -workflows(pi); +workflows(pi, { + settledRetention: { maxRuns: 8, maxBytes: 64 * 1024 }, +}); for (const handler of handlers.get("session_start") ?? []) { await handler({}, { ...ctx, @@ -990,6 +992,85 @@ test("an oversized legacy replay is rejected without a success record", async () } }); +test("extension retention stays bounded and reports evictions under settled-run pressure", async () => { + const before = (await status.execute("e2e-retention-before", {})) as { + details: { + retention: { + retainedRuns: number; + retainedBytes: number; + evictedRuns: number; + }; + }; + }; + const beforeEvictions = before.details.retention.evictedRuns; + let sessionCreations = 0; + __setWorkflowTestAgentSessionFactory(async () => { + sessionCreations++; + return { session: fakeAgentSession(`pressure output ${sessionCreations}`) }; + }); + const script = + 'export const meta = { name: "retention-pressure" };\n' + + 'const r = await agent("pressure fixture", { agent_type: "reviewer", label: "pressure-agent" });\n' + + 'log("pressure log: " + r.output);\n' + + 'return { ok: r.ok, output: r.output };'; + + const runIds: string[] = []; + try { + for (let index = 0; index < 16; index++) { + const result = (await workflow.execute( + `e2e-retention-pressure-${index}`, + { script, wait: true }, + undefined, + undefined, + ctx, + )) as { details: { runId?: unknown; status?: unknown } }; + assert.equal(result.details.status, "completed"); + assert.equal(typeof result.details.runId, "string"); + runIds.push(result.details.runId as string); + } + } finally { + __setWorkflowTestAgentSessionFactory(undefined); + } + + for (const runId of runIds) { + const persisted = readWorkflowJson(runId); + assert.equal((persisted.agents as unknown[]).length, 1); + assert.equal((persisted.logs as unknown[]).length, 1); + } + + const after = (await status.execute("e2e-retention-after", {})) as { + content: Array<{ text: string }>; + details: { + runs: Array<{ name?: unknown; total: number }>; + retention: { + retainedRuns: number; + retainedBytes: number; + evictedRuns: number; + settledRunsEvicted: number; + }; + settledRunsEvicted: number; + }; + }; + assert.equal(sessionCreations, 16); + assert.equal(after.details.retention.retainedRuns, 8); + assert.ok(after.details.retention.retainedBytes <= 64 * 1024); + assert.ok(after.details.retention.evictedRuns - beforeEvictions >= 8); + assert.equal( + after.details.retention.settledRunsEvicted, + after.details.retention.evictedRuns, + ); + assert.equal( + after.details.settledRunsEvicted, + after.details.retention.evictedRuns, + ); + const retainedPressureRuns = after.details.runs.filter( + (run) => run.name === "retention-pressure", + ); + assert.equal(retainedPressureRuns.length, 8); + assert.ok(retainedPressureRuns.every((run) => run.total === 1)); + assert.match(after.content[0]!.text, /evicted\/omitted/); +}); + test.after(() => { for (const handler of handlers.get("session_shutdown") ?? []) { void handler({}, ctx); diff --git a/tests/extensions/workflows/result-delivery.test.ts b/tests/extensions/workflows/result-delivery.test.ts index c61ebb7d..19cab416 100644 --- a/tests/extensions/workflows/result-delivery.test.ts +++ b/tests/extensions/workflows/result-delivery.test.ts @@ -5,6 +5,7 @@ import { type WorkflowDetails, } from "../../../extensions/workflows/model.ts"; import { createWorkflowResultDelivery } from "../../../extensions/workflows/result-delivery.ts"; +import { projectWorkflowDetails } from "../../../extensions/workflows/retention.ts"; function details(runId: string): WorkflowDetails { return { @@ -35,6 +36,40 @@ function details(runId: string): WorkflowDetails { }; } +test("bigint and cyclic results remain deliverable through a bounded projection", async () => { + const run = details("wf_non_json_delivery"); + const cyclic: Record = { count: 1n }; + cyclic.self = cyclic; + run.result = cyclic; + + const projection = projectWorkflowDetails(run, 128 * 1024); + assert.ok(projection); + const delivered: WorkflowDetails[] = []; + const delivery = createWorkflowResultDelivery({ + isIdle: () => false, + persist: () => {}, + deliver: async (envelopes) => { + delivered.push(...envelopes.map((envelope) => envelope.details)); + return envelopes.map((envelope) => ({ + deliveryId: envelope.deliveryId, + delivered: true, + })); + }, + }); + + delivery.defer({ + deliveryId: projection.delivery!.id, + runId: projection.runId, + details: projection, + }); + await delivery.parentSettled(); + + assert.equal(delivery.size(), 0); + assert.equal(delivered.length, 1); + assert.equal(delivered[0]?.result, "[result omitted from memory]"); + assert.equal(delivered[0]?.delivery?.state, "delivered"); +}); + test("failed delivery stays pending and retries with the same per-run id", async () => { const run = details("wf_aa"); const persisted: string[] = [];