feat(stream): implement declared event ingress

Author: schickling

INVARIANTS.md

@@ -11,9 +11,10 @@ materialization, messaging, DING, or presence must preserve them.
 | **Transport-decoupled lifecycle** | Each task is isolated from a supervisor/transport process-group or cgroup cascade. | `tests/transport_isolation.rs`; `tests/transport_isolation_macos.rs` |
 | **Clean exec teardown** | Killing an exec task reaps its whole process group. | `tests/exec_backend.rs::exec_kill_reaps_the_whole_process_group_not_just_the_leader` |
 | **Bounded restart diagnostics** | Relaunching an exec task preserves the just-finished log as one prior generation while bounding retained diagnostics to current plus prior. Final retirement removes the PID and both logs. | `tests/exec_backend.rs::exec_restart_reap_keeps_bounded_diagnostics_and_final_remove_cleans_them`; `tests/run.rs::up_once_finally_removes_dead_retired_tasks_without_restarting_them` |
-| **Derived companion lifecycle** | A generated DING starts only with an eligible canonical agent, is suppressed or stopped while that target is held, suspended, unavailable, retired, or terminally parked, and remains coupled without changing explicit sibling task behavior. Healthy compact startup still launches both tasks in one pass, and targeted reconciliation holds a missing generated DING rather than broadening to its agent. | `tests/run.rs::fresh_compact_agent_launches_with_its_derived_ding`; `tests/run.rs::absent_adopt_only_compact_agent_holds_its_derived_ding`; `tests/run.rs::held_adopt_only_compact_agent_stops_its_live_derived_ding`; `tests/run.rs::failed_compact_agent_restart_stops_its_live_derived_ding`; `tests/run.rs::failed_compact_agent_reap_stops_its_live_derived_ding`; `tests/run.rs::parked_compact_agent_stops_its_live_derived_ding`; `tests/run.rs::parked_compact_agent_does_not_relaunch_its_exited_derived_ding`; `tests/run.rs::retired_compact_agent_stops_agent_and_derived_ding`; `tests/run.rs::suspend_and_resume_cover_derived_ding_sibling_continuity_and_inbox_retention`; `tests/run.rs::selected_missing_derived_ding_is_held_without_broadening_to_its_agent`; `tests/run.rs::up_once_collects_spawn_errors_without_aborting` |
+| **Derived companion lifecycle** | A generated DING or launched stream starts only with an eligible canonical agent, is suppressed or stopped while that target is held, suspended, unavailable, retired, or terminally parked, and remains coupled without changing explicit sibling task behavior. A stream lowers its authored adapter launch directly, parks and surfaces independently, and neither makes an otherwise empty agent runnable nor claims a delivery transport. Stream authoring is serialized, authority-scoped, source-preserving, and fail-closed for Nix ownership and invalid declarations. Healthy compact startup launches companions in one pass, and targeted reconciliation holds a missing generated companion rather than broadening to its agent. | `crates/agent-spec/tests/discovery.rs::streams_are_typed_and_only_launched_streams_lower_to_derived_exec_tasks`; `crates/agent-spec/tests/discovery.rs::stream_names_launches_and_task_collisions_fail_closed`; `src/agent_author.rs::stream_add_supports_external_command_and_argv_and_remove_is_idempotent`; `src/agent_author.rs::stream_authoring_enforces_authority_nix_ownership_and_canonical_validation`; `tests/run.rs::fresh_compact_agent_launches_with_its_derived_ding`; `tests/run.rs::held_adopt_only_compact_agent_stops_its_live_derived_ding`; `tests/run.rs::parked_compact_agent_stops_its_live_derived_ding`; `tests/run.rs::retired_compact_agent_stops_agent_and_derived_ding`; `tests/run.rs::selected_missing_derived_ding_is_held_without_broadening_to_its_agent`; `tests/run.rs::fresh_compact_agent_launches_with_its_derived_stream`; `tests/run.rs::retired_compact_agent_stops_agent_and_derived_stream`; `tests/run.rs::suspended_compact_agent_stops_its_derived_stream_without_touching_a_sibling`; `tests/run.rs::held_adopt_only_compact_agent_stops_its_live_derived_stream`; `tests/run.rs::a_crash_looping_stream_parks_and_surfaces_without_disturbing_its_agent`; `tests/run.rs::parked_compact_agent_stops_its_live_derived_stream`; `tests/run.rs::selected_missing_derived_stream_is_held_without_broadening_to_its_agent`; `tests/run.rs::a_stream_alone_does_not_make_an_agent_runnable`; `tests/run.rs::a_stream_does_not_claim_a_delivery_transport` |
 | **Exactly-once-safe native bus** | Messages use stable `<unix-ms>-<rand6>.md` files. An archive filename is a durable receipt that shadows and cleans restored inbox replicas and makes repeated archive cleanup idempotent. | `src/message.rs::filename_grammar`; `src/message.rs::archive_receipt_suppresses_and_idempotently_cleans_a_restored_inbox_copy`; `tests/message.rs` |
 | **Idempotent service requests** | A declared non-agent service principal publishes one exact JSON request per caller-supplied idempotency key to a canonical Agent Spec inbox. Concurrent or crash-replayed publication reuses the reserved filename; conflicting key reuse fails. The typed reply routes to the principal's canonical inbox without an Agent Spec identity or orphan mailbox. | `tests/request_cli.rs::stable_request_key_atomically_deduplicates_one_canonical_agent_message`; `tests/request_cli.rs::concurrent_replays_publish_exactly_one_request`; `tests/request_cli.rs::typed_reply_routes_to_the_principal_and_status_is_a_tagged_json_union`; `tests/request_cli.rs::request_api_rejects_agent_impersonation_and_unknown_flat_principals` |
