Execution

How a MachinePlan is run: what runs when, what a step is sent and what is collected from it.

execute_plan

flowchart TD
    MP["MachinePlan"] --> VAL["1. Validate the plan<br/>and what the run binds"]
    VAL --> SC["2. Scopes: the machine,<br/>and each map's body"]
    SC --> UN["3. Units of a scope:<br/>chains of steps, and maps"]
    UN --> RUN["4. Start each unit when<br/>everything it reads is settled"]
    RUN --> SEG["5. A chain is one segment,<br/>run by the EngineRuntime"]
    SEG --> RES["6. PipelineResult:<br/>terminals, step reports, verdict"]

execute_plan(plan, runtime, inputs, input_is_sequence, arguments, callbacks) runs one plan:

1. Validate. The plan is held to its own invariants (MachinePlan::validate), and the run to the plan: every InputSlot is given exactly one value, and nothing is given that the plan does not take.
2. Scopes. The machine is a scope, and so is the body of each ForEach (Orchestrator).
3. Units. The steps of a scope are divided into chains (Pipelined execution); each map directly inside the scope is a unit of its own.
4. Schedule. A unit starts as soon as everything it reads is settled (Scheduling).
5. Run. A chain is converted to a ResolvedGraph and run as one segment by the EngineRuntime: cartridges are discovered and hosted, each step’s request is opened, fed and collected (One step’s request). A map runs its body once per item (Maps).
6. Result. A PipelineResult: one PlanTerminal per Output node, the outcome of every run of every map, a StepReport for every step that did not simply complete, and the run’s verdict.

The EngineRuntime is the only thing that differs between hosts: how cartridges are hosted, whether terminal output is persisted, and what a failed run of a map does (Maps). Scheduling, maps and result assembly are the same everywhere.

Source: capdag-rs/src/orchestrator/execute_plan.rs, executor.rs.

Cartridge discovery

Cartridge resolution for each cap URN in the graph:

1. Dev cartridges (local override): Check dev_cartridges (a map of local binary paths) using is_dispatchable matching. This lets developers test local builds without publishing.
2. Cartridge registry (remote): Fall back to the cartridge registry for published cartridges. The cartridge registry is versioned: a binary bakes a CARTRIDGE_REGISTRY_VERSION (currently 2), and the published manifest URL carries a version prefix segment (.../v2/manifest), with the bare /manifest denoting the frozen v0 state. On disk, cartridges live at {registry_slug}/v{cartridge_registry_version}/{channel}/{name}/{version}/, and discovery pins the v{version} directory to the host’s baked version exactly as it pins the channel, so a v2 cartridge is invisible to a v1 host. The registry_slug is a path-safe transform of the registry URL’s authority (host, plus :port): the substring after :// up to the first /, ?, or #, ASCII-lowercased, with every character outside [a-z0-9.-] replaced by - (a null registry URL becomes the literal dev); it is not a hash, and it depends only on the authority, so every version of one registry shares a slug. A cartridge registry is ingested only if its manifest’s fabricRegistryUrl matches the client’s own baked fabric registry, so every registry a client uses shares one fabric and keeps caps comparable. The local fabric cache is keyed by registry version, so a staging re-publish of the same version on a mutable channel can serve stale bytes; resolve --no-cache and capdag cache clear|refresh bypass or renew it.
3. Download and verify: If the binary is not cached locally, download it and verify its SHA256 hash.
4. Register: Add the cartridge binary to the CartridgeHostRuntime.

All manifest caps are registered because peer invocations from one cartridge may target caps in another cartridge that are not part of the DAG. For example, an ML cartridge’s handler may call modelcartridge to download a model; this cap is not in the DAG but must be routable.

Source: executor.rs.

A step’s inputs

One step is one request. Each input of the step is its own STREAM_START/CHUNK/STREAM_END sequence within that request, delivered under the URN the cartridge knows it by: the cap’s in= for the main input, the arg’s URN for any other. A gather is one stream: its members’ values in binding order, as one sequence. Values an arg is given by the run (the argument ledger) and constants written in the machine are sent in the same request.

Source: executor.rs.

Scheduling

A unit is ready when every node it reads has been produced or is known to be absent. Ready units start at once; units with no path between them run at the same time, in the machine as in a map’s body. Kahn’s algorithm over the steps gives the dependency order and rejects a cycle; it does not serialize them.

