diff --git a/.changeset/27971-map-cancelled-item.md b/.changeset/27971-map-cancelled-item.md new file mode 100644 index 00000000000..55a326534c6 --- /dev/null +++ b/.changeset/27971-map-cancelled-item.md @@ -0,0 +1,22 @@ +--- +'@objectstack/service-automation': minor +'@objectstack/spec': minor +--- + +fix(service-automation)!: a `map` whose item run was cancelled records the item as cancelled when the map's own run is resumed + +Clause-②: yes (widening) +Metadata-protocol: added +Ruling-ref: 6523833308 + + + +`cancelRun` on a `map` item's run terminates that item and leaves the map's run parked on it. Resuming the map's run by its own id used to re-run the map, which moved past the item with nothing recorded: `results` held only the later items' outputs, `results[i]` stopped being item `i`, and the run finished `completed`. + +- **A cancelled item keeps its slot.** The resume reads the item run's history. When it says `cancelled`, the map goes on, and that item's entry in `outputVariable` (and in the node's `results`) is the closed marker `{ $status: 'cancelled', $itemRunId }` instead of an output. A downstream node reads it where it reads the results: `results[i]['$status'] == 'cancelled'`. A child flow's output keys are its declared variable names, which never start with `$`, so no real output looks like the marker. +- **Never a run or a success.** The node's `count` is the items that ran; the marker is not counted, and the cancelled item adds nothing to the step's totals. +- **Nothing to record is refused.** An item whose suspension is gone without a `cancelled` entry in its run history (no entry, an unreadable store, a run that completed without reaching the map) is refused `409 RESUME_CHILD_PENDING` before anything moves, the same code as an item that is still suspended. +- **Through a `subflow` too.** Resuming a run that waits on the map through a `subflow` link records the marker one level down, the same way. +- A caller still cannot write the map's handoff variables: a resume signal that writes the engine's `$` namespace is refused `400 INVALID_SIGNAL`, cancelled item or not. + +`MapConfigSchema.outputVariable`'s description in `@objectstack/spec` now says an element can be that marker. diff --git a/content/docs/automation/flows.mdx b/content/docs/automation/flows.mdx index 83713fecee3..ce519ec05fc 100644 --- a/content/docs/automation/flows.mdx +++ b/content/docs/automation/flows.mdx @@ -1512,7 +1512,22 @@ The run holds a **single program counter** the whole time: only one item's approval is open at any moment; when it is decided the engine **re-enters** the map node to start the next item. Resuming the map's own run id while an item is still awaiting a service-gated decision is **refused** (403) — otherwise the -batch would step past that item as if it had been decided. v1 is sequential and +batch would step past that item as if it had been decided; an item open to that +route is refused **409 `RESUME_CHILD_PENDING`**, naming the item run to resume. + +**A cancelled item keeps its slot.** When an item's run is cancelled +(`cancelRun`, which leaves the map's run parked), resuming the map's own run id +moves the batch on, and that item's entry in `outputVariable` (and in the +node's `results`) is the closed marker +`{ $status: 'cancelled', $itemRunId: '' }` instead of an +output, so `results[i]` is still item `i`. The marker is never a run or a +success: the node's `count` is the items that ran. A downstream step reads it +where it reads the results — `signoffResults[i]['$status'] == 'cancelled'` (a +child flow's output keys never start with `$`, so no real output looks like +it). An item whose run is gone *without* a `cancelled` entry in its run history +is refused with the same 409: the map has nothing true to record for it. + +v1 is sequential and fail-fast (the first item whose subflow fails fails the map). Concurrent fan-out (all items at once) is the larger [ADR-0039](https://github.com/objectstack-ai/objectstack/blob/main/docs/adr/0039-token-scope-tree-execution.md) Track B work. Worked example: `showcase_release_signoff` → `showcase_one_task_signoff`. diff --git a/content/docs/references/automation/builtin-node-config.mdx b/content/docs/references/automation/builtin-node-config.mdx index 3baea037e0c..bb49840d5e0 100644 --- a/content/docs/references/automation/builtin-node-config.mdx +++ b/content/docs/references/automation/builtin-node-config.mdx @@ -251,7 +251,7 @@ A value: a CEL value envelope `{ dialect: 'cel', source }` evaluated by the expr | **indexVariable** | `string` | optional | Optional variable holding the current index — a name without a leading `$` (the `$` names are the flow engine's own) | | **itemObject** | `string` | optional | When items are records, the object they belong to (exposes each item as the child's record) | | **input** | `Record` | optional | Params passed to each item's subflow, keyed by its input variables: each value a CEL value envelope `{ dialect: 'cel', source }` evaluated per item (the item variable is in scope), or a literal written as it is — a `{…}` template token is refused | -| **outputVariable** | `string` | optional | Each item's subflow output, collected in order — bound to a name without a leading `$` (the `$` names are the flow engine's own), read as `{{ name }}` | +| **outputVariable** | `string` | optional | Each item's subflow output, collected in order — bound to a name without a leading `$` (the `$` names are the flow engine's own), read as `{{ name }}`. An item whose run was cancelled keeps its slot as the closed marker `{ $status: 'cancelled', $itemRunId }`: never an output, and not counted as a run | --- diff --git a/docs/qa/platform-checklist/areas/automation.json b/docs/qa/platform-checklist/areas/automation.json index c0e9113c9c5..9fd049c3b13 100644 --- a/docs/qa/platform-checklist/areas/automation.json +++ b/docs/qa/platform-checklist/areas/automation.json @@ -1634,7 +1634,7 @@ "title": "The flow-dispatch HTTP status contract: every never-dispatched, ran-and-failed and non-terminal outcome answers its own status + ADR-0112 code, at every door that dispatches a flow", "since": "v17", "status": "active", - "revision": 9, + "revision": 10, "priority": "P1", "surface": "api", "personas": ["seeded admin (admin@objectos.ai / admin123)", "org member for the resume 403 arm — created by the admin with POST /api/v1/auth/admin/create-user and past its password change (POST /api/v1/auth/change-password): a held session answers 403 PASSWORD_RESET_REQUIRED, the right status for the wrong reason"], @@ -1728,7 +1728,7 @@ "resume · 400 INVALID_SIGNAL — signal wrote the engine's $ namespace, retryable", "resume · 400 INVALID_SCREEN_INPUT — bag violates the screen's field contract, retryable", "resume · 409 RESUME_IN_PROGRESS — a concurrent resume holds the run, retryable", - "resume · 409 RESUME_CHILD_PENDING — a parent-id resume of a map whose item run is still suspended; the message names the item run to resume, retryable. No stock flow reaches it: showcase_release_signoff's item parks on an approval, which answers 403 PERMISSION_DENIED first, so a run stages a map over a raw-resumable item (a screen) and resumes the map's own run id", + "resume · 409 RESUME_CHILD_PENDING — a parent-id resume of a map whose item run is still suspended; the message names the item run to resume, retryable. The same code answers when the item's suspension is gone without a cancelled entry in its run history (no entry, or an unreadable store); the message names the remedy, cancelling the parent run, and nothing moves. A CANCELLED item is not refused: the map continues and records {$status: 'cancelled', $itemRunId} in its slot. No stock flow reaches it: showcase_release_signoff's item parks on an approval, which answers 403 PERMISSION_DENIED first, so a run stages a map over a raw-resumable item (a screen) and resumes the map's own run id", "resume · 503 STORE_UNAVAILABLE — durable store unreadable, retryable", "resume · 404 RUN_NOT_FOUND — no such suspension, or a STALE one; terminal", "resume · 400 FLOW_FAILED — the resumed run consumed its pause and failed; terminal", @@ -1800,6 +1800,12 @@ "date": "2026-10-30", "change": "acceptance[4] and the variants name the resume door's map-item refusal, 409 RESUME_CHILD_PENDING (PR #27951): a parent-id resume of a map whose item run is still suspended is refused before anything moves, naming the item run. Read at base f6adf480 (run-resume.ts refuseMapItemSkip; runtime automation.ts code→status table)", "ref": "#27971" + }, + { + "revision": 10, + "date": "2026-10-30", + "change": "the RESUME_CHILD_PENDING variant names the code's second arm and the cancelled-item record (PR #28176): an item whose suspension is gone without a cancelled history entry is refused with the same 409, naming the remedy (cancel the parent run); a cancelled item is recorded in its slot and the map continues. Read at head of PR #28176 (run-resume.ts continuePastCancelledMapItem; builtin/map-node.ts)", + "ref": "#27971" } ] }, diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index c49328815e7..2ae267581a0 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -5946,7 +5946,9 @@ export class ObjectStackClient { * intact keep their own codes and stay retryable — `INVALID_SIGNAL` / * `INVALID_SCREEN_INPUT` (400), `RESUME_IN_PROGRESS` (409), * `RESUME_CHILD_PENDING` (409: the run waits on a `map` item run that is - * still suspended; the message names that run, resume it instead), + * still suspended; the message names that run, resume it instead — or + * whose suspension is gone without a `cancelled` history entry; the + * message names the remedy, cancelling the parent run), * `STORE_UNAVAILABLE` (503). */ resume: async ( diff --git a/packages/services/service-automation/README.md b/packages/services/service-automation/README.md index c8aea163c19..ba9fb148fb9 100644 --- a/packages/services/service-automation/README.md +++ b/packages/services/service-automation/README.md @@ -309,7 +309,9 @@ if (result.success) { machine-readable classification is `code` (not `errorCode`) — resume refusals such as `RUN_NOT_FOUND`, `STORE_UNAVAILABLE`, `RESUME_IN_PROGRESS`, `RESUME_CHILD_PENDING` (a parent-id resume of a `map` whose item run is still suspended; resume the item run the -message names), plus the trigger-time +message names. The same code answers when the item's suspension is gone without a +`cancelled` entry in its run history; cancel the parent run to end the batch. A cancelled +item is recorded in its slot instead), plus the trigger-time `FLOW_DISABLED` / `FLOW_NO_START_NODE`. ⚠️ `runAs` on `AutomationContext` is derived by the engine from the flow definition — diff --git a/packages/services/service-automation/src/builtin/map-node.ts b/packages/services/service-automation/src/builtin/map-node.ts index f6d0f4838fc..be1ff365724 100644 --- a/packages/services/service-automation/src/builtin/map-node.ts +++ b/packages/services/service-automation/src/builtin/map-node.ts @@ -13,6 +13,40 @@ import { collectionNotAList, resolveCollectionSlot, resolveValueSlotMap } from ' /** Hard cap on map fan-out — turns a runaway collection into a clean error. */ const MAX_MAP_ITEMS = 10_000; +/** + * [#27971] The result a `map` records in an item's slot when that item's run was + * cancelled (`cancelRun`) while the map waited on it, and the map's own run is + * then resumed. The item keeps its position, so `results[i]` is still item `i`; + * the marker is never a run or a success, and `count` leaves it out. + * + * Closed and `$`-keyed: a child flow's output keys are its output variable + * names, which never start with `$`, so no item output can look like it. + * Downstream nodes read it where they read the results: + * `results[i]['$status'] == 'cancelled'`. + */ +export interface MapItemCancelledMarker { + readonly $status: 'cancelled'; + /** The item's cancelled run, as its run history records it. */ + readonly $itemRunId: string; +} + +/** Whether a recorded result is the {@link MapItemCancelledMarker}. */ +export function isMapItemCancelledMarker(value: unknown): value is MapItemCancelledMarker { + if (value === null || typeof value !== 'object' || Array.isArray(value)) return false; + const keys = Object.keys(value); + const marker = value as Record; + return keys.length === 2 && marker.$status === 'cancelled' && typeof marker.$itemRunId === 'string'; +} + +/** + * The handoff variable the resume gate sets, on the map node, when it lets a + * resume continue past a cancelled item: `.$mapItemCancelled` holds the + * cancelled item run id. Engine-owned like `$mapItemDone`; a caller cannot write it. + */ +export function mapItemCancelledVariable(nodeId: string): string { + return `${nodeId}.$mapItemCancelled`; +} + /** * `map` built-in node — **sequential multi-instance** (ADR-0037 Track A2). * @@ -87,7 +121,7 @@ export function registerMapNode(engine: AutomationEngine, ctx: PluginContext): v // `LEDGER_DECLARED_NODE_CONFIG_SCHEMAS`), so this stays the open map // the Studio keyValue editor renders, like the CRUD `fields`. input: { type: 'object', additionalProperties: true, title: 'Item input', description: 'Params passed to each item’s subflow: each value a CEL value envelope evaluated per item (the item variable is in scope), or a literal.' }, - outputVariable: { type: 'string', title: 'Output variable', description: 'Each item’s subflow output, collected in order.' }, + outputVariable: { type: 'string', title: 'Output variable', description: 'Each item’s subflow output, collected in order. An item whose run was cancelled keeps its slot as { $status: \'cancelled\', $itemRunId }.' }, }, required: ['collection', 'flowName'], }, @@ -142,6 +176,15 @@ export function registerMapNode(engine: AutomationEngine, ctx: PluginContext): v variables.delete(`${node.id}.$mapItemDone`); variables.delete(`${node.id}.$mapItemOutput`); } + // [#27971] Re-entry past an item whose run was cancelled: the resume gate + // read `cancelled` in that run's history and says so here. The item keeps + // its slot with the marker, not an output — nothing ran, so no totals move. + const cancelledItemRunId = variables.get(mapItemCancelledVariable(node.id)); + if (typeof cancelledItemRunId === 'string' && cancelledItemRunId) { + const marker: MapItemCancelledMarker = { $status: 'cancelled', $itemRunId: cancelledItemRunId }; + state.results.push(marker); + } + variables.delete(mapItemCancelledVariable(node.id)); const parentRunId = variables.get('$runId'); @@ -329,7 +372,9 @@ export function registerMapNode(engine: AutomationEngine, ctx: PluginContext): v if (outVar) variables.set(outVar, state.results); return { success: true, - output: { results: state.results, count: state.results.length }, + // `count` is the items that RAN: a cancelled item's marker holds its + // slot in `results` but is never counted as a run (#27971). + output: { results: state.results, count: state.results.filter(r => !isMapItemCancelledMarker(r)).length }, metrics: { selected, acted, ...(unmeasured ? { unmeasuredEffect: true } : {}), ...rolledFailures() }, }; }, diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index c114672ffda..bfa8c2dd148 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -165,6 +165,7 @@ import { forgetSuspendedRun, evictConsumedSuspension, takeSubflowParentStrand, + type GatedResumeContinuation, } from './run-resume.js'; // [#25222] The dispatch and suspension claims' bodies: claiming, settling and // reading a dispatch, pruning the expired in-process claims, minting a run id, @@ -1547,7 +1548,7 @@ export function isRetryableResumeRefusal(code: AutomationResult['code']): boolea const ENGINE_BUILT_SIGNAL = Symbol('objectstack.automation.resume.engineBuilt'); /** Tag a signal the engine constructed for its own continuation. */ -function engineBuilt(signal: ResumeSignal): ResumeSignal { +export function engineBuilt(signal: ResumeSignal): ResumeSignal { return Object.assign(signal, { [ENGINE_BUILT_SIGNAL]: true }); } @@ -6467,6 +6468,13 @@ export class AutomationEngine implements IAutomationService { * still suspended answers a coded refusal naming that child's run id, and * nothing moves. Resuming the child is the door; the map continues when it * completes. See {@link refuseGatedResume}. + * + * **A cancelled item is recorded, never skipped (#27971).** When the item's + * run was cancelled (`cancelRun`, which leaves the parent parked), resuming + * the map's run advances it and the item keeps its slot in the results as + * `{ $status: 'cancelled', $itemRunId }`, outside `count`. An item whose + * suspension is gone WITHOUT a `cancelled` history row is refused with the + * same code: the map has nothing true to record for it. */ resume(runId: string, signal?: ResumeSignal): Promise { // The body lives in ./run-resume.ts (#25222). @@ -6503,7 +6511,10 @@ export class AutomationEngine implements IAutomationService { * Authority is judged first, so a service-owned item keeps its 403; an item * open to this route is refused with a coded 409 naming the child run id to * resume instead, before any state is touched. A `subflow` link is not a - * skip — the signal is delegated to the child — and is unchanged. + * skip — the signal is delegated to the child — and is unchanged. A map + * whose item row is gone answers by that item's run history (#27971): + * `cancelled` hands {@link resume} a continuation that records the item's + * marker; anything else is refused with the same code. * * **An UNDECLARED node type is refused, not deferred** (#5561 step two). It * used to be let through on the schema default's inherited `'any'`; a pause @@ -6517,13 +6528,14 @@ export class AutomationEngine implements IAutomationService { * never declared — the failure mode this whole issue is about, restated as a * log line. */ - private refuseGatedResume(runId: string, signal?: ResumeSignal): Promise { + private refuseGatedResume(runId: string, signal?: ResumeSignal): Promise { // The body lives in ./run-resume.ts (#25222). const host: ThisParameterType = { resolveSuspensionChain: (...a) => this.resolveSuspensionChain(...a), resolveSuspendedNodeType: (...a) => this.resolveSuspendedNodeType(...a), resolveResumeAuthority: (...a) => this.resolveResumeAuthority(...a), resolveDeclaredResumeAuthority: (...a) => this.resolveDeclaredResumeAuthority(...a), + getRun: (...a) => this.getRun(...a), logger: this.logger, }; return refuseGatedResume.call(host, runId, signal); diff --git a/packages/services/service-automation/src/map-cancelled-item-resume.test.ts b/packages/services/service-automation/src/map-cancelled-item-resume.test.ts new file mode 100644 index 00000000000..36bc4e64455 --- /dev/null +++ b/packages/services/service-automation/src/map-cancelled-item-resume.test.ts @@ -0,0 +1,239 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * A resume of a run parked on a `map` node whose item run was CANCELLED records + * that item as cancelled, in its own slot, and the map goes on (#27971). + * + * `cancelRun` on the item terminates it and never touches the parent, so the + * parent stays parked on `map:` with nothing to wake it. Measured on + * the in-process door before this change: a parent-id resume re-ran the map, + * which advanced past the item with nothing recorded — `results` held only the + * later items' outputs, `results[0]` was no longer item 0, and the parent + * finished `completed`. + * + * The rules pinned here: the cancelled item's slot holds the closed marker + * `{ $status: 'cancelled', $itemRunId }`; `count` leaves it out; a downstream + * node reads it where it reads the results; an item whose suspension is gone + * WITHOUT a `cancelled` history row (or whose history cannot be read) is + * refused with #27877's code, nothing moved; a caller still cannot write the + * handoff variable. The still-suspended refusal and the ordinary up-bubble are + * the controls. + */ + +import { describe, it, expect, beforeEach } from 'vitest'; +import { defineActionDescriptor } from '@objectstack/spec/automation'; +import { AutomationEngine } from './engine.js'; +import { registerMapNode } from './builtin/map-node.js'; +import { registerSubflowNode } from './builtin/subflow-node.js'; +import { MAP_ITEM_PENDING_REFUSAL_CODE } from './run-resume.js'; + +function silentLogger(): any { + return { info() {}, warn() {}, error() {}, debug() {}, child() { return silentLogger(); } }; +} +function pluginCtx(): any { + return { logger: silentLogger(), getService() { throw new Error('none'); } }; +} + +/** start → `open_pause` (raw-resumable) → `after` → end; `after` outputs which item ran. */ +const ITEM_FLOW = { + name: 'one_item', + label: 'one_item', + type: 'autolaunched', + variables: [{ name: 'tag', type: 'text', isInput: true }, { name: 'done', type: 'text', isOutput: true }], + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'pause', type: 'open_pause', label: 'Pause' }, + { id: 'after', type: 'after', label: 'After' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'pause' }, + { id: 'e2', source: 'pause', target: 'after' }, + { id: 'e3', source: 'after', target: 'end' }, + ], +}; + +/** start → map → `peek` (the downstream reader) → end. */ +const BATCH_FLOW = { + name: 'batch_flow', + label: 'batch_flow', + type: 'autolaunched', + variables: [{ name: 'items', type: 'list', isInput: true }], + nodes: [ + { id: 'hs', type: 'start', label: 'Start' }, + { + id: 'signoffs', type: 'map', label: 'Sign off each', + config: { + collection: { dialect: 'cel', source: 'items' }, flowName: 'one_item', outputVariable: 'results', + input: { tag: { dialect: 'cel', source: 'item.id' } }, + }, + }, + { id: 'pk', type: 'peek', label: 'Peek' }, + { id: 'he', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'h1', source: 'hs', target: 'signoffs' }, + { id: 'h2', source: 'signoffs', target: 'pk' }, + { id: 'h3', source: 'pk', target: 'he' }, + ], +}; + +describe('a parent-id resume of a map whose item run was cancelled (#27971)', () => { + let engine: AutomationEngine; + let seen: { results: unknown; nodeResults: unknown; count: unknown } | undefined; + + beforeEach(() => { + seen = undefined; + engine = new AutomationEngine(silentLogger()); + engine.registerNodeExecutor({ + type: 'open_pause', + descriptor: defineActionDescriptor({ + type: 'open_pause', version: '1.0.0', name: 'Open Pause', supportsPause: true, resumeAuthority: 'any', + }), + async execute() { return { success: true, suspend: true }; }, + }); + engine.registerNodeExecutor({ + type: 'after', + async execute(_node, variables) { variables.set('done', `ran ${String(variables.get('tag'))}`); return { success: true }; }, + }); + engine.registerNodeExecutor({ + type: 'peek', + async execute(_node, variables) { + seen = { results: variables.get('results'), nodeResults: variables.get('signoffs.results'), count: variables.get('signoffs.count') }; + return { success: true }; + }, + }); + registerMapNode(engine, pluginCtx()); + registerSubflowNode(engine, pluginCtx()); + engine.registerFlow('one_item', ITEM_FLOW as never); + engine.registerFlow('batch_flow', BATCH_FLOW as never); + }); + + const launch = () => engine.execute('batch_flow', { params: { items: [{ id: 't1' }, { id: 't2' }] } }); + const parked = (flowName: string) => engine.listSuspendedRuns().filter(r => r.flowName === flowName); + const mapState = (runId: string) => + (engine as any).suspendedRuns.get(runId)?.variables?.['signoffs.$mapState']; + + it('records the cancelled item in its slot, outside count, and the map goes on to the next item', async () => { + const paused = await launch(); + const [item1] = parked('one_item'); + expect(await engine.cancelRun(item1!.runId, 'operator')).toBe(true); + + const resumed = await engine.resume(paused.runId!, {}); + + expect(resumed.success).toBe(true); + expect(resumed.status).toBe('paused'); + // Item 1's slot holds the marker; the map parked on item 2. + expect(mapState(paused.runId!)).toEqual({ + started: 2, results: [{ $status: 'cancelled', $itemRunId: item1!.runId }], + }); + const [item2] = parked('one_item'); + expect(item2!.runId).not.toBe(item1!.runId); + + const finished = await engine.resume(item2!.runId, {}); + expect(finished.success).toBe(true); + + // The downstream node reads the marker where it reads the results, and + // `count` is the one item that RAN. + const marker = { $status: 'cancelled', $itemRunId: item1!.runId }; + expect(seen).toEqual({ results: [marker, { done: 'ran t2' }], nodeResults: [marker, { done: 'ran t2' }], count: 1 }); + expect(engine.listSuspendedRuns()).toEqual([]); + expect((await engine.getRun(paused.runId!))?.status).toBe('completed'); + }); + + it('records it one level down when the resumed id is a subflow parent of the map', async () => { + engine.registerFlow('outer_flow', { + name: 'outer_flow', label: 'outer_flow', type: 'autolaunched', + nodes: [ + { id: 'os', type: 'start', label: 'Start' }, + { id: 'call', type: 'subflow', label: 'Call batch', config: { flowName: 'batch_flow', input: { items: [{ id: 't1' }] } } }, + { id: 'oe', type: 'end', label: 'End' }, + ], + edges: [{ id: 'o1', source: 'os', target: 'call' }, { id: 'o2', source: 'call', target: 'oe' }], + } as never); + const outer = await engine.execute('outer_flow'); + const [item1] = parked('one_item'); + await engine.cancelRun(item1!.runId); + + const resumed = await engine.resume(outer.runId!, {}); + + expect(resumed.success).toBe(true); + expect(seen).toEqual({ + results: [{ $status: 'cancelled', $itemRunId: item1!.runId }], + nodeResults: [{ $status: 'cancelled', $itemRunId: item1!.runId }], + count: 0, + }); + expect(engine.listSuspendedRuns()).toEqual([]); + }); + + it('refuses, nothing moved, when the item row is gone without a cancelled history row', async () => { + const paused = await launch(); + const [item1] = parked('one_item'); + // The suspension is gone but nothing cancelled it: its history still says + // `paused` (a consumed strand, or a child that completed without bubbling). + (engine as any).suspendedRuns.delete(item1!.runId); + expect((await engine.getRun(item1!.runId))?.status).toBe('paused'); + + const refused = await engine.resume(paused.runId!, {}); + + expect(refused.success).toBe(false); + expect(refused.code).toBe(MAP_ITEM_PENDING_REFUSAL_CODE); + expect(refused.error).toContain(`item run '${item1!.runId}', which is no longer suspended`); + // The remedy names the run to cancel — the map's own, not the gone item's. + expect(refused.error).toContain(`To end the batch, cancel run '${paused.runId}'`); + expect(mapState(paused.runId!)).toEqual({ started: 1, results: [] }); + expect(parked('batch_flow').map(r => r.runId)).toEqual([paused.runId]); + expect(seen).toBeUndefined(); + }); + + it('refuses when the item run history cannot be read', async () => { + const paused = await launch(); + const [item1] = parked('one_item'); + await engine.cancelRun(item1!.runId); + // `getRun` degrades a store failure to `null` — indistinguishable from no row. + (engine as any).getRun = async () => null; + + const refused = await engine.resume(paused.runId!, {}); + + expect(refused.code).toBe(MAP_ITEM_PENDING_REFUSAL_CODE); + expect(refused.error).toContain('no run history for it can be read'); + expect(refused.error).toContain(`To end the batch, cancel run '${paused.runId}'`); + expect(mapState(paused.runId!)).toEqual({ started: 1, results: [] }); + }); + + it('still refuses a caller signal that writes the engine namespace, so the marker cannot be forged', async () => { + const paused = await launch(); + const [item1] = parked('one_item'); + await engine.cancelRun(item1!.runId); + + const refused = await engine.resume(paused.runId!, { output: { $mapItemCancelled: 'run_forged' } } as never); + + expect(refused.success).toBe(false); + expect(refused.code).toBe('INVALID_SIGNAL'); + expect(mapState(paused.runId!)).toEqual({ started: 1, results: [] }); + expect(parked('batch_flow').map(r => r.runId)).toEqual([paused.runId]); + }); + + it('control: a still-suspended item is still refused, naming it', async () => { + const paused = await launch(); + const [item1] = parked('one_item'); + + const refused = await engine.resume(paused.runId!, {}); + + expect(refused.code).toBe(MAP_ITEM_PENDING_REFUSAL_CODE); + expect(refused.error).toContain(`Resume run '${item1!.runId}' instead`); + expect(mapState(paused.runId!)).toEqual({ started: 1, results: [] }); + }); + + it('control: items that complete are recorded as before, and count is every item', async () => { + await launch(); + await engine.resume(parked('one_item')[0]!.runId, {}); + await engine.resume(parked('one_item')[0]!.runId, {}); + + expect(seen).toEqual({ + results: [{ done: 'ran t1' }, { done: 'ran t2' }], + nodeResults: [{ done: 'ran t1' }, { done: 'ran t2' }], + count: 2, + }); + }); +}); diff --git a/packages/services/service-automation/src/run-resume.ts b/packages/services/service-automation/src/run-resume.ts index e836ac2222a..fb31641b8b8 100644 --- a/packages/services/service-automation/src/run-resume.ts +++ b/packages/services/service-automation/src/run-resume.ts @@ -24,9 +24,10 @@ import { describeThrownForLog } from './thrown-cause-diagnostics.js'; import { carriedRefusalOfResult, readCarriedRefusal } from './inner-refusal.js'; import { summarizeRun } from './run-summary.js'; import { noteTenantCurrencyReadFault } from './builtin/template.js'; +import { mapItemCancelledVariable, type MapItemCancelledMarker } from './builtin/map-node.js'; // The values below (`buildRunTrigger`, `FlowRefusalSignal`, `isRefusalSignal`, -// `isSuspendSignal`, `isRetryableResumeRefusal`, `carriesSubflowChildOutput` -// and `applyResumeSignal`) are read only inside the function bodies, at call +// `isSuspendSignal`, `isRetryableResumeRefusal`, `carriesSubflowChildOutput`, +// `applyResumeSignal` and `engineBuilt`) are read only inside the function bodies, at call // time, never while a module is evaluating, so the import cycle with // ./engine.js is safe under ESM live bindings; the tsup bundle is one file and // has no cycle. @@ -38,6 +39,7 @@ import { isRetryableResumeRefusal, carriesSubflowChildOutput, applyResumeSignal, + engineBuilt, type AutomationEngine, type ChildRunRefusal, type ExecutionLogEntry, @@ -75,6 +77,7 @@ export interface RunResumeHost { readonly resolveSuspendedNodeType: AutomationEngine['resolveSuspendedNodeType']; readonly resolveResumeAuthority: AutomationEngine['resolveResumeAuthority']; readonly resolveDeclaredResumeAuthority: AutomationEngine['resolveDeclaredResumeAuthority']; + readonly getRun: AutomationEngine['getRun']; readonly resolveCanonicalDescriptor: AutomationEngine['resolveCanonicalDescriptor']; readonly loadSuspendedRunStrict: AutomationEngine['loadSuspendedRunStrict']; readonly store: AutomationEngine['store']; @@ -388,8 +391,9 @@ export async function expandDeclaredLookups(this: Pick, runId: string, signal?: ResumeSignal): Promise { - const refusal = await this.refuseGatedResume(runId, signal); - if (refusal) return refusal; + const gated = await this.refuseGatedResume(runId, signal); + if (gated && 'continueWith' in gated) return this.resumeInternal(runId, gated.continueWith, false); + if (gated) return gated; // An ABSENT signal is an EMPTY caller submission, never an exemption // (commit 7307191db). This is the in-process door, and `resume(runId)` used to // skip the screen contract that `resume(runId, {})` is held to: @@ -420,8 +424,19 @@ export async function resume(this: Pick = 'RESUME_CHILD_PENDING'; +/** + * [#27971] What the gate hands {@link resume} when a resume may go on past a + * `map` item whose run was cancelled: the signal to continue with, which tells + * that map to record the item as cancelled. Not a refusal, and not a plain + * pass either — the caller's own signal alone would advance the map past the + * item with nothing recorded. + */ +export interface GatedResumeContinuation { + readonly continueWith: ResumeSignal; +} + /** The body of {@link AutomationEngine.refuseGatedResume}; that method's docblock is its contract. */ -export async function refuseGatedResume(this: Pick, runId: string, signal?: ResumeSignal): Promise { +export async function refuseGatedResume(this: Pick, runId: string, signal?: ResumeSignal): Promise { const chain = await this.resolveSuspensionChain(runId); const run = chain[chain.length - 1]; if (!run) return null; @@ -433,7 +448,7 @@ export async function refuseGatedResume(this: Pick, runId: string, c return null; } +/** + * [#27971] The resume of a run whose deepest link is a `map:` one with no + * successor in `chain` — the map is parked on an item whose suspension row is + * gone. Re-running the map would advance it past that item, so what the item's + * run history says decides the answer: + * + * - `cancelled` (an operator's `cancelRun`, which never touches the parent): + * the map may go on, and the continuation tells it to record the item's + * {@link MapItemCancelledMarker} in its slot; + * - anything else — no history row, an unreadable store (`getRun` degrades to + * `null`), a run that completed without bubbling, a consumed strand: the map + * cannot record what happened to the item, so the resume is refused with + * #27877's code before any state is touched. + * + * `null` when the chain ends anywhere but on such a link parked at the built-in + * `map` node (the one executor that reads the handoff; another type that mints + * a `map:` correlation records nothing from it), and when the caller's signal + * writes the engine's `$` namespace: that bag is refused `INVALID_SIGNAL` by + * `resumeInternal` with the run left parked, and folding it into an + * engine-built signal would exempt it from that guard. + */ +async function continuePastCancelledMapItem( + this: Pick, + runId: string, + chain: readonly SuspendedRun[], + nodeType: string | undefined, + signal?: ResumeSignal, +): Promise { + const link = chain[chain.length - 1]!; + if (nodeType !== 'map') return null; + if (typeof link.correlation !== 'string' || !link.correlation.startsWith('map:')) return null; + const itemRunId = link.correlation.slice('map:'.length); + if (!itemRunId) return null; + + const item = await this.getRun(itemRunId); + if (item?.status !== 'cancelled') { + const at = link.runId === runId ? `'${link.nodeId}'` : `'${link.nodeId}' (linked run '${link.runId}')`; + const found = item ? `its run history records '${item.status}', not 'cancelled'` : 'no run history for it can be read'; + this.logger.warn( + `[automation] refused resume of run '${runId}': parked on map node ${at}, whose item run ` + + `'${itemRunId}' is no longer suspended and ${found} — resuming here would advance the map past ` + + `that item with nothing recorded for it`, + ); + return { + success: false, + code: MAP_ITEM_PENDING_REFUSAL_CODE, + error: + `Run '${runId}' is ${link.runId === runId ? '' : `waiting on run '${link.runId}', which is `}paused at ` + + `map node '${link.nodeId}' on item run '${itemRunId}', which is no longer suspended, and ${found}. ` + + `The map records an item only when it completes or when its run was cancelled, so resuming here ` + + `would advance the map past that item with nothing recorded for it. Nothing was changed. To end the ` + + `batch, cancel run '${link.runId}'${link.runId === runId ? '' : ` and run '${runId}', which waits on it`}; ` + + `if the run history store was unreachable, retry once it is readable`, + }; + } + + if (signal && applyResumeSignal(new Map(), signal, link.nodeId).length > 0) return null; + return { + continueWith: engineBuilt({ + ...(signal ?? {}), + variables: { ...(signal?.variables ?? {}), [mapItemCancelledVariable(link.nodeId)]: itemRunId }, + }), + }; +} + /** The body of {@link AutomationEngine.refuseUndeclaredSuspension}, moved unedited; that method's docblock is its contract. */ export function refuseUndeclaredSuspension(this: Pick, nodeType: string): NodeExecutionResult | null { const descriptor = this.resolveCanonicalDescriptor(nodeType); diff --git a/packages/spec/src/api/error-code-ledger.zod.ts b/packages/spec/src/api/error-code-ledger.zod.ts index c1a6a563db7..3754c1e4b15 100644 --- a/packages/spec/src/api/error-code-ledger.zod.ts +++ b/packages/spec/src/api/error-code-ledger.zod.ts @@ -1323,7 +1323,7 @@ export const ERROR_CODE_LEDGER = { // line — no HTTP door; registered under `@objectstack/metadata-protocol` // too, so this row is provenance, not identity. 'NOT_OVERRIDABLE', - 'RESUME_CHILD_PENDING', // [#27877] parent-id resume of a map whose item child run is still suspended (409; resume the named child) + 'RESUME_CHILD_PENDING', // [#27877] parent-id resume of a map whose item child run is still suspended (409; resume the named child) — or [#27971] whose item row is gone without a `cancelled` history (409; cancel the parent run) 'RESUME_IN_PROGRESS', // duplicate resume refused while the first is running 'RUN_NOT_FOUND', // no suspension for this run id — unresumable for good 'STORE_UNAVAILABLE', // durable suspended-run store unreadable — existence unknown diff --git a/packages/spec/src/automation/builtin-node-config.zod.ts b/packages/spec/src/automation/builtin-node-config.zod.ts index f878eaf3ca0..a3f76b6652a 100644 --- a/packages/spec/src/automation/builtin-node-config.zod.ts +++ b/packages/spec/src/automation/builtin-node-config.zod.ts @@ -1487,7 +1487,7 @@ export const MapConfigSchema = lazySchema(() => strictObject({ ), /** Variable the ordered list of per-item outputs is bound to — never a `$` name: those are the engine's (#22502). */ outputVariable: flowBoundVariableNameSchema('outputVariable').optional() - .describe("Each item's subflow output, collected in order — bound to a name without a leading `$` (the `$` names are the flow engine's own), read as `{{ name }}`"), + .describe("Each item's subflow output, collected in order — bound to a name without a leading `$` (the `$` names are the flow engine's own), read as `{{ name }}`. An item whose run was cancelled keeps its slot as the closed marker `{ $status: 'cancelled', $itemRunId }`: never an output, and not counted as a run"), })); export type MapConfig = z.input; diff --git a/packages/spec/src/contracts/automation-service.ts b/packages/spec/src/contracts/automation-service.ts index 2d27992419e..df9f3bdf7fc 100644 --- a/packages/spec/src/contracts/automation-service.ts +++ b/packages/spec/src/contracts/automation-service.ts @@ -531,6 +531,13 @@ export interface AutomationResult { * to **409**. Distinct from `'RESUME_IN_PROGRESS'` on purpose: that one is * benign and needs nothing from the caller, this one is answered only by * resuming a different run id. + * The same code answers the map's second refusal (#27971): the item's + * suspension is gone and its run history does not record it as + * `cancelled` (no entry, an unreadable store, or another status), so the + * map has nothing true to record for that item. Nothing moves; `error` + * names the item run and the remedy, which is cancelling the parent run + * to end the batch, or retrying once the store is readable. A cancelled + * item is not refused: the map records it and goes on. * - `'INVALID_SCREEN_INPUT'` — the run is parked on a `screen` node and * the submitted bag violates that screen's declared field contract: a * `required` field the caller WAS asked for is missing, or a key the