+| **Bounded idempotent stream ingress** | A running agent accepts events only for a declared stream. Within the retained 128-receipt ring, concurrent or crash-replayed `(stream, event-id)` publication reuses one canonical filename, conflicting content fails, and supersession archives only the matching keyed predecessor or the stream-wide head through ordinary archive semantics. State remains bounded and honestly treats an identity evicted from the ring as new without searching inbox or archive history. Events do not write the Sent ledger and DING marks them as stream work. | `tests/event_e2e.rs::stable_event_identity_publishes_exactly_one_canonical_message`; `tests/event_e2e.rs::concurrent_replays_publish_exactly_one_event`; `tests/event_e2e.rs::conflicting_reuse_and_undeclared_or_suspended_ingress_fail_closed`; `tests/event_e2e.rs::supersede_collapses_only_the_matching_key_and_preserves_archive_receipts`; `tests/event_e2e.rs::keyless_supersede_replaces_the_stream_wide_head`; `tests/event_e2e.rs::crash_replay_honors_an_archive_receipt_and_never_restores_the_inbox_copy`; `tests/event_e2e.rs::subject_frontmatter_injection_is_refused_before_any_write`; `tests/event_e2e.rs::stream_state_is_bounded_and_forgets_only_beyond_its_honest_horizon`; `tests/event_e2e.rs::event_emit_cli_returns_a_stable_json_receipt_and_ding_marks_the_record`; `tests/stream_authoring_cli.rs::a_direct_adapter_launch_executes_the_exact_event_cli_contract`; `tests/run.rs::suspend_and_resume_relaunch_the_agent_and_stream_together` |
 | **Fail-closed observed native DING** | Each unread message becomes one normalized `[DING]` frame. Fresh delivery records ownership, then preserves the one combined bracketed-paste, 0.5 second delay, and Return transaction. PTY and Return success are transport only: `Delivered` additionally requires adapter classification of the expected notice text in a submitted-prompt or queued-message pattern while the lowest live composer is empty or an accepted idle placeholder. Retry never re-pastes and may send one bare Return only after two adjacent `RetainedSafe` observations. A maintained adapter's positive `NotRetained` observation releases only an already archived staged head; unread, blocked, timed-out, errored, unknown, and unrecognized states retain ownership and later FIFO work remains blocked. Ownership prevents duplicate paste across command failures, receipt ambiguity, archive races, and restart adoption without letting a vanished archived head block FIFO indefinitely. Startup backlog otherwise becomes one generic recovery DING; new arrivals remain FIFO; `busy` delivers immediately; only fresh `dnd` defers. | `src/ding/mod.rs::poke_text_normalizes_and_bounds_untrusted_fields`; `src/ding/mod.rs::malicious_controls_cannot_escape_the_single_paste_frame`; `src/ding/mod.rs::pty_delivery_uses_face607_delay_order_and_seconds`; `src/ding/mod.rs::maintained_composer_classifiers_require_exact_idle_state`; `src/ding/mod.rs::successful_transport_with_retained_or_unproven_pixels_is_not_delivered`; `src/ding/mod.rs::ambiguous_transport_receipt_and_retry_errors_retain_staged_ownership`; `src/ding/mod.rs::adapter_recognized_notice_with_an_empty_live_composer_is_a_positive_receipt`; `src/ding/mod.rs::staged_retry_submits_only_retained_safe_and_requires_a_receipt`; `src/ding/mod.rs::staged_retry_keeps_unproven_and_retained_blocked_owned`; `src/ding/mod.rs::staged_ownership_survives_archive_and_never_repastes`; `src/ding/mod.rs::archived_not_retained_releases_fifo_without_repasting_owned_notice`; `src/ding/mod.rs::unread_not_retained_keeps_fifo_ownership_without_repasting`; `src/ding/mod.rs::pty_commands_have_a_real_outer_timeout`; `src/ding/mod.rs::session_watch_has_startup_grace_debounce_and_live_reset`; `src/ding/mod.rs::new_arrivals_is_fifo_and_archive_receipts_prevent_reding`; `src/ding/mod.rs::pending_delivery_ignores_busy_but_respects_fresh_dnd_archive_and_retry`; `src/ding/mod.rs::startup_recovery_notice_retries_in_memory`; `src/ding/mod.rs::startup_backlog_gets_one_generic_recovery_then_new_arrivals_poke` |
 | **Mutation-only filesystem wakeups** | Supervisor and DING filesystem watchers ignore read/open access events and wake early only for create, modify, rename, or remove events. Their own catalog and inbox reads therefore cannot bypass the bounded timer cadence or form a Linux inotify CPU loop. | `src/watch.rs::only_mutations_wake_watch_loops`; `src/watch.rs::linux_reads_are_silent_but_real_mutations_wake`; `src/ding/mod.rs::idle_ding_does_not_spin_on_its_own_inbox_reads`; `src/run.rs::idle_supervisor_does_not_spin_on_its_own_catalog_reads` |
 | **Bounded DING PTY probe churn** | An unsafe or active composer retains its FIFO notice but deferred delivery retries use a bounded backoff, so each inbox poll cannot spawn another short-lived PTY probe. | `src/ding/mod.rs::deferred_delivery_backoff_bounds_short_lived_pty_attempts` |