• What limits concurrency is admission, and only admission. A started step waits for its cap’s whole pool chain like any dispatch (Error handling). Two steps whose chains share a bounded pool still run one after the other, in arrival order. There is no executor-wide limit to configure.
• Results do not depend on timing. A gather delivers its members in binding order and a map collects in item order, whichever finished first.
• One failure ends its siblings. Each scope has a failure signal. When a unit fails for good and nothing tolerates the failure, the signal of its scope is raised: every other live request in that scope, and in every scope inside it, is cancelled as collateral before anything is joined (see Error handling). A failure inside a map’s run raises that run’s signal first; whether it reaches the map’s scope is the map’s to say (Maps).

Source: executor.rs, execute_plan.rs.

Maps

execute_plan runs a ForEach’s body once per item, the runs concurrently, and collects each node of the body that something outside reads into a sequence in item order.

• Zip. A map over several sequences takes item i of each for run i. Sequences whose lengths are known and differ fail before any run starts; sequences read as they arrive fail when one ends while another still has items. Either way the run fails with ExecutionError::ZipLengthMismatch, naming each source node and how many items it gave.
• Whole values. A node from outside the map, a constant, and a sequence bound to a sequence arg are delivered whole to every run. That includes the sequence the map iterates.
• Nested maps. A sequence produced inside a body and bound to an arg that takes one value is a map inside the map: its body runs once per item, per run of the outer body. Read from inside the outer body, what the inner map collected is that run’s sequence (a step that takes a sequence folds it there, once per outer run). Read from outside both, the inner sequences are appended, in outer item order: one flat sequence.
• Addressing. A run of a body is addressed by the map’s step token and its index, and by the run of each enclosing map it is inside (ForEachBodyCoordinate::enclosing). The argument ledger journals a body’s dispatch under that address.
• A run that fails. When a step of a body fails for good and its own policy does not tolerate it, the host’s MapFailurePolicy decides. Fail: the map fails, and with it the run, as any step’s failure would. DropItem: that item contributes nothing to what the map collects, its failure is recorded on its body outcome, the other runs go on, and the run ends completed with tolerated failures; a map every run of which failed still fails. The capdag CLI runs maps with Fail.

Absent values and skipped steps

A node is absent when the output that would have produced it did not fire: an optional output the cartridge opened no stream for, an output of a skipped step, or an output of a step that failed under on_error=absent.

• A step with a required arg bound to an absent node is skipped: no request is opened, it is reported skipped (a StepReport naming the absent nodes), and every output it would have produced is absent.
• A non-required arg bound to an absent node is sent its default, or not sent when it has none.
• A gather sends the members that are present, in order; with none present it sends an empty sequence.
• In a map, a run whose terminal step was skipped, or whose taken output did not fire, adds nothing to that output’s collected sequence.
• A plan Output whose node is absent is returned as PlanTerminal::NotProduced, which is neither an empty value nor an error.

Absence is decided when the producing request ends (END with no stream for that output), so a hop out of an optional output is never pipelined: the consumer is opened only once the output has fired.

One step’s request

What the executor does for a single cap invocation:

Input collection

Every node the step reads is settled before its request is opened: its value was produced by an earlier unit, by the step before it in the chain (which streams to it live), or was given to the run. A node that is neither produced nor absent when the step is reached is a defect in the schedule and fails with ExecutionError::NoIncomingData.

sequenceDiagram
    participant EX as Executor
    participant SW as RelaySwitch
    participant P as Cartridge

    EX->>SW: register_external_request(rid, cap_urn)
    Note over SW: Returns (XID, response_rx)

    loop For each input of the step
        EX->>SW: STREAM_START (stream_id, media_urn)
        EX->>SW: CHUNK(s)
        EX->>SW: STREAM_END
    end
    EX->>SW: END

    Note over SW: Routes to cartridge via cap matching
    SW->>P: (forwarded)

    loop select! loop
        alt Response frame on rx
            P-->>EX: STREAM_START → which output this stream is
            P-->>EX: CHUNK → that output's data
            P-->>EX: LOG → forward progress
            P-->>EX: END → done
            P-->>EX: ERR → fail
        else Pump timeout (200ms)
            Note over EX: read_from_masters_timeout<br/>Routes peer frames through switch
        else Activity timeout (120s)
            Note over EX: ExecutionError::ActivityTimeout
        end
    end

