Configure opt-in scheduled sinks and drain owned receiver workers

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/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/scheduled-sink-factory.md

@@ -0,0 +1,55 @@
+# Opt-in scheduled sink configuration
+
+An emitter may use the exclusive `sink: {scheduled: ...}` kind. It cannot be
+combined with a graph, command, JSONL or alert sink. Shuttle's exact terminal
+publication receipt is the acknowledgement boundary. Submission or a local
+terminal result leaves the emitter's durable row pending.
+
+The emitter requires `trigger: changes`, `event: scheduled`, `baseline_emits:
+true`, the default `item` key and BLOCKED-to-UNBLOCKED transition. Generation
+rewriting, recovery, missing-as-from and renewal are refused: they would change
+the canonical signed slot identity or acknowledge a different protocol.
+
+The scheduled mapping has exactly these fields:
+
+| Field | Meaning |
+| --- | --- |
+| `scope` | The six immutable `ExecutionScope` fields. |
+| `ledger` | Absolute SQLite path in a provisioned, owned persistent directory. |
+| `shuttle` | `executable`, `config`, `config_sha256`, `timeout_s` for the existing process port. |
+| `execute` | Fixed `argv`, `cwd`, `env`, `result_fd_env`, `timeout_s`, `output_limit`. |
+| `workers`, `capacity` | Explicit bounds, with `1 <= workers <= capacity <= 64`. |
+
+The signed scope's `manifest_sha256` must equal
+`execution_manifest_sha256(execute)`: SHA256 of UTF-8 canonical JSON
+`{"execute": RECIPE, "version": 1}` with sorted keys, compact separators and no
+non-finite numbers. The literal recipe includes the complete explicit environment
+and resource limits. Changing an argument, environment value or bound requires
+a new digest and corresponding admitted declaration. The factory copies the
+recipe so later mutation of the input mapping cannot change execution.
+
+This digest binds the recipe, not executable file bytes. Runtime admission must
+still verify immutable artifact paths/content, parent directory custody and the
+exact declaration/profile/query. The factory does not invent an eligibility
+query, register keys or supply a catalogue adapter. Credentials belong in the
+approved local execution policy and must never be printed as configuration.
+
+Before opening a ledger, the factory checks configuration shape, emitter policy,
+recipe digest, ledger custody and the pinned Shuttle config's bytes/ownership.
+The process port checks that config again through an opened descriptor on every
+phase. Existing ledgers must be regular owned files, not symlinks. Directory
+ancestry and shared-UID isolation remain operator responsibilities.
+
+Orderly consumer exit drains every constructed closeable sink before closing the
+emitter database and releasing its consumer lease. Partial startup failures do
+the same. A completed worker still needs outbox redelivery before acknowledgement.
+Abrupt termination leaves claims unknown; cleanup is not permission to replay
+execution. Preserve both the slot ledger and emitter database across rollback.
+No claim of survival through supervisor shutdown or host reboot is made.
+
+`python scripts/demo-scheduled-sink-factory.py` uses actual temporary child
+processes and SQLite outboxes, with an explicitly synthetic Shuttle executable.
+The recorded result shows one hold outcome, one child invocation and four phase
+receipts. This is source integration evidence, not live signing, profile loading,
+schedule activation or deployment acceptance. The demo requires development test
+dependencies because it reuses the isolated process fixture.

pyproject.toml

@@ -32,7 +32,7 @@ chaski-once = "once:main"
 py-modules = [
     "alertmanager_sink", "async_scheduled_sink", "change_stream", "chaski", "command_sink",
     "consumer_lease", "demo", "emitter", "external_event", "failure_family_adapter",
-    "fixed_outcome_executor", "graph_sink", "incremental", "once", "scheduled_event_sink", "scheduled_projection", "scheduled_receiver", "shuttle_process_port", "slot_receiver",
+    "fixed_outcome_executor", "graph_sink", "incremental", "once", "scheduled_event_sink", "scheduled_projection", "scheduled_receiver", "scheduled_sink_factory", "shuttle_process_port", "slot_receiver",
 ]
 
 [tool.pytest.ini_options]

scheduled_sink_factory.py

