feat(scheduled): signed scheduled work chain — receiver, executor, sink factory, catalogue adapter (folds #28-#31)

async_scheduled_sink.py

@@ -0,0 +1,95 @@
+"""Bounded background delivery, with receipt only after signed publication.
+
+An operator supplies one immutable scope and trusted callbacks. Incoming data
+may select only its UTC slot. Threads remain in the owning service's cgroup;
+this adapter cannot keep a child alive after that service is stopped.
+"""
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from contextvars import copy_context
+from dataclasses import asdict
+import json
+import threading
+
+from scheduled_receiver import ExecutionScope, receive, run_id
+from slot_receiver import SlotLedger
+
+
+class DeliveryPending(ConnectionError):
+    """No terminal receipt yet; leave the durable emitter row undelivered."""
+
+
+class AsyncScheduledSink:
+    """A sink for complete admitted bindings, not generic emitter envelopes.
+
+    deliver() never waits for a child: it raises DeliveryPending until receive()
+    has persisted and published the outcome. A failed worker raises its actual
+    error on the next delivery; another delivery then re-enters the durable
+    receiver, which permits publication retries but never child replay.
+
+    Capacity counts queued, running AND completed-but-unacknowledged futures.
+    A thread pool's unbounded queue therefore never becomes an unbounded job
+    queue. The owner must call close() during orderly shutdown. Abrupt death
+    leaves claimed slots UNKNOWN; reconcile them, never erase the ledger.
+    """
+
+    def __init__(self, ledger: SlotLedger, scope: ExecutionScope, port_factory,
+                 execute, *, workers: int = 1, capacity: int = 4):
+        scope.validate()
+        if (type(workers) is not int or type(capacity) is not int or
+                not 1 <= workers <= capacity <= 64):
+            raise ValueError("require 1 <= workers <= capacity <= 64")
+        if not callable(port_factory) or not callable(execute):
+            raise ValueError("trusted port factory and executor required")
+        self.ledger, self.scope = ledger, scope
+        self.port_factory, self.execute = port_factory, execute
+        self.capacity = capacity
+        self._lock = threading.Lock()
+        self._pending = {}
+        self._closed = False
+        self._pool = ThreadPoolExecutor(max_workers=workers,
+                                        thread_name_prefix="scheduled-receiver")
+
+    def delivery_scope(self):
+        return {"receiver": "async-scheduled-v1", **asdict(self.scope)}
+
+    def deliver(self, delivery: dict) -> None:
+        if type(delivery) is not dict:
+            raise ValueError("complete admitted binding required")
+        slot = delivery.get("scheduled_at")
+        identity = run_id(self.scope, slot)
+        binding = {**asdict(self.scope), "scheduled_at": slot}
+        if (json.dumps(delivery, sort_keys=True, allow_nan=False) !=
+                json.dumps(binding, sort_keys=True, allow_nan=False)):
+            raise ValueError("delivery differs from pinned scope")
+        with self._lock:
+            if self._closed:
+                raise ConnectionError("scheduled receiver is closed")
+            future = self._pending.get(identity)
+            if future is None:
+                if len(self._pending) >= self.capacity:
+                    raise DeliveryPending("scheduled receiver capacity reached")
+                context = copy_context()
+                future = self._pool.submit(context.run, self._receive, slot, binding)
+                self._pending[identity] = future
+                # Submission is not a durable receipt, even if a fast worker
+                # happens to finish before this call exits.
+                raise DeliveryPending("scheduled delivery submitted; receipt pending")
+            if not future.done():
+                raise DeliveryPending("scheduled delivery still pending")
+            try:
+                future.result()  # terminal publication verified, or real error
+            finally:
+                del self._pending[identity]
+
+    def _receive(self, slot, binding):
+        port = self.port_factory(slot)
+        return receive(self.ledger, self.scope, slot, binding, port,
+                       lambda: self.execute(slot))
+
+    def close(self):
+        """Stop admission and drain workers; call outside the reactor tick."""
+        with self._lock:
+            self._closed = True
+        self._pool.shutdown(wait=True)

chaski.py

@@ -342,7 +342,13 @@ def load_emitters(path: Path) -> list[tuple[emitter.Emitter, dict]]:
         if missing:
             raise RuleError(f"emitter {d.get('label', '?')!r} is missing {missing}")
         kinds = set(d["sink"])
-        if kinds != {"jsonl"} and not ("graph" in kinds and kinds <= {"graph", "command", "alertmanager"}):
+        if kinds == {"scheduled"}:
+            from scheduled_sink_factory import validate
+            try:
+                validate(d["sink"]["scheduled"])
+            except (ValueError, TypeError, OSError) as exc:
+                raise RuleError(f"emitter {d['label']}: invalid scheduled sink: {exc}") from None
+        elif kinds != {"jsonl"} and not ("graph" in kinds and kinds <= {"graph", "command", "alertmanager"}):
             raise RuleError(f"emitter {d['label']}: sink must be {{jsonl: <path>}} or {{graph: {{...}}}},"
                             " optionally with command: {...} and/or alertmanager: {...} beside graph")
         if "command" in kinds and not (d["sink"]["command"] or {}).get("argv"):
@@ -394,6 +400,12 @@ def load_emitters(path: Path) -> list[tuple[emitter.Emitter, dict]]:
             raise RuleError("emitter trigger must be schedule, changes or external")
         if e.externally_triggered and d.get("baseline_emits") is not True:
             raise RuleError("external emitters require baseline_emits: true to retain their first event")
+        if kinds == {"scheduled"}:
+            from scheduled_sink_factory import validate_emitter
+            try:
+                validate_emitter(e)
+            except ValueError as exc:
+                raise RuleError(f"emitter {d['label']}: {exc}") from None
         out.append((e, d["sink"]))
     return out
 
@@ -404,6 +416,9 @@ def make_sink(e: emitter.Emitter, sink: dict, quipu_url: str):
     never from the rules or the command line."""
     import emitter as em
 
+    if set(sink) == {"scheduled"}:
+        from scheduled_sink_factory import build
+        return build(e, sink["scheduled"])
     if "jsonl" in sink:
         return em.JsonlSink(sink["jsonl"])
     import graph_sink
@@ -481,18 +496,49 @@ def run_consumer(args) -> int:
     """Enter only after both persistent cursor and inbox custody are held."""
     rules = load_rules(args.rules)
     reactor = Reactor(Quipu(args.quipu), rules, args.state, args.event_poll)
-    if args.emitters:
-        import emitter
-
-        conn = emitter.connect(args.emitter_db or args.state.with_name("emitter.db"))
-        budget = emitter.WriteBudget(args.emitter_write_interval)  # ONE, shared by all
-        from incremental import ChangeFeed, ChangeRunner
-
-        reactor.runners = [(ChangeRunner if e.change_driven else emitter.Runner)(
-            e, conn, make_sink(e, sink, args.quipu), budget) for e, sink in load_emitters(args.emitters)]
-        changing = [r for r in reactor.runners if r.emitter.change_driven]
-        if changing:
-            reactor.change_feed = ChangeFeed(reactor.quipu, changing)
+    conn, sinks = None, []
+    try:
+        if args.emitters:
+            import emitter
+
+            conn = emitter.connect(args.emitter_db or args.state.with_name("emitter.db"))
+            budget = emitter.WriteBudget(args.emitter_write_interval)  # ONE, shared by all
+            from incremental import ChangeFeed, ChangeRunner
+
+            for e, definition in load_emitters(args.emitters):
+                sink = make_sink(e, definition, args.quipu)
+                sinks.append(sink)  # retain custody even if runner construction fails
+                reactor.runners.append((ChangeRunner if e.change_driven else emitter.Runner)(
+                    e, conn, sink, budget))
+            changing = [r for r in reactor.runners if r.emitter.change_driven]
+            if changing:
+                reactor.change_feed = ChangeFeed(reactor.quipu, changing)
+        return _serve_consumer(args, reactor, rules)
+    finally:
+        # Drain before releasing ConsumerLease; pending outbox rows remain
+        # unacknowledged until their next delivery verifies the result.
+        try:
+            _close_sinks(sinks)
+        finally:
+            if conn is not None:
+                conn.close()
+
+
+def _close_sinks(sinks):
+    error = None
+    for sink in reversed(sinks):
+        try:
+            close = getattr(sink, 'close', None)
+            if close is not None:
+                close()
+        except Exception as exc:
+            if error is None:
+                error = exc
+    if error is not None:
+        raise error
+
+
+def _serve_consumer(args, reactor, rules):
     LOG.info("chaski: %d rule(s), event poll %ds", len(rules), reactor.event_poll_s)
     if args.once:
         now = time.time()

docs/demos/async-scheduled.log

@@ -0,0 +1,13 @@
+Script started on 2026-10-15 06:37:02-04:00 [COMMAND="stty -onlcr && .venv/bin/python scripts/demo-async-scheduled.py" <not executed on terminal>]
+Source demo: real worker thread and SQLite; signing callbacks are fixtures.
+Existing receiver positive control: True
+Async handoff available: True
+First delivery: PENDING, no receipt
+Duplicate delivery: PENDING, no receipt
+Worker blocked while caller continued: True
+Execution callbacks: 1 ; publications: 0
+Later delivery: RECEIPT after terminal publication
+Published actual outcome: skipped-by-hold
+Final execution callbacks: 1
+
+Script done on 2026-10-15 06:37:02-04:00 [COMMAND_EXIT_CODE="0"]

docs/demos/async-scheduled.timing

@@ -0,0 +1,6 @@
+0.035160 117
+0.000015 30
+0.020770 36
+0.009772 125
+0.010488 93
+0.000004 29

docs/demos/fixed-outcome-executor.log

@@ -0,0 +1,10 @@
+Script started on 2026-10-15 08:37:01-04:00 [COMMAND="stty -onlcr && .venv/bin/python -I scripts/demo-fixed-outcome-executor.py" <not executed on terminal>]
+Receiver identity positive control: True
+Fixed outcome executor available: True
+Child: {'outcome': 'succeeded', 'exit_code': 0}
+Child: {'outcome': 'skipped-by-hold', 'exit_code': 0}
+Child: {'outcome': 'failed', 'exit_code': 7}
+Exit-zero child without result: UNKNOWN; preserve claim, no retry.
+No production schedule, signing key, graph or cron was used.
+
+Script done on 2026-10-15 08:37:01-04:00 [COMMAND_EXIT_CODE="0"]

docs/demos/fixed-outcome-executor.timing

@@ -0,0 +1,6 @@
+0.030625 80
+0.026840 48
+0.023159 54
+0.016917 45
+0.020396 67
+0.000159 61

docs/demos/scheduled-catalogue-adapter.log

@@ -0,0 +1,13 @@
+Script started on 2026-10-15 09:35:17-04:00 [COMMAND="stty -onlcr && .venv/bin/python -I scripts/demo-scheduled-catalogue-adapter.py" <not executed on terminal>]
+Projection positive control: True
+Incremental catalogue adapter available: True
+Describe reader calls: 0
+Bootstrap discover/evaluate reader calls: 2
+After idle ticks, reader calls: 2
+After next due slot, reader calls: 3
+Missing identity UNKNOWN count: 1
+Retained next deadline: 3000.0
+Retained queued slots: 2
+Reader is a fixture; no graph, production job, key, profile or cron used.
+
+Script done on 2026-10-15 09:35:18-04:00 [COMMAND_EXIT_CODE="0"]

docs/demos/scheduled-catalogue-adapter.timing

@@ -0,0 +1,9 @@
+0.036667 34
+0.000032 46
+0.207628 25
+0.142928 44
+0.003707 34
+0.064391 37
+0.077769 34
+0.000005 31
+0.000012 99

docs/demos/scheduled-event.log

@@ -0,0 +1,12 @@
+Script started on 2026-10-15 07:19:09-04:00 [COMMAND="stty -onlcr && .venv/bin/python scripts/demo-scheduled-event.py" <not executed on terminal>]
+Source demo: actual emitter/outbox, worker and ledger; signing is a fixture.
+Existing async receiver positive control: True
+Native scheduled envelope adapter available: True
+Observed due declaration; durable outbox rows: 1
+Two delivery attempts; child still blocked: True
+Execution callbacks: 1 ; published: 0 ; receipt: None
+Published actual outcome: skipped-by-hold
+Terminal receipt: 690.0 ; execution callbacks: 1
+Outbox pinned emitter and recipient: schedule operator
+
+Script done on 2026-10-15 07:19:09-04:00 [COMMAND_EXIT_CODE="0"]

docs/demos/scheduled-event.timing

@@ -0,0 +1,6 @@
+0.054368 124
+0.000013 50
+0.063978 49
+0.022699 103
+0.016719 91
+0.000049 55

docs/demos/scheduled-projection.log

@@ -0,0 +1,10 @@
+Script started on 2026-10-15 07:48:58-04:00 [COMMAND="stty -onlcr && .venv/bin/python scripts/demo-scheduled-projection.py" <not executed on terminal>]
+Receiver identity positive control: True
+Pinned schedule projection available: True
+Effective 1250; first eligible aligned slot: 1800
+Three observations; two current-slot events: 2
+Next deadline: 3000
+Missing required identity: REFUSED; retained outbox rows: 2
+Execution callbacks: 0; no sink or production schedule enabled.
+
+Script done on 2026-10-15 07:48:58-04:00 [COMMAND_EXIT_CODE="0"]

docs/demos/scheduled-projection.timing

@@ -0,0 +1,5 @@
+0.032912 41
+0.000014 43
+0.001719 50
+0.070044 67
+0.000009 124

docs/demos/scheduled-sink-factory.log

@@ -0,0 +1,11 @@
+Script started on 2026-10-15 09:18:55-04:00 [COMMAND="stty -onlcr && .venv/bin/python -I scripts/demo-scheduled-sink-factory.py" <not executed on terminal>]
+Existing sink constructor positive control: True
+Configured scheduled sink factory available: True
+Submit delivery, acknowledged/pending: (0, 1)
+After terminal publication, acknowledged/pending: (1, 0)
+Phase receipts: ['admit', 'started', 'finished', 'publish']
+Explicit child outcome: {'outcome': 'skipped-by-hold', 'exit_code': 0}
+Actual child invocations: 1
+No production graph, schedule, key or cron was used; signing is a fixture.
+
+Script done on 2026-10-15 09:18:55-04:00 [COMMAND_EXIT_CODE="0"]

docs/demos/scheduled-sink-factory.timing

@@ -0,0 +1,6 @@
+0.052777 49
+0.000014 50
+0.101955 46
+0.173618 57
+0.000054 131
+0.000013 103

docs/fixed-outcome-executor.md

@@ -0,0 +1,39 @@
+# Fixed command outcome adapter
+
+`FixedOutcomeExecutor` is a synchronous callback for the scheduled receiver. A
+trusted caller constructs it with fixed absolute argv, an existing directory,
+an explicit environment, a result descriptor variable and bounded runtime/output.
+Delivery metadata cannot choose those values. The slot is validated as identity
+metadata and is never interpolated into the command. The adapter does not inherit
+the caller's ambient environment or invoke a shell.
+
+The child receives a writable anonymous descriptor in the configured variable.
+It must write one bounded JSON object with `schema_version: 1`, `outcome`, and
+`exit_code`; a bounded optional `reason` is accepted and discarded. Supported
+outcomes are `succeeded`, `failed`, and `skipped-by-hold`. Failure requires a
+nonzero exit, and the other outcomes require zero. The declared exit must match
+the actual process exit. Duplicate/extra keys, malformed JSON, missing results,
+signals, output overflow, timeout and unconfirmed cleanup raise `ExecutionUnknown`.
+Neither exit zero nor a log message is sufficient evidence of success.
+
+The caller passes this callback to `scheduled_receiver.receive`, after the
+receiver has durably claimed and admitted the slot. A valid hold remains distinct
+from success. Lost publication replays the retained outcome without executing the
+child again. Unknown execution preserves the claim for operator disposition;
+it must never cause an automatic second execution.
+
+Combined stdout/stderr is drained and discarded under a byte limit. Results have
+a separate 4096-byte limit. A new process group permits cleanup of ordinary
+descendants, but is not kernel containment: a process that creates a new session
+can escape that group. Host execution still needs its approved cgroup/supervisor,
+immutable artifact and manifest admission, credential policy and timeout policy.
+The adapter does not provide parent-session survival or artifact custody.
+
+Outcome metadata is unsigned. The surrounding signed receiver/port must retain
+the documented distinction between signed lifecycle state and unsigned metadata.
+This module neither enables a schedule nor installs a production executor.
+
+Run `python scripts/demo-fixed-outcome-executor.py` for isolated child-process
+examples. The retained recording shows a real descriptor result, an explicit
+hold, a nonzero failure, and an exit-zero child whose missing descriptor result
+is refused. Tests also check timeout cleanup and durable no-repeat behavior.