Cap invocation

1. Call RelaySwitch::register_external_request(rid, cap_urn) to set up routing. Returns (XID, response_rx): the XID for frame routing and an mpsc channel for receiving response frames.
2. Send input streams: For each input, send STREAM_START (with stream_id and media_urn) → CHUNK(s) (with payload split at max_chunk boundaries) → STREAM_END.
3. Send END to signal all inputs are delivered.

Response collection loop

The response is collected with a tokio::select! loop that does two things concurrently:

tokio::select! {
    pump_result = read_from_masters_timeout(200ms) => {
        // Route peer frames through the RelaySwitch
    }
    Some(frame) = rx.recv() => {
        last_activity = Instant::now();  // reset timeout
        match frame.frame_type {
            StreamStart => { /* which declared output this stream is */ }
            Chunk => { /* that output's data: stored, forwarded, or dropped */ }
            End   => { /* response complete */ }
            Log   => { /* forward progress to callback */ }
            Err   => { /* execution failed */ }
        }
    }
}

The concurrent pump (read_from_masters_timeout) is necessary. Without it, a cartridge doing a peer call during request handling would deadlock: the peer request frames would arrive at the switch but have no path to the target cartridge, because the executor’s response loop is only reading its own response channel, not pumping frames through the switch.

The 200ms timeout on read_from_masters_timeout ensures the loop checks both the response channel and the frame pump regularly.

Activity timeout

DEFAULT_ACTIVITY_TIMEOUT_SECS = 120. Any frame received on rx.recv() resets last_activity. If no frame arrives for 120 seconds, the executor aborts with ExecutionError::ActivityTimeout.

While idle, the executor logs a warning every 30 seconds to aid debugging.

Individual caps can override the default timeout via the activity_timeout_secs key in their metadata. This is checked before the default is applied.

Source: executor.rs (DEFAULT_ACTIVITY_TIMEOUT_SECS, ACTIVITY_TIMEOUT_METADATA_KEY).

Progress subdivision

Every unit of a scope has a share of the scope’s progress in proportion to the steps it holds, and the run’s progress is the sum of what each has contributed:

run_progress = sum over units of  share * unit_progress      (unit_progress in [0, 1])

A unit contributes nothing before it starts, its mapped cartridge progress while it runs, and its whole share when it ends: completed, skipped, or failed and tolerated. Several units running at once each move their own term, so the total never goes down and never depends on which finished first. A map’s share is divided among its runs in the same way, and the map reports the aggregate of its runs under its own step token. See Progress mapping for the mapping formula.

Error handling

pub enum ExecutionError {
    StepFailed { step_token_id, source },            // any failure, attributed to its immutable strand step
    CartridgeNotFound { cap_urn },                    // no cartridge provides the cap
    CartridgeDownloadFailed(String),
    CartridgeExecutionFailed { cap_urn, code, class, details, arg_urn }, // cartridge returned ERR
    EffectContractViolation { cap_urn, effect, runtime_input, expected, actual },
    StreamContractViolation { cap_urn, declared_is_sequence, declared_streaming, emitted_is_sequence, emitted_unbounded },
    NoIncomingData { node },                          // missing source node data
    IoError(std::io::Error),
    HostError(String),
    CartridgeUnavailable { cap_urn, details },
    FabricRegistryError(String),
}

A failed step ends the run unless its policy says otherwise (Machine notation):

• retry=N. A failure whose attribution class is retryable (resource, environment, internal; Error handling) is attempted again, up to N further times, after a delay that backs off with jitter (net_retry::RetryPolicy). input and user are never retried. A step with a policy is a chain of its own: nothing it emits is forwarded to another step until an attempt has ended successfully, and what a failed attempt emitted, to a file as much as to memory, is discarded. Every attempt reads the argument values the first one read; an update that arrives between attempts is reported already-dispatched. Each attempt is recorded on the step with its class, code and message.
• on_error=absent. When the step has failed for good, its outputs are absent (Absent values and skipped steps), the step is reported failed-tolerated with the failure it had, and the run goes on. A cancel is never tolerated.
• Neither. The failure is terminal for the run, attributed to its step.