crates/agent-spec/src/declared.rs

@@ -39,6 +39,7 @@ pub enum DeclaredDiagnosticCode {
     TaskNameMissing,
     UnsupportedSchedule,
     DuplicateRoutingField,
+    UnsupportedStreamInterval,
 }
 
 impl DeclaredDiagnosticCode {
@@ -49,6 +50,7 @@ impl DeclaredDiagnosticCode {
             Self::TaskNameMissing => "task-name-missing",
             Self::UnsupportedSchedule => "unsupported-schedule",
             Self::DuplicateRoutingField => "duplicate-routing-field",
+            Self::UnsupportedStreamInterval => "unsupported-stream-interval",
         }
     }
 }
@@ -301,14 +303,31 @@ pub fn parse_declared_document(source_name: &Path, source: &str) -> DeclaredPars
         }
         for child in &node.children {
             match child.name.as_str() {
-                "pty" | "exec" if child.argument(0).and_then(DeclaredValue::as_str).is_none() => {
+                "pty" | "exec" | "stream"
+                    if child.argument(0).and_then(DeclaredValue::as_str).is_none() =>
+                {
                     diagnostics.push(shape_diagnostic(
                         source_name,
                         child.span,
                         DeclaredDiagnosticCode::TaskNameMissing,
                         format!("{} task must have one positional string name", child.name),
                     ));
                 }
+                // A command-bearing stream lowers to a derived exec companion. A command-less
+                // stream is an external ingress endpoint. `every` would make either one a schedule,
+                // which is the reserved `schedule` node's business.
+                "stream" => {
+                    for field in child.children_named("every") {
+                        diagnostics.push(shape_diagnostic(
+                            source_name,
+                            field.span,
+                            DeclaredDiagnosticCode::UnsupportedStreamInterval,
+                            "stream `every` is reserved for the future `schedule` contract; a stream \
+                             declares a long-running event source"
+                                .to_owned(),
+                        ));
+                    }
+                }
                 "schedule" => diagnostics.push(shape_diagnostic(
                     source_name,
                     child.span,

crates/agent-spec/src/discovery.rs

@@ -80,10 +80,7 @@ pub fn discover_strict(root: &Path) -> Discovered {
 ///
 /// `root` supplies the same path defaults as [`discover`]. The returned warnings describe only
 /// this file.
-pub fn discover_file(
-    root: &Path,
-    path: &Path,
-) -> anyhow::Result<(Vec<AgentSpec>, Vec<String>)> {
+pub fn discover_file(root: &Path, path: &Path) -> anyhow::Result<(Vec<AgentSpec>, Vec<String>)> {
     let raws = parse_raw_file(path)?;
     load_specs(root, path, raws)
 }

crates/agent-spec/src/kdl_format.rs

@@ -77,7 +77,10 @@ fn agent_node_to_raw(node: &DeclaredNode) -> anyhow::Result<RawSpec> {
             "workspace" => raw.workspace = arg_string(child),
             "supervisor" => raw.supervisor = arg_string(child),
             "retired" => {
-                anyhow::ensure!(raw.retired.is_none(), "agent declares `retired` more than once");
+                anyhow::ensure!(
+                    raw.retired.is_none(),
+                    "agent declares `retired` more than once"
+                );
                 raw.retired = Some(Some(arg_bool(child)));
             }
             "desired-state" => {
@@ -136,10 +139,9 @@ fn agent_node_to_raw(node: &DeclaredNode) -> anyhow::Result<RawSpec> {
                         && child.entries[0].name.is_none(),
                     "agent `deliver` must contain exactly one positional string"
                 );
-                raw.deliver = Some(Some(
-                    arg_string(child)
-                        .ok_or_else(|| anyhow::anyhow!("agent `deliver` value must be a string"))?,
-                ));
+                raw.deliver = Some(Some(arg_string(child).ok_or_else(|| {
+                    anyhow::anyhow!("agent `deliver` value must be a string")
+                })?));
             }
             "claude" => {
                 anyhow::ensure!(
@@ -166,6 +168,15 @@ fn agent_node_to_raw(node: &DeclaredNode) -> anyhow::Result<RawSpec> {
                     raw.exec.insert(name, task_node_to_raw(child)?);
                 }
             }
+            "stream" => {
+                if let Some(name) = arg_string(child) {
+                    let stream = stream_node_to_raw(child, &name)?;
+                    anyhow::ensure!(
+                        raw.stream.insert(name.clone(), stream).is_none(),
+                        "agent declares `stream \"{name}\"` more than once"
+                    );
+                }
+            }
             // meta, harness, model, persona, permissions, transport, strategy, … — ignored.
             _ => {}
         }
@@ -286,8 +297,7 @@ fn common_driver_fields(
 }
 
 fn claude_driver_node_to_raw(node: &DeclaredNode) -> anyhow::Result<ClaudeDriver> {
-    let (model, effort, dev_channels, prompt, args) =
-        common_driver_fields(node, "claude", true)?;
+    let (model, effort, dev_channels, prompt, args) = common_driver_fields(node, "claude", true)?;
     Ok(ClaudeDriver {
         model,
         effort,
@@ -434,6 +444,47 @@ fn task_node_to_raw(node: &DeclaredNode) -> anyhow::Result<RawTask> {
     Ok(t)
 }
 
+/// `stream "<name>" { command "…" }` or `stream "<name>" { argv "prog" "arg" }`.
+///
+/// The child set is deliberately minimal. A stream declares WHERE events come from; everything about
+/// how they are supervised is inherited from the agent (restart policy, teardown, parking), and
+/// everything about how they are delivered is the bus contract. `every` is rejected by the
+/// declaration parser rather than accepted here: an interval makes this scheduled work, which is
+/// the reserved `schedule` node's contract.
+fn stream_node_to_raw(node: &DeclaredNode, name: &str) -> anyhow::Result<crate::spec::RawStream> {
+    let mut stream = crate::spec::RawStream::default();
+    for child in &node.children {
+        match child.name.as_str() {
+            "command" => {
+                anyhow::ensure!(
+                    stream.command.is_none(),
+                    "stream '{name}' has duplicate `command`"
+                );
+                stream.command = Some(arg_string(child).ok_or_else(|| {
+                    anyhow::anyhow!("stream '{name}' `command` must be one positional string")
+                })?);
+            }
+            "argv" => {
+                anyhow::ensure!(
+                    stream.argv.is_none(),
+                    "stream '{name}' has duplicate `argv`"
+                );
+                stream.argv = Some(argv(child)?);
+            }
+            "every" => anyhow::bail!(
+                "stream '{name}' declares `every`; scheduled work is the reserved `schedule` \
+                 contract, a stream is a long-running event source"
+            ),
+            other => anyhow::bail!("stream '{name}' has unsupported field `{other}`"),
+        }
+    }
+    anyhow::ensure!(
+        !(stream.command.is_some() && stream.argv.is_some()),
+        "stream '{name}' must declare at most one of `command` or `argv`"
+    );
+    Ok(stream)
+}
+
 fn env_node_to_raw(node: &DeclaredNode) -> std::collections::BTreeMap<String, String> {
     let mut env = std::collections::BTreeMap::new();
     for child in &node.children {

crates/agent-spec/src/lib.rs

@@ -44,6 +44,6 @@ pub use discovery::{
 };
 pub use spec::{
     AgentDesiredState, AgentSpec, ClaudeDriver, CodexDriver, DeliveryTransport, Driver, JobType,
-    Resource, Restart, RestartMode, Task, TaskKind, TaskLifecycle, parse_duration,
-    validate_desired_state_reason,
+    Resource, Restart, RestartMode, STREAM_TASK_PREFIX, Stream, StreamLaunch, Task, TaskKind,
+    TaskLifecycle, parse_duration, stream_name_of_task, validate_desired_state_reason,
 };