@@ -0,0 +1,102 @@
+"""Build an opt-in scheduled receiver from trusted local configuration only."""
+from __future__ import annotations
+
+from dataclasses import fields
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import stat
+
+from async_scheduled_sink import AsyncScheduledSink
+from fixed_outcome_executor import FixedOutcomeExecutor
+from scheduled_event_sink import ScheduledEventSink
+from scheduled_receiver import ExecutionScope
+from shuttle_process_port import ShuttleProcessPort
+from slot_receiver import SlotLedger
+
+
+def _keys(value, expected):
+    if type(value) is not dict or set(value) != set(expected):
+        raise ValueError('complete supported scheduled sink configuration required')
+
+
+def execution_manifest_sha256(recipe):
+    """Hash the literal versioned execution recipe; not executable file bytes."""
+    raw = json.dumps({'version': 1, 'execute': recipe}, sort_keys=True,
+                     separators=(',', ':'), allow_nan=False).encode()
+    return hashlib.sha256(raw).hexdigest()
+
+
+def validate(config):
+    """Validate and copy fixed execution policy, without opening a ledger."""
+    _keys(config, ('scope', 'ledger', 'shuttle', 'execute', 'workers', 'capacity'))
+    _keys(config['scope'], (f.name for f in fields(ExecutionScope)))
+    scope = ExecutionScope(**config['scope'])
+    scope.validate()
+    _keys(config['shuttle'], ('executable', 'config', 'config_sha256', 'timeout_s'))
+    shuttle = dict(config['shuttle'])
+    # Constructor validation has no subprocess, graph query or key operation.
+    ShuttleProcessPort(shuttle['executable'], shuttle['config'], shuttle['config_sha256'],
+                       scope, 0, timeout_s=shuttle['timeout_s'])
+    _keys(config['execute'], ('argv', 'cwd', 'env', 'result_fd_env', 'timeout_s', 'output_limit'))
+    execute = FixedOutcomeExecutor(**config['execute'])
+    if execution_manifest_sha256(config['execute']) != scope.manifest_sha256:
+        raise ValueError('execution recipe differs from the signed manifest digest')
+    workers, capacity = config['workers'], config['capacity']
+    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')
+    ledger = config['ledger']
+    if type(ledger) is not str or not Path(ledger).is_absolute():
+        raise ValueError('absolute persistent ledger path required')
+    return scope, ledger, shuttle, execute, workers, capacity
+
+
+def _check_local_files(ledger, shuttle):
+    """Early feedback only; the process port rechecks pinned config per phase."""
+    parent = Path(ledger).parent
+    info = parent.stat()
+    if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o022:
+        raise ValueError('provisioned owned ledger parent must not be writable by others')
+    if os.path.lexists(ledger):
+        info = Path(ledger).lstat()
+        if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o022:
+            raise ValueError('existing ledger must be an owned regular file')
+    fd = os.open(shuttle['config'], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+    with os.fdopen(fd, 'rb') as source:
+        info = os.fstat(source.fileno())
+        if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or
+                info.st_mode & 0o022 or info.st_size > 65536):
+            raise ValueError('pinned Shuttle config custody unavailable')
+        raw = source.read(65537)
+        if len(raw) > 65536 or hashlib.sha256(raw).hexdigest() != shuttle['config_sha256']:
+            raise ValueError('Shuttle config differs from pinned bytes')
+
+
+def validate_emitter(emitter):
+    if (emitter.event != 'scheduled' or not emitter.baseline_emits or
+            emitter.generation_event_ids or emitter.resolve_recovery or
+            emitter.missing_is_from or emitter.renew_interval_s or not emitter.change_driven or
+            emitter.externally_triggered or emitter.key != 'item' or
+            emitter.from_verdict != 'BLOCKED' or emitter.to_verdict != 'UNBLOCKED' or
+            type(emitter.label) is not str or
+            not re.fullmatch(r'[A-Za-z][A-Za-z0-9_.-]{0,127}', emitter.label) or
+            type(emitter.owner) is not str or not 0 < len(emitter.owner) <= 256):
+        raise ValueError('scheduled sink requires canonical non-recovering scheduled events')
+
+
+def build(emitter, config):
+    scope, ledger, shuttle, execute, workers, capacity = validate(config)
+    validate_emitter(emitter)
+    _check_local_files(ledger, shuttle)
+    receiver = AsyncScheduledSink(SlotLedger(ledger), scope,
+        lambda slot: ShuttleProcessPort(shuttle['executable'], shuttle['config'],
+            shuttle['config_sha256'], scope, slot, timeout_s=shuttle['timeout_s']),
+        execute, workers=workers, capacity=capacity)
+    try:
+        return ScheduledEventSink(receiver, emitter.label, emitter.owner)
+    except BaseException:
+        receiver.close()
+        raise

scripts/demo-scheduled-sink-factory.py

@@ -0,0 +1,48 @@
+"""Actual local processes and outbox; Shuttle admission/signing is a fixture."""
+import importlib.util
+import json
+import logging
+from pathlib import Path
+import sys
+import tempfile
+
+source = (Path(sys.argv[2]).resolve() if len(sys.argv) == 3 and sys.argv[1] == '--source'
+          else Path(__file__).resolve().parents[1])
+sys.path.insert(0, str(source))
+from chaski import make_sink  # noqa: E402
+
+print('Existing sink constructor positive control:', callable(make_sink))
+present = importlib.util.find_spec('scheduled_sink_factory') is not None
+print('Configured scheduled sink factory available:', present)
+if not present:
+    sys.exit(0)
+from emitter import connect, deliver_pending, observe  # noqa: E402
+from scheduled_receiver import run_id  # noqa: E402
+from test_scheduled_sink_factory import configured  # noqa: E402
+
+# The acknowledgement counts below show the expected asynchronous pending
+# state; omit its long canonical event identifier from this short recording.
+logging.getLogger('chaski.emitter').setLevel(logging.ERROR)
+with tempfile.TemporaryDirectory() as root:
+    load, _, _, scope, trace, counter, _, _ = configured.__wrapped__(Path(root))
+    definition, options = load()
+    sink = make_sink(definition, options, 'http://unused.invalid')
+    conn = connect(Path(root)/'outbox.sqlite')
+    try:
+        observe(conn, definition, [dict(item=scope.job, owner=definition.owner,
+                verdict='UNBLOCKED', event_id=run_id(scope, 600))], 600)
+        print('Submit delivery, acknowledged/pending:',
+              deliver_pending(conn, sink, 600, definition.label))
+        with sink.receiver._lock:
+            future = next(iter(sink.receiver._pending.values()))
+        future.result(timeout=5)
+        print('After terminal publication, acknowledged/pending:',
+              deliver_pending(conn, sink, 1600, definition.label))
+        phases = [json.loads(x) for x in trace.read_text().splitlines()]
+        print('Phase receipts:', [p['action'] for p in phases])
+        print('Explicit child outcome:', phases[2]['packet']['outcome'])
+        print('Actual child invocations:', counter.read_text())
+        print('No production graph, schedule, key or cron was used; signing is a fixture.')
+    finally:
+        sink.close()
+        conn.close()