A run therefore ends in one of four ways: completed; completed with tolerated failures (every output that could be produced was, and at least one step is failed-tolerated or one item of a map was dropped); failed; cancelled. The first two are the RunVerdict of a PipelineResult; the other two are the error execute_plan returns. A host reports the second as its own verdict, never as success.

ExecutionError gains ZipLengthMismatch { sources } and OutputContractViolation { cap_urn, detail } (a response stream that names no declared output, a second stream for one output, a required output with no stream, or a carried type that fails its audit; detail says which, naming the output).

Teardown before join. When any task of a pipelined chain fails (the input feed, an intermediate forward (an upstream ERR), or the terminal collect), the chain is torn down first and joined second: every still-live request of the chain is cancelled as collateral: a Cancel attributed to the originating failure’s class, code ABORTED_COLLATERAL, message naming the failing step (12.2 §Cancel). Its live feeds close, its credit waiters release; only then are the feed/forward tasks awaited, and they end promptly because their requests are gone. The chain returns the originating error, attributed at its step with its declared class; the consequences the joined tasks report are logged, never returned. A live source upstream of a failed cap can therefore never hold a run in RUNNING, and the run’s record is the real failure, not “cancelled”. The fault is observed the moment it happens: the send/forward tasks report their typed error on a chain fault channel that the terminal collect races (tokio::select!), so the collector does not wait for an END that cannot come.

Pipelined execution

The reference executor pipelines chains: maximal runs of steps in which each step’s response streams live into the next step’s main input. A step continues the chain of the step that produces its input when all of these hold:

• it has exactly one input stream, and that is the producer’s main output as it is: not a gather, and not another output of the producer;
• nothing else reads anything the producer produces: no other step, no other taken output, and nothing outside the steps being run (a machine output, a step of another scope);
• the producer’s main output is not optional: an output that may not fire is known to be present or absent only when its request ends;
• neither step has a policy: a step that may be attempted again, or whose failure is tolerated, runs alone;
• the two caps’ pool chains share no bounded pool, and the hop crosses no streaming contract boundary (§Streaming contracts below: a producer that may emit without a length promise into a consumer that does not consume without one).

Every other step heads a chain and is fed from materialized values. Within a chain, all cap invocations open up front and per-hop forwarding tasks (forward_frames) stream frames from cap i to cap i+1 live, so the downstream cap receives its first item before the upstream cap has emitted its last. Credit flows per hop (acquire from the downstream cap’s window, grant the upstream cap as consumed, L10/L14), and the head step’s input feed runs concurrently with terminal collection (L15).

A shared bounded pool between two adjacent caps is an execution boundary because an admission permit represents ownership of concrete pool slots from REQ through terminal response. The executor runs the first cap to completion, materializes its output, and only then acquires the dependent invocation. It must not pre-acquire both: adjacent caps whose chains meet in a capacity-one pool would leave the first invocation unfed while acquisition of the second waits for the slot the first holds. Two same-cartridge caps in DISJOINT bounded pools share no permit and keep streaming. This boundary preserves FIFO admission and process-loss isolation; it is not a retry or fallback.

Every output STREAM_START is audited against the producing cap’s declared contracts at receipt, at all three receipt points: the pipelined hop (forward_frames), the batch terminal collector (collect_terminal_output), and the incremental terminal consumer (TerminalOutput). The audit (ResponseAudit::audit_frame) holds the stream to the output it names (ResponseContract, every output the cap definition declares), in order:

1. which output it is: a stream that names no output is the main output’s; a stream that names one (frame key 21, matched as a tagged URN) is that declared output’s, and never the main one. The output must have no stream yet. A main-output stream from a cap that declares no output, a stream naming an undeclared output or the main one, and a second stream for one output are each ExecutionError::OutputContractViolation;
2. what it carries: the main output is held to the effect contract, the emitted media URN against CapUrn::is_conformant_runtime_output (see Input and output and Cap URN structure: the effect contract at runtime), a violation being ExecutionError::EffectContractViolation; another output must carry a value of its own media URN, or exactly the cap’s input when it carries = "input", a violation being OutputContractViolation;
3. the stream contract of that output (OutputContract, its CapOutput.is_sequence + CapOutput.streaming): the frame’s is_sequence must equal the declared cardinality, and an unbounded emission is only allowed from an output declared streaming; a violation is ExecutionError::StreamContractViolation, naming both the declaration and the emission.

When the request ends successfully (ResponseAudit::audit_end), every required output must have had a stream; one that did not is OutputContractViolation. An optional output without a stream did not fire.

All are hard failures attributed internal: the cartridge broke its own declaration. Because is_sequence is audited, the plan’s cardinality reasoning can rely on the declaration.

What each receipt point takes from a response is the stream of the output it is there for. The stream of an output nothing takes is audited like any other and then consumed: its chunks are counted and its producer credited, so a cap never stalls on an output no step reads, and they are not delivered anywhere.

Forwarded STREAM_START frames are then relabeled to the URN the downstream step’s input is delivered under (StepInput.stream_urn) before being sent, exactly as send_step_input labels a chain head’s streams. Cartridges demux input streams by arg-URN equivalence (see 13.2), so the producer’s richer output URN must not reach the consumer as the stream label. The audit runs FIRST: the relabel substitutes the plan’s declared arg URN for the producer’s claim, which without the audit would mask a lying cap. The plan already proved the producer’s output conforms to the edge’s input; forward_frames re-asserts that conformance and fails hard on a violation (a wiring/type bug must surface at the hop, not as a missing-arg error inside the cartridge).

Regime note: a pipelined intermediate node’s data is never materialized into the executor’s result map, which is the point. Consumers needing an intermediate value must reference it as a fan-out (which breaks the segment there and materializes it). Chain sinks that DO materialize (split boundaries) are additionally captured to disk when the host provides a transient root (see §Transient run artifacts (host capture)).

For incremental consumption of a terminal cap’s output (unbounded streams), use TerminalOutput { next_item(), finish() }, the streaming counterpart of collect_terminal_output, sharing its frame semantics (END-carried final progress, credit granting, callback delivery).

Mid-run argument state (RunArgumentLedger)

execute_plan does not take an immutable argument map: it reads per-step arguments through a RunArgumentLedger, the plan’s mutable argument state and its dispatch journal under one mutex. That mutex is the ordering rule for the update/dispatch race: the dispatch of a chain’s steps, of the steps of one run of a map, and an argument update serialize on it, and each update’s per-step outcome is decided in the same critical section that applies it.

• dispatch_steps journals a chain’s steps as dispatched-and-exhausted and returns the values their invocations read, in one atomic step, so an update lands entirely before or entirely after the chain’s steps, never between them. A step attempted again under retry reads what its first attempt read.
• dispatch_steps_in_map journals the dispatch of steps in one run of a map (that run and every run before it have taken their values) and returns the values that run reads.
• exhaust_steps marks a map’s steps closed when its item sources end; no further dispatch exists for an update to reach.
• apply validates a whole batch (every token must name a cap node of THIS plan; step tokens are minted per plan, so a foreign token refuses the batch; and an update addressed to an arg the machine binds, to a node or to a constant, refuses the batch naming the step and the arg: that value is the machine’s), then reports per update: applied (no dispatch yet), applied-to-remaining-bodies(n) (n runs of the map already took the old value), or already-dispatched (value left untouched, because falsifying the record of work that already ran is exactly what the journal exists to prevent).

Keys are the plan’s cap node ids (the strand steps’ minted tokens). A host exposes the ledger for the lifetime of one execute_plan call; the reference CLI constructs one per run from its parsed arguments.

This layer exists only in the reference implementation. The other implementations do not include the plan executor (execute_plan, maps, per-step argument delivery) and so have no argument ledger.

Live-feed machines and unbounded terminals

A machine whose source is a live-feed reference (13.2 §Reference media) is not a special execution mode. It is the natural state of a machine, of which the bounded file run is the degenerate case whose input happens to end immediately. The execution core for live-feed machines lives in the CapDAG library and is exercised by the capdag CLI directly (capdag run on a machine with a live source, including the built-in media:live;synthetic feed); floom-engine consumes the same library surface and adds only platform-specific host mediation (sandboxed capture, TCC attribution). The CLI is the reference execution surface.

Two consequences for the executor:

• Head feeding is unchanged where the runtime resolves the reference: the engine passes the selector record as the head cap’s main-input arg value (a small reference value, exactly like a file path), and the cartridge runtime’s transport-resolution layer turns it into the unbounded input stream in-process. Host-mediated capture (the same capdag::capture dispatch, opened by the host) covers the two cases where the cartridge cannot be the resolver: a sandboxed cartridge that cannot open the device, and a live source feeding a map directly: per-item dispatch is engine-side, so the ENGINE opens the feed and dispatches one run of the body per delivered item while the capture runs (in-flight bodies bounded, item snapshots published append-only, the tap registered so a stop closes it and the bodies drain).
• Unbounded intermediates at mandatory split boundaries spool to disk. A chain split the executor cannot avoid (same-admission-domain bounded caps, fan-out) materialises its sink; when that sink’s stream is unbounded it streams to a temp-file spool instead of memory, and the downstream chain (or map driver) streams items back out of the spool. Memory stays bounded; the data is complete because the split semantics already required the feed to end first.
• Unbounded terminals must be consumed incrementally. A terminal whose STREAM_START declares unbounded MUST be routed to the incremental consumer (TerminalOutput) or an IncrementalWriter (persisted sinks); the buffering collect_terminal_output refuses unbounded streams by contract (12.4 §Unbounded). Recording-to-disk is the persisted-sink path applied to a live run; a properly stopped recording is a complete file because of the drain semantics below.

Streaming contracts

Boundedness is a definition-level property of a cap, declared alongside cardinality and orthogonal to it, and never a URN axis (the URN is the dispatch identity; a feed of frames and a file of frames are the same type) or inferred from is_sequence (a sequence can be bounded; a scalar blob can be unbounded, such as a live WAV):

• CapArg.streaming: the argument is consumed without a length promise: the cap processes items as they arrive and may be wired live to an unbounded producer. Only the main input may declare it (10 RULE14): side arguments are values the runtime demuxes whole.
• CapOutput.streaming: the output may be emitted without a length promise (an unbounded STREAM_START, 12.4).

Cap::streaming_shape() returns (input_streams, output_streams), the streaming counterpart of sequence_shape(), and both travel on the fabric definition (streaming in the cap TOML and in fabric.js’s published JSON, absent = false).

The hop rule. Within a linear chain, a hop from a producer whose output streaming is true into a consumer whose main input streaming is false is a chain boundary (split_chains_at_streaming_boundary): the producer runs to completion, its output is materialised (spooled to disk when it turns out unbounded; see §Live-feed machines), and the consumer is fed the bounded whole after the producer ENDs. Every other hop pipelines. The decision is made from the declarations, so the same machine behaves the same way whatever cartridges happen to serve it. Pool topology plays no part in the streaming-boundary decision.

Why RULE14. A feed is a property of the one stream a cap is wired to live. An option declared streaming would let an unbounded stream reach a collector that must refuse it (L16) with nothing in the plan having said so; keeping streaming on the main input alone keeps the hop rule one-dimensional and the audit exact.

Transient run artifacts (host capture)

Intermediate chain sinks (the transcription between two model caps, the frames between a bridge and its ForEach) are neither outputs (no listing; not things meant to keep) nor messages (too big for memory). When the runtime provides a transient root (EngineRuntime::transient_artifact_root, default None; the CLI and reference runtimes keep temp+delete), the executor captures every intermediate chain sink at the moment it materializes, mid-run:

• A memory-materialized (bounded) sink is written in its canonical byte form; a spool-engaged (unbounded) sink’s spool file is adopted: moved, not copied, and downstream consumers read the adopted path.
• Each capture is one directory: {root}/{node_id}/data plus a provenance.json sidecar carrying media urn, cardinality, byte count, and, for sequences, every item’s [offset, len], so a reader previews item N with one bounded read. Publication is atomic: captures stage into {node_id}.tmp and rename into place, so a published directory always holds its sidecar; a .tmp directory is an in-flight (or crashed) capture, invisible to readers, owned by the host’s TTL reaper.
• EngineRuntime::on_transient_artifact fires per capture and returns Result; a publication failure is a hard execution error, because an inspection surface that silently diverges from disk is an illegal state.
• ForEach body sinks are never captured (body guts are excluded by design; a map’s items and outputs have their own surfaces).
• There are NO database rows: the filesystem is the index, an absent directory is the legal expired state, and the host’s eager GC (run deletion, startup sweep, TTL reaper) owns every lifetime.

One node, one truth. A chain sink’s data is recorded in the result map’s memory form OR as a spool path, never both. A spool-engaged sink does not appear in node_data (its in-memory remainder is not the stream), and the map driver refuses an input present in both as an executor contract breach rather than guessing. With transients on, adopted spool paths are excluded from plan-end spool cleanup; the reaper owns them.

Runs stop

A machine is running or it is not. “Cancel” belongs to the task layer and the wire frame that implements abort (a Cancel frame cancels a request); at the run level the vocabulary and the semantics are stop:

• A run stops on its own when its input stream ends: a file right after its content; a live feed when a stop condition (duration_ms, max_items) fires or the device closes. Same END, same path; bounded input is the case where this happens immediately.
• A run is stopped by closing the tap: the runtime closes the request’s live feeds, the feed’s input stream ends, and the machine drains: every in-flight item flows through, terminals end normally, writers finalize. The outputs of a stopped run are complete (a stopped recording is a valid file). On the wire this is a CloseStream frame (12.2 §CloseStream) to the feed-bearing request: the runtime closes feeds and lets the request end naturally instead of interrupting the handler. It is not a Cancel of any kind (a Cancel always cancels), and a CloseStream for a request holding no open feed is logged and ignored: stop is stop because the frame says so, never because of what the runtime happens to find.
• A run is aborted by the cancel cascade, a Cancel attributed to the user, the hard variant for a run that must die now, in-flight work discarded (the force_kill flag kills processes instead of cancelling cooperatively). Abort is the escalation, never the default stop. The host’s own aborts (stale reaper, memory pressure) are Cancels attributed to the host’s reason (ABORTED, class environment / resource); the executor’s teardown of a chain around a failed step is a Cancel attributed to the originating failure (ABORTED_COLLATERAL, its class). Every terminal carries that attribution, so “cancelled” on a surface always means a human cancelled.

Control surfaces. Stop and abort are DIFFERENT controls everywhere: the engine exposes TaskService.StopTaskInput (close the running task’s live taps: host-opened feeds directly, cartridge-resolved feeds via a CloseStream frame to their FEED-BEARING requests, which the shared executor records through the FlowObserver seam; then the run drains and completes WITH its outputs; the task’s status is untouched) alongside CancelTask (abort). On the wire the stop is RelaySwitch::stop_request_feeds, the CloseStream FRAME sent to the feed-bearing request WITHOUT touching host-side request state (terminating host state is the abort’s job; doing it on a stop severs routing mid-drain and wedges the run).

Stop is a verified request. A stop returns a verdict, never a hard-coded success: stop_feed_requests (bifaci::live_feed) closes the host taps, sends every CloseStream, and then awaits the evidence: each stopped request reaching its terminal at the switch, bounded by STOP_INPUT_DRAIN_TIMEOUT, and answers StopInputsOutcome { host_taps_closed, feed_requests_stopped, feeds_ended } where success means the input has ended. Nothing open to stop is NothingToStop (the engine maps it to FAILED_PRECONDITION); a feed that does not end within the timeout is FeedDidNotEnd naming the request (DEADLINE_EXCEEDED); a feed that ends with ERR is FeedFailed. TaskService.StopTaskInput returns StopTaskInputResponse with the same three counts, and both desktop hosts render the engine’s verdict or its refusal.

The loom run header renders a LIVE run’s controls first-class: “Stop input” (tap-off) and “Abort”, with no hidden modifier; a bounded run keeps one Stop (cooperative cancel, ⌥/Alt force-kills). In the capdag CLI the first Ctrl-C is the stop (taps close, the machine drains, outputs are emitted) and prints the verdict (“Input stopped: N host tap(s) closed, M feed request(s) stopped, M ended”) or the refusal; a second Ctrl-C, at any point, aborts with a Cancel attributed to the user. Stopping the input is also how a capacity-bounded cartridge frees up mid-machine: its live consumer finishes on the drained input and the permit passes to the next cap.

There is no plan-time restriction on unbounded inputs: is_sequence on the consuming arg decides the wiring (a sequence-consuming fold links directly and defers emission until END, which stopping delivers; an item-consuming cap gets ForEach), and streaming decides the hop (§Streaming contracts: a bounded-only consumer downstream of a streaming producer is fed the whole after the producer ends). Emission timing is op policy, not type structure. A fold over a live feed is a valid deferred-emission plan: concat over a microphone emits the whole recording when the run stops.