From 88edbde14fda818d083cc9a88ee6ec77231e006a Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Thu, 15 Oct 2026 05:52:32 -0400 Subject: [PATCH 1/7] feat(receiver): bridge signed phases through the fixed Shuttle CLI --- docs/demos/shuttle-process.log | 8 ++ docs/demos/shuttle-process.timing | 2 + docs/scheduled-receiver.md | 36 +++++++++ pyproject.toml | 2 +- scripts/demo-shuttle-process.py | 46 +++++++++++ shuttle_process_port.py | 95 ++++++++++++++++++++++ test_shuttle_process_port.py | 126 ++++++++++++++++++++++++++++++ 7 files changed, 314 insertions(+), 1 deletion(-) create mode 100644 docs/demos/shuttle-process.log create mode 100644 docs/demos/shuttle-process.timing create mode 100644 scripts/demo-shuttle-process.py create mode 100644 shuttle_process_port.py create mode 100644 test_shuttle_process_port.py diff --git a/docs/demos/shuttle-process.log b/docs/demos/shuttle-process.log new file mode 100644 index 0000000..fbe56dd --- /dev/null +++ b/docs/demos/shuttle-process.log @@ -0,0 +1,8 @@ +Script started on 2026-10-15 05:52:16-04:00 [COMMAND="stty -onlcr; python scripts/demo-shuttle-process.py" ] +Actual subprocess and durable slot ledger; CLI receipts are a fixture. +Held result: skipped-by-hold +Delivery attempts: 2; execution callbacks: 1 +Repeat result identical: True +Changed pinned config: REFUSED before CLI invocation + +Script done on 2026-10-15 05:52:16-04:00 [COMMAND_EXIT_CODE="0"] diff --git a/docs/demos/shuttle-process.timing b/docs/demos/shuttle-process.timing new file mode 100644 index 0000000..1725bd9 --- /dev/null +++ b/docs/demos/shuttle-process.timing @@ -0,0 +1,2 @@ +0.244914 175 +0.000139 53 diff --git a/docs/scheduled-receiver.md b/docs/scheduled-receiver.md index 319eb3c..20e1dda 100644 --- a/docs/scheduled-receiver.md +++ b/docs/scheduled-receiver.md @@ -51,3 +51,39 @@ publication the second time. The small recording is retained with this source proposal, replayed using `scriptreplay --log-out scheduled-receiver.log --log-timing scheduled-receiver.timing`. It requires downloading both files; no inline video playback is claimed. + +## Separate Shuttle environment + +`shuttle_process_port.ShuttleProcessPort` implements the same four-phase port +through a fixed absolute Shuttle CLI path. Supply an absolute local config path, +its SHA256, the admitted `ExecutionScope` and UTC slot. Each call uses +`--require-existing-keys scheduled `, JSON on stdin and a bounded timeout; +a delivery cannot choose executable, arguments, config, graph query or key. +The config must be a regular file owned by the current Unix user, at most 64KiB, +and not writable by group or others. The final component must not be a symlink. +The hashed opened descriptor is passed to Shuttle through `/dev/fd`, so replacing +the config pathname does not substitute different bytes. The executable and +parent directories require operator custody; this is not isolation from another +process running as the same user or an executable hash admission mechanism. + +The CLI must return its exact verified phase receipt. Nonzero exit, timeout, +oversized output, an empty success, wrong phase or malformed receipt is UNKNOWN +and cannot authorize execution. The receiver's existing claim/replay behavior +still applies: failure after claim does not rerun the child. A successful +`publish` means the Shuttle SDK accepted its exact registered-signature read-back, +not merely that this process returned zero. + +This library does not launch a background worker or make a long child safe to +execute in a synchronous command sink. An asynchronous domain execution handoff +and installed child-survival acceptance are separate runtime prerequisites. + +Run `python -m pytest -q test_shuttle_process_port.py` for real process and file +descriptor controls. Its CLI is explicitly a fixture that records packets and +returns phase receipts; it does not prove real keys, registration, graph export, +macOS acceptance or production scheduling. Those require the installed Shuttle +SDK and the scoped runtime smoke. + +Process bridge demo: `python scripts/demo-shuttle-process.py` shows two deliveries, +one execution callback and config drift refusal using actual subprocesses and +a durable ledger. [Recording](demos/shuttle-process.log) and +[timing](demos/shuttle-process.timing) retain that fixture-only limitation. diff --git a/pyproject.toml b/pyproject.toml index 4d639aa..c114451 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,7 @@ chaski-once = "once:main" py-modules = [ "alertmanager_sink", "change_stream", "chaski", "command_sink", "consumer_lease", "demo", "emitter", "external_event", "failure_family_adapter", - "graph_sink", "incremental", "once", "scheduled_receiver", "slot_receiver", + "graph_sink", "incremental", "once", "scheduled_receiver", "shuttle_process_port", "slot_receiver", ] [tool.pytest.ini_options] diff --git a/scripts/demo-shuttle-process.py b/scripts/demo-shuttle-process.py new file mode 100644 index 0000000..61e1ebc --- /dev/null +++ b/scripts/demo-shuttle-process.py @@ -0,0 +1,46 @@ +"""Real subprocess bridge and slot ledger, fixture CLI receipts, no real keys/graph.""" +import hashlib +from pathlib import Path +import sys +import tempfile + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from scheduled_receiver import ExecutionScope, receive +from shuttle_process_port import PortRefused, ShuttleProcessPort +from slot_receiver import SlotLedger + +with tempfile.TemporaryDirectory() as root: + root = Path(root) + config, cli = root/'config.json', root/'fixture-cli' + config.write_text('{}') + config.chmod(0o600) + cli.write_text('#!'+sys.executable+'\n'+'''import json,sys +from pathlib import Path +assert sys.argv[1:3] == ['--require-existing-keys', 'scheduled'] +assert Path(sys.argv[5]).read_text() == '{}' +json.load(sys.stdin) +print(json.dumps({'action': sys.argv[3], 'result': 'verified'})) +''') + cli.chmod(0o700) + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') + port = ShuttleProcessPort(cli, config, hashlib.sha256(config.read_bytes()).hexdigest(), scope, 600) + calls = [] + def execute(): + calls.append(1) + return {'outcome': 'skipped-by-hold', 'exit_code': 0} + packet = {**scope.__dict__, 'scheduled_at': 600} + ledger = SlotLedger(root/'slots.sqlite') + first = receive(ledger, scope, 600, packet, port, execute) + second = receive(ledger, scope, 600, packet, port, execute) + print('Actual subprocess and durable slot ledger; CLI receipts are a fixture.') + print('Held result:', first['outcome']) + print('Delivery attempts: 2; execution callbacks:', len(calls)) + print('Repeat result identical:', first == second) + config.write_text('{"changed": true}') + try: + port.admit(scope, 600) + except PortRefused: + print('Changed pinned config: REFUSED before CLI invocation') + else: + raise AssertionError('changed config accepted') diff --git a/shuttle_process_port.py b/shuttle_process_port.py new file mode 100644 index 0000000..48b6ce2 --- /dev/null +++ b/shuttle_process_port.py @@ -0,0 +1,95 @@ +"""Fixed local Shuttle CLI bridge; no scheduler, launcher or child executor. + +The operator pins the executable and config digest. Each call passes the same +opened config descriptor to Shuttle, so replacement of the pathname cannot +change the admitted configuration. Parent-directory custody remains local. +""" +from __future__ import annotations + +from dataclasses import asdict +import hashlib +import json +import os +from pathlib import Path +import stat +import subprocess +import tempfile + +from scheduled_receiver import ExecutionScope, run_id + + +class PortRefused(RuntimeError): + """No verified phase receipt; never authorizes child execution or replay.""" + + +class ShuttleProcessPort: + def __init__(self, executable: Path, config: Path, config_sha256: str, + scope: ExecutionScope, scheduled_at: int, *, timeout_s: float = 30): + scope.validate() + self.run = run_id(scope, scheduled_at) + if not Path(executable).is_absolute() or not Path(config).is_absolute(): + raise ValueError("absolute operator-owned executable and config required") + if (type(config_sha256) is not str or len(config_sha256) != 64 or + any(c not in '0123456789abcdef' for c in config_sha256)): + raise ValueError("pinned config SHA256 required") + if type(timeout_s) not in (int, float) or not 0 < timeout_s <= 60: + raise ValueError("bounded phase timeout required") + self.executable, self.config = str(executable), Path(config) + self.digest, self.scope, self.slot = config_sha256, scope, scheduled_at + self.timeout_s = timeout_s + + def _call(self, action, scope, run=None, outcome=None): + if type(scope) is not ExecutionScope or scope != self.scope: + raise PortRefused("scope differs from pinned delivery") + if action != 'admit' and run != self.run: + raise PortRefused("run differs from pinned slot") + packet = {'scope': asdict(scope), 'scheduled_at': self.slot} + if action != 'admit': + packet['run'] = run + if outcome is not None: + packet['outcome'] = outcome + try: + fd = os.open(self.config, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + with os.fdopen(fd, 'rb') as config: + info = os.fstat(config.fileno()) + if (not stat.S_ISREG(info.st_mode) or info.st_size > 65536 or + info.st_uid != os.getuid() or info.st_mode & 0o022): + raise PortRefused("config must be bounded, owned and not writable by others") + raw = config.read(65537) + if len(raw) > 65536 or hashlib.sha256(raw).hexdigest() != self.digest: + raise PortRefused("config differs from pinned bytes") + config.seek(0) + # Temporary files bound memory use even if a broken SDK emits + # more output than its small receipt contract permits. + with tempfile.TemporaryFile() as out, tempfile.TemporaryFile() as err: + result = subprocess.run( + [self.executable, '--require-existing-keys', 'scheduled', action, + '--config', f'/dev/fd/{config.fileno()}'], + input=json.dumps(packet, allow_nan=False).encode(), + stdout=out, stderr=err, pass_fds=(config.fileno(),), + timeout=self.timeout_s, check=False) + if result.returncode or out.tell() > 4096 or err.tell() > 4096: + raise PortRefused("Shuttle phase failed or exceeded receipt bound") + out.seek(0) + receipt = out.read(4097) + expected = {'action': action, 'result': 'verified'} + # Exact canonical receipt rejects duplicate keys, trailing output, + # wrong action, and a generic exit-zero without verification. + if receipt.strip() != json.dumps(expected).encode(): + raise PortRefused("Shuttle did not return the exact phase receipt") + except (OSError, subprocess.TimeoutExpired, ValueError) as exc: + raise PortRefused(f"Shuttle phase unavailable: {type(exc).__name__}") from None + + def admit(self, scope, scheduled_at): + if type(scheduled_at) is not int or scheduled_at != self.slot: + raise PortRefused("delivery slot differs from pinned slot") + self._call('admit', scope) + + def started(self, run, scope): + self._call('started', scope, run) + + def finished(self, run, scope, outcome): + self._call('finished', scope, run, outcome) + + def publish(self, run, scope, outcome): + self._call('publish', scope, run, outcome) diff --git a/test_shuttle_process_port.py b/test_shuttle_process_port.py new file mode 100644 index 0000000..c15fe02 --- /dev/null +++ b/test_shuttle_process_port.py @@ -0,0 +1,126 @@ +"""Real process/descriptor controls; subprocess is a fixture CLI, not Shuttle crypto.""" +from dataclasses import replace +import hashlib +import json +import sys + +import pytest + +from scheduled_receiver import ExecutionScope, receive, run_id +from shuttle_process_port import PortRefused, ShuttleProcessPort +from slot_receiver import SlotLedger + + +@pytest.fixture +def setup(tmp_path): + trace = tmp_path / 'trace.jsonl' + config = tmp_path / 'config.json' + config.write_text(json.dumps({'trace': str(trace), 'mode': 'ok'})) + config.chmod(0o600) + executable = tmp_path / 'shuttle-fixture' + executable.write_text('#!' + sys.executable + '\n' + '''import json,sys,time +from pathlib import Path +assert sys.argv[1:3] == ['--require-existing-keys', 'scheduled'] +action = sys.argv[3] +assert sys.argv[4] == '--config' +config = json.loads(Path(sys.argv[5]).read_text()) +packet = json.load(sys.stdin) +with open(config['trace'], 'a') as out: + out.write(json.dumps({'action': action, 'packet': packet})+'\\n') +mode = config['mode'] +if mode == 'timeout': time.sleep(10) +if mode == 'fail': sys.exit(3) +if mode == 'large': print('x'*4097) +elif mode == 'duplicate': print('{"action":"bad","action":"'+action+'","result":"verified"}') +elif mode == 'wrong': print(json.dumps({'action': 'wrong', 'result': 'verified'})) +elif mode == 'empty': pass +else: print(json.dumps({'action': action, 'result': 'verified'})) +''') + executable.chmod(0o700) + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') + def port(mode='ok', timeout=5): + config.write_text(json.dumps({'trace': str(trace), 'mode': mode})) + return ShuttleProcessPort(executable, config, hashlib.sha256(config.read_bytes()).hexdigest(), + scope, 600, timeout_s=timeout) + return port, scope, trace, config, executable + + +def test_real_process_calls_all_phases_and_duplicate_only_publishes(setup, tmp_path): + factory, scope, trace, _, _ = setup + port = factory() + delivery = {**scope.__dict__, 'scheduled_at': 600} + calls = [] + def execute(): + calls.append(1) + return {'outcome': 'skipped-by-hold', 'exit_code': 0} + ledger = SlotLedger(tmp_path/'ledger.sqlite') + result = receive(ledger, scope, 600, delivery, port, execute) + assert receive(ledger, scope, 600, delivery, port, execute) == result + assert calls == [1] + events = [json.loads(x) for x in trace.read_text().splitlines()] + assert [e['action'] for e in events] == ['admit', 'started', 'finished', 'publish', 'admit', 'publish'] + assert events[2]['packet']['outcome'] == {'outcome': 'skipped-by-hold', 'exit_code': 0} + assert events[2]['packet']['run'] == run_id(scope, 600) + + +@pytest.mark.parametrize('mode', ['fail', 'large', 'duplicate', 'wrong', 'empty', 'timeout']) +def test_failed_or_unverified_admission_never_executes_child(setup, tmp_path, mode): + factory, scope, _, _, _ = setup + port = factory(mode, timeout=.1 if mode == 'timeout' else 5) + with pytest.raises(PortRefused): + receive(SlotLedger(tmp_path/'ledger.sqlite'), scope, 600, + {**scope.__dict__, 'scheduled_at': 600}, port, + lambda: pytest.fail('unverified admission executed child')) + + +def test_pinned_config_drift_refuses_before_process(setup): + factory, scope, trace, config, _ = setup + port = factory() + config.write_text('{}') + with pytest.raises(PortRefused, match='pinned bytes'): + port.admit(scope, 600) + assert not trace.exists() + + +@pytest.mark.parametrize('change', ['symlink', 'writable', 'fifo']) +def test_config_custody_refuses_without_blocking_or_launch(setup, change, tmp_path): + factory, scope, trace, config, _ = setup + port = factory() + if change == 'writable': + config.chmod(0o666) + else: + config.unlink() + if change == 'symlink': + config.symlink_to(tmp_path/'elsewhere') + else: + import os + os.mkfifo(config) + with pytest.raises(PortRefused): + port.admit(scope, 600) + assert not trace.exists() + + +def test_scope_slot_and_run_mismatch_refuse_before_process(setup): + factory, scope, trace, _, _ = setup + port = factory() + for callback in [lambda: port.admit(scope, True), lambda: port.admit(scope, 601), + lambda: port.admit(replace(scope, host='urn:other'), 600), + lambda: port.started('other', scope)]: + with pytest.raises(PortRefused): + callback() + assert not trace.exists() + + +def test_replaced_config_path_does_not_replace_opened_pinned_config(setup, monkeypatch): + import subprocess + factory, scope, trace, config, _ = setup + port = factory() + run = subprocess.run + def replace_path(*args, **kwargs): + config.rename(config.with_suffix('.original')) + config.write_text('{}') + return run(*args, **kwargs) + monkeypatch.setattr(subprocess, 'run', replace_path) + port.admit(scope, 600) + assert json.loads(trace.read_text())['action'] == 'admit' From 1d9291539c9b761a97f8213dd794a0d6d5a0c7ce Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Thu, 15 Oct 2026 06:36:49 -0400 Subject: [PATCH 2/7] feat(receiver): bound asynchronous signed delivery until receipt --- async_scheduled_sink.py | 95 ++++++++++++++ docs/demos/async-scheduled.log | 13 ++ docs/demos/async-scheduled.timing | 6 + docs/scheduled-receiver.md | 39 ++++++ pyproject.toml | 2 +- scripts/demo-async-scheduled.py | 63 ++++++++++ test_async_scheduled_sink.py | 198 ++++++++++++++++++++++++++++++ 7 files changed, 415 insertions(+), 1 deletion(-) create mode 100644 async_scheduled_sink.py create mode 100644 docs/demos/async-scheduled.log create mode 100644 docs/demos/async-scheduled.timing create mode 100644 scripts/demo-async-scheduled.py create mode 100644 test_async_scheduled_sink.py diff --git a/async_scheduled_sink.py b/async_scheduled_sink.py new file mode 100644 index 0000000..3f0d52d --- /dev/null +++ b/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) diff --git a/docs/demos/async-scheduled.log b/docs/demos/async-scheduled.log new file mode 100644 index 0000000..83ba297 --- /dev/null +++ b/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" ] +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"] diff --git a/docs/demos/async-scheduled.timing b/docs/demos/async-scheduled.timing new file mode 100644 index 0000000..7e77af2 --- /dev/null +++ b/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 diff --git a/docs/scheduled-receiver.md b/docs/scheduled-receiver.md index 20e1dda..0bed821 100644 --- a/docs/scheduled-receiver.md +++ b/docs/scheduled-receiver.md @@ -1,5 +1,44 @@ # Signed scheduled delivery bridge +## Bounded asynchronous handoff + +`async_scheduled_sink.AsyncScheduledSink` runs the existing signed receiver +on a bounded thread pool. The operator supplies one immutable `ExecutionScope`, +the persistent ledger, a trusted `port_factory(scheduled_at)`, and a trusted +`execute(scheduled_at)` callback. Delivery data must repeat the exact scope and +integer UTC slot; it cannot supply commands or choose a different identity. +The callback must enforce the admitted command's runtime and output limits. + +`deliver(binding)` raises `DeliveryPending` after submission and while work is +running. It returns only on a later call after the receiver verifies terminal +publication. Existing emitter delivery therefore keeps its durable outbox row +undelivered and retries with its normal backoff. Pending counts as a delivery +failure in existing metrics; it does not mean the child failed. A terminal hold +or failed child can have a successful delivery receipt for its actual outcome. +The emitter's retry cadence bounds receipt latency independently of completion. + +Capacity includes queued, running, and completed work awaiting receipt. Duplicate +in-flight deliveries reuse one future. Worker exceptions reach the next delivery +unchanged; later retries use the persistent slot ledger, so publication may retry +but a claimed child cannot run again. Context variables are copied on submission, +including any strict-admission context set by the caller. + +The owner calls `close()` outside the reactor tick to stop admission and drain +workers. This can wait for the full callback duration. Abrupt termination leaves +claims UNKNOWN. Threads and children remain inside the owning service's cgroup; +this adapter cannot provide survival across that service's exit. + +This source adapter accepts complete bindings. A generic emitter envelope does +not contain that binding: production configuration still needs a separately +validated schedule/envelope adapter and trusted fixed-command executor. No sink +factory or production configuration enables this class automatically. Do not +acknowledge an upstream delivery merely because a future was submitted. + +Reproduce the blocked-child/receipt/capacity controls with +`python -m pytest -q test_async_scheduled_sink.py`. These use real threads, +SQLite claims and the actual emitter outbox, with synthetic signing callbacks. +They do not establish installed signing, schedule coverage or natural firings. + `scheduled_receiver.receive` connects an already admitted delivery to the persistent `SlotLedger` and a trusted `SignedRunPort`. It is a library entry point: importing it starts no listener, timer, scheduler or child process. diff --git a/pyproject.toml b/pyproject.toml index c114451..3f76081 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,7 +30,7 @@ chaski-once = "once:main" [tool.setuptools] py-modules = [ - "alertmanager_sink", "change_stream", "chaski", "command_sink", + "alertmanager_sink", "async_scheduled_sink", "change_stream", "chaski", "command_sink", "consumer_lease", "demo", "emitter", "external_event", "failure_family_adapter", "graph_sink", "incremental", "once", "scheduled_receiver", "shuttle_process_port", "slot_receiver", ] diff --git a/scripts/demo-async-scheduled.py b/scripts/demo-async-scheduled.py new file mode 100644 index 0000000..be84de5 --- /dev/null +++ b/scripts/demo-async-scheduled.py @@ -0,0 +1,63 @@ +"""Real threads and ledger; synthetic signing callbacks and no external I/O.""" +from dataclasses import asdict +import importlib.util +from pathlib import Path +import sys +import tempfile +import threading + +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 scheduled_receiver import ExecutionScope, receive # noqa: E402 +from slot_receiver import SlotLedger # noqa: E402 + +print('Source demo: real worker thread and SQLite; signing callbacks are fixtures.') +print('Existing receiver positive control:', callable(receive)) +present = importlib.util.find_spec('async_scheduled_sink') is not None +print('Async handoff available:', present) +if not present: + sys.exit(0) +from async_scheduled_sink import AsyncScheduledSink, DeliveryPending # noqa: E402 + +with tempfile.TemporaryDirectory() as root: + entered, release = threading.Event(), threading.Event() + calls, publications = [], [] + class Port: + def admit(self, scope, slot): pass + def started(self, run, scope): pass + def finished(self, run, scope, outcome): pass + def publish(self, run, scope, outcome): publications.append(outcome) + def execute(slot): + calls.append(slot) + entered.set() + assert release.wait(5) + return {'outcome': 'skipped-by-hold', 'exit_code': 0} + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'operator', 'urn:example:workflow') + delivery = {**asdict(scope), 'scheduled_at': 600} + sink = AsyncScheduledSink(SlotLedger(Path(root)/'slots.sqlite'), scope, + lambda slot: Port(), execute, capacity=1) + try: + for label in ('First delivery', 'Duplicate delivery'): + try: + sink.deliver(delivery) + except DeliveryPending: + print(label + ': PENDING, no receipt') + else: + raise AssertionError('acknowledged blocked child') + assert entered.wait(5) + print('Worker blocked while caller continued:', not release.is_set()) + print('Execution callbacks:', len(calls), '; publications:', len(publications)) + release.set() + # Demo waits for completion explicitly; deliver never waits on workers. + with sink._lock: + future = next(iter(sink._pending.values())) + future.result(timeout=5) + sink.deliver(delivery) + print('Later delivery: RECEIPT after terminal publication') + print('Published actual outcome:', publications[0]['outcome']) + print('Final execution callbacks:', len(calls)) + finally: + release.set() + sink.close() diff --git a/test_async_scheduled_sink.py b/test_async_scheduled_sink.py new file mode 100644 index 0000000..cc53a1a --- /dev/null +++ b/test_async_scheduled_sink.py @@ -0,0 +1,198 @@ +"""Real threads, durable claims and emitter receipts; no production execution.""" +from contextvars import ContextVar +from dataclasses import asdict +import json +import threading + +import pytest + +from async_scheduled_sink import AsyncScheduledSink, DeliveryPending +from emitter import connect, deliver_pending +from scheduled_receiver import ExecutionScope +from slot_receiver import SlotLedger, SlotUnknown + + +@pytest.fixture +def rig(tmp_path): + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'operator', 'urn:example:workflow') + entered, release, published = (threading.Event() for _ in range(3)) + calls, phases = [], [] + fail_publish = [False] + + class Port: + def admit(self, scope, slot): + phases.append('admit') + + def started(self, run, scope): + phases.append('started') + + def finished(self, run, scope, outcome): + phases.append('finished') + + def publish(self, run, scope, outcome): + phases.append('publish') + if fail_publish[0]: + raise ConnectionError('publication unavailable') + published.set() + + def execute(slot): + calls.append(slot) + entered.set() + assert release.wait(5), 'test did not release blocked worker' + return {'outcome': 'skipped-by-hold', 'exit_code': 0} + + ledger = SlotLedger(tmp_path/'slots.sqlite') + sink = AsyncScheduledSink(ledger, scope, lambda slot: Port(), execute, + workers=1, capacity=1) + yield sink, scope, entered, release, published, calls, phases, fail_publish, ledger + release.set() + sink.close() + + +def binding(scope, slot=600): + return {**asdict(scope), 'scheduled_at': slot} + + +def drain_worker(sink): + # Wait for actual worker completion without timing guesses; only the test + # waits, never deliver() or the reactor. + with sink._lock: + futures = list(sink._pending.values()) + for future in futures: + try: + future.result(timeout=5) + except ConnectionError: + pass + + +def test_emitter_stays_undelivered_while_child_blocks_then_receipts(rig, tmp_path): + sink, scope, entered, release, published, calls, phases, _, _ = rig + conn = connect(tmp_path/'emitter.sqlite') + try: + conn.execute('INSERT INTO outbox(event_id,emitter,item,generation,payload,created) ' + 'VALUES(?,?,?,?,?,?)', + ('event1', 'schedule', scope.job, 1, json.dumps(binding(scope)), 0)) + assert deliver_pending(conn, sink, 0, 'schedule') == (0, 1) + assert entered.wait(5) + assert conn.execute('SELECT delivered_at FROM outbox').fetchone() == (None,) + # A real running child remains blocked, yet delivery can be called + # again immediately and cannot launch another child. + assert deliver_pending(conn, sink, 30, 'schedule') == (0, 1) + assert calls == [600] and not published.is_set() + release.set() + drain_worker(sink) + assert published.is_set() + assert deliver_pending(conn, sink, 90, 'schedule') == (1, 0) + assert conn.execute('SELECT delivered_at FROM outbox').fetchone() == (90,) + assert phases == ['admit', 'started', 'finished', 'publish'] + finally: + conn.close() + + +def test_capacity_includes_finished_unacknowledged_work_and_reuses_slot(rig): + sink, scope, entered, release, _, calls, _, _, _ = rig + with pytest.raises(DeliveryPending): + sink.deliver(binding(scope)) + assert entered.wait(5) + with pytest.raises(DeliveryPending, match='capacity'): + sink.deliver(binding(scope, 1200)) + release.set() + drain_worker(sink) + with pytest.raises(DeliveryPending, match='capacity'): + sink.deliver(binding(scope, 1200)) + sink.deliver(binding(scope)) + with pytest.raises(DeliveryPending): + sink.deliver(binding(scope, 1200)) + drain_worker(sink) + sink.deliver(binding(scope, 1200)) + assert calls == [600, 1200] + + +def test_publication_retry_and_receiver_recreation_never_repeat_child(rig): + sink, scope, entered, release, _, calls, _, fail_publish, ledger = rig + fail_publish[0] = True + with pytest.raises(DeliveryPending): + sink.deliver(binding(scope)) + assert entered.wait(5) + release.set() + drain_worker(sink) + with pytest.raises(ConnectionError, match='publication unavailable'): + sink.deliver(binding(scope)) + fail_publish[0] = False + with pytest.raises(DeliveryPending): + sink.deliver(binding(scope)) + drain_worker(sink) + sink.deliver(binding(scope)) + sink.close() + replacement = AsyncScheduledSink(ledger, scope, sink.port_factory, + lambda slot: pytest.fail('replayed child')) + try: + with pytest.raises(DeliveryPending): + replacement.deliver(binding(scope)) + drain_worker(replacement) + replacement.deliver(binding(scope)) + assert calls == [600] + finally: + replacement.close() + + +def test_claimed_unknown_slot_stays_unknown_without_child(rig): + sink, scope, _, _, _, calls, _, _, ledger = rig + from slot_receiver import slot_id + with pytest.raises(RuntimeError): + ledger.run(slot_id(scope.job, scope.revision, 600), binding(scope), + lambda: (_ for _ in ()).throw(RuntimeError('crashed after claim'))) + with pytest.raises(DeliveryPending): + sink.deliver(binding(scope)) + with pytest.raises(SlotUnknown): + drain_worker(sink) + with pytest.raises(SlotUnknown): + sink.deliver(binding(scope)) + assert calls == [] + + +@pytest.mark.parametrize('change', [{'host': 'urn:example:other'}, + {'scheduled_at': True}, {'command': 'anything'}, + {'scheduled_at': -1}]) +def test_bad_binding_refuses_before_admission(rig, change): + sink, scope, _, _, _, calls, phases, _, _ = rig + with pytest.raises(ValueError): + sink.deliver({**binding(scope), **change}) + assert not sink._pending and calls == [] and phases == [] + + +def test_context_is_copied_at_submit_and_restored_in_caller(rig): + sink, scope, _, _, _, _, _, _, _ = rig + flag = ContextVar('strict_admission', default=False) + observed = [] + def execute(slot): + observed.append(flag.get()) + flag.set(False) + return {'outcome': 'succeeded', 'exit_code': 0} + sink.execute = execute + token = flag.set(True) + try: + with pytest.raises(DeliveryPending): + sink.deliver(binding(scope)) + drain_worker(sink) + sink.deliver(binding(scope)) + assert observed == [True] and flag.get() is True + finally: + flag.reset(token) + + +def test_close_refuses_new_work(rig): + sink, scope, _, _, _, calls, phases, _, _ = rig + sink.close() + with pytest.raises(ConnectionError, match='closed'): + sink.deliver(binding(scope)) + assert calls == [] and phases == [] + + +@pytest.mark.parametrize('workers,capacity', [(0, 1), (True, 1), (2, 1), (1, 65)]) +def test_unbounded_or_invalid_capacity_refuses(rig, workers, capacity): + sink, scope, _, _, _, _, _, _, ledger = rig + with pytest.raises(ValueError): + AsyncScheduledSink(ledger, scope, sink.port_factory, sink.execute, + workers=workers, capacity=capacity) From 0b4a0eb56800020bc21846927d1c21514b179493 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Thu, 15 Oct 2026 07:20:14 -0400 Subject: [PATCH 3/7] feat(receiver): verify native scheduled emitter envelopes --- docs/demos/scheduled-event.log | 12 +++ docs/demos/scheduled-event.timing | 6 ++ docs/scheduled-receiver.md | 43 +++++++- pyproject.toml | 2 +- scheduled_event_sink.py | 75 ++++++++++++++ scripts/demo-scheduled-event.py | 75 ++++++++++++++ test_scheduled_event_sink.py | 162 ++++++++++++++++++++++++++++++ 7 files changed, 369 insertions(+), 6 deletions(-) create mode 100644 docs/demos/scheduled-event.log create mode 100644 docs/demos/scheduled-event.timing create mode 100644 scheduled_event_sink.py create mode 100644 scripts/demo-scheduled-event.py create mode 100644 test_scheduled_event_sink.py diff --git a/docs/demos/scheduled-event.log b/docs/demos/scheduled-event.log new file mode 100644 index 0000000..c650d33 --- /dev/null +++ b/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" ] +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"] diff --git a/docs/demos/scheduled-event.timing b/docs/demos/scheduled-event.timing new file mode 100644 index 0000000..d7f590f --- /dev/null +++ b/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 diff --git a/docs/scheduled-receiver.md b/docs/scheduled-receiver.md index 0bed821..b5fa094 100644 --- a/docs/scheduled-receiver.md +++ b/docs/scheduled-receiver.md @@ -28,17 +28,50 @@ workers. This can wait for the full callback duration. Abrupt termination leaves claims UNKNOWN. Threads and children remain inside the owning service's cgroup; this adapter cannot provide survival across that service's exit. -This source adapter accepts complete bindings. A generic emitter envelope does -not contain that binding: production configuration still needs a separately -validated schedule/envelope adapter and trusted fixed-command executor. No sink -factory or production configuration enables this class automatically. Do not -acknowledge an upstream delivery merely because a future was submitted. +The asynchronous receiver accepts complete bindings. `ScheduledEventSink` below +converts native emitter envelopes into that binding. Production configuration +still needs a validated catalogue projection and trusted fixed-command executor. +No sink factory or production configuration enables either class automatically. +Do not acknowledge an upstream delivery merely because a future was submitted. Reproduce the blocked-child/receipt/capacity controls with `python -m pytest -q test_async_scheduled_sink.py`. These use real threads, SQLite claims and the actual emitter outbox, with synthetic signing callbacks. They do not establish installed signing, schedule coverage or natural firings. +### Native emitter envelopes + +Wrap the asynchronous receiver with +`scheduled_event_sink.ScheduledEventSink(receiver, emitter, recipient)`. The +operator pins the emitter label, human routing recipient and receiver scope. +The adapter accepts the existing emitter's `scheduled` envelope only when its +item and routing match those pins. It decodes the slot from the canonical +`scheduled-v1-` event identifier and re-encodes the complete expected scope to +verify exact equality. Alternate encodings, foreign scopes, duplicate or extra +JSON fields, missing fields, boolean slots and unbounded timestamps refuse +before the asynchronous receiver is called. Delivery timestamps and emitter +generation cannot replace the frozen scheduled slot. + +The adapter exposes a stable delivery scope containing the emitter, recipient +and inner receiver scope. Existing outbox delivery freezes this scope before +its first call, preserving the original binding across retries. A scheduling +adapter must emit `run_id(scope, scheduled_at)` as its deterministic `event_id` +and use `event: scheduled`; it must not set `generation_event_ids`, which +replaces that identifier with a hash. Set `baseline_emits` when an already-due +slot should emit during bootstrap. These options require the reviewed schedule +projection; this envelope adapter cannot infer a schedule or admit execution. + +`test_scheduled_event_sink.py` uses actual `observe` and `deliver_pending` with +the durable outbox, real worker threads and slot ledger. Signing callbacks are +synthetic. It verifies one blocked execution, no early receipt and terminal +publication before acknowledgment, alongside rejected scopes and malformed +events. All execution admission remains the trusted signed port's job. + +Replay `python scripts/demo-scheduled-event.py`, or download the output-only +[terminal recording](demos/scheduled-event.log) and [timing](demos/scheduled-event.timing) +for `scriptreplay`. The demo prints actual outbox and receipt state with +synthetic signing callbacks; it does not contact a service or run a real job. + `scheduled_receiver.receive` connects an already admitted delivery to the persistent `SlotLedger` and a trusted `SignedRunPort`. It is a library entry point: importing it starts no listener, timer, scheduler or child process. diff --git a/pyproject.toml b/pyproject.toml index 3f76081..6de0bc1 100644 --- a/pyproject.toml +++ b/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", - "graph_sink", "incremental", "once", "scheduled_receiver", "shuttle_process_port", "slot_receiver", + "graph_sink", "incremental", "once", "scheduled_event_sink", "scheduled_receiver", "shuttle_process_port", "slot_receiver", ] [tool.pytest.ini_options] diff --git a/scheduled_event_sink.py b/scheduled_event_sink.py new file mode 100644 index 0000000..7dc22b6 --- /dev/null +++ b/scheduled_event_sink.py @@ -0,0 +1,75 @@ +"""Translate a pinned emitter envelope into the existing scheduled binding. + +Only the canonical event identifier carries the frozen scope and UTC slot. +Delivery time, generation and evidence can never choose a command or slot. +""" +from __future__ import annotations + +import base64 +import json +import math +import re + +from scheduled_receiver import run_id + + +class ScheduledEventSink: + def __init__(self, receiver, emitter: str, recipient: str): + if (type(emitter) is not str or + not re.fullmatch(r'[A-Za-z][A-Za-z0-9_.-]{0,127}', emitter) or + type(recipient) is not str or not 0 < len(recipient) <= 256): + raise ValueError('pinned emitter and recipient required') + receiver.scope.validate() + self.receiver, self.scope = receiver, receiver.scope + self.emitter, self.recipient = emitter, recipient + + def delivery_scope(self): + return {'receiver': 'scheduled-event-v1', 'emitter': self.emitter, + 'recipient': self.recipient, 'binding': self.receiver.delivery_scope()} + + def deliver(self, event: dict) -> None: + required = {'event_id', 'event', 'emitter', 'item', 'generation', + 'recipient', 'evidence', 'observed_at'} + if (type(event) is not dict or not required <= event.keys() or + event.keys() - required - {'work_kind'}): + raise ValueError('complete supported emitter envelope required') + if (event['event'] != 'scheduled' or event['emitter'] != self.emitter or + event['item'] != self.scope.job or event['recipient'] != self.recipient): + raise ValueError('event differs from pinned receiver routing') + generation, observed = event['generation'], event['observed_at'] + if (type(generation) is not int or not 0 <= generation < 2**63 or + type(observed) not in (int, float) or + not 0 <= observed <= 253402300799 or not math.isfinite(observed)): + raise ValueError('invalid emitter metadata') + evidence = event['evidence'] + if evidence is not None and (type(evidence) is not str or len(evidence) > 8192): + raise ValueError('bounded textual evidence required') + if 'work_kind' in event and (type(event['work_kind']) is not str or + not re.fullmatch(r'[A-Za-z][A-Za-z0-9_-]{0,127}', event['work_kind'])): + raise ValueError('invalid work kind') + identity = event['event_id'] + if (type(identity) is not str or len(identity) > 24000 or + not identity.startswith('scheduled-v1-')): + raise ValueError('bounded canonical scheduled identifier required') + try: + encoded = identity[len('scheduled-v1-'):] + raw = base64.b64decode(encoded + '=' * (-len(encoded) % 4), + altchars=b'-_', validate=True) + binding = json.loads(raw) + if type(binding) is not dict: + raise ValueError('scheduled binding must be an object') + slot = binding.get('scheduled_at') + if type(slot) is not int or not 0 <= slot <= 253402300799: + raise ValueError('bounded integer UTC slot required') + # Exact re-encoding refuses a foreign scope, duplicate JSON keys, + # extra fields, booleans in integer positions, padding and alternate + # JSON spellings. The observed delivery timestamp is not an anchor. + if run_id(self.scope, slot) != identity: + raise ValueError('scheduled identifier differs from pinned scope') + except (ValueError, UnicodeError, RecursionError) as exc: + raise ValueError('invalid canonical scheduled identifier') from exc + del binding['version'] + self.receiver.deliver(binding) + + def close(self): + self.receiver.close() diff --git a/scripts/demo-scheduled-event.py b/scripts/demo-scheduled-event.py new file mode 100644 index 0000000..064fed0 --- /dev/null +++ b/scripts/demo-scheduled-event.py @@ -0,0 +1,75 @@ +"""Actual emitter/outbox, thread and ledger; synthetic signing, no network.""" +import importlib.util +import json +import logging +from pathlib import Path +import sys +import tempfile +import threading + +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 async_scheduled_sink import AsyncScheduledSink # noqa: E402 +from emitter import Emitter, connect, deliver_pending, observe # noqa: E402 +from scheduled_receiver import ExecutionScope, run_id # noqa: E402 +from slot_receiver import SlotLedger # noqa: E402 + +# Pending delivery is expected here. Show the measured outbox/receipt state +# below instead of logging the entire reversible fixture identifier twice. +logging.getLogger('chaski.emitter').setLevel(logging.ERROR) + +print('Source demo: actual emitter/outbox, worker and ledger; signing is a fixture.') +print('Existing async receiver positive control:', callable(AsyncScheduledSink.deliver)) +present = importlib.util.find_spec('scheduled_event_sink') is not None +print('Native scheduled envelope adapter available:', present) +if not present: + sys.exit(0) +from scheduled_event_sink import ScheduledEventSink # noqa: E402 + +with tempfile.TemporaryDirectory() as root: + root = Path(root) + entered, release = threading.Event(), threading.Event() + calls, publications = [], [] + class Port: + def admit(self, scope, slot): pass + def started(self, run, scope): pass + def finished(self, run, scope, outcome): pass + def publish(self, run, scope, outcome): publications.append(outcome) + def execute(slot): + calls.append(slot) + entered.set() + assert release.wait(5) + return {'outcome': 'skipped-by-hold', 'exit_code': 0} + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') + definition = Emitter('schedule', ['unused'], 60, 'operator', + event='scheduled', baseline_emits=True) + conn = connect(root/'emitter.sqlite') + inner = AsyncScheduledSink(SlotLedger(root/'slots.sqlite'), scope, + lambda slot: Port(), execute) + sink = ScheduledEventSink(inner, 'schedule', 'operator') + try: + queued = observe(conn, definition, [{'item': scope.job, 'verdict': 'UNBLOCKED', + 'event_id': run_id(scope, 600), 'owner': 'operator'}], 600) + print('Observed due declaration; durable outbox rows:', len(queued)) + assert deliver_pending(conn, sink, 600, 'schedule') == (0, 1) + assert entered.wait(5) + assert deliver_pending(conn, sink, 630, 'schedule') == (0, 1) + receipt = conn.execute('SELECT delivered_at FROM outbox').fetchone()[0] + print('Two delivery attempts; child still blocked:', not release.is_set()) + print('Execution callbacks:', len(calls), '; published:', len(publications), '; receipt:', receipt) + release.set() + with inner._lock: + future = next(iter(inner._pending.values())) + future.result(timeout=5) + assert deliver_pending(conn, sink, 690, 'schedule') == (1, 0) + receipt = conn.execute('SELECT delivered_at FROM outbox').fetchone()[0] + print('Published actual outcome:', publications[0]['outcome']) + print('Terminal receipt:', receipt, '; execution callbacks:', len(calls)) + frozen = json.loads(conn.execute('SELECT delivery_scope FROM outbox').fetchone()[0]) + print('Outbox pinned emitter and recipient:', frozen['emitter'], frozen['recipient']) + finally: + release.set() + sink.close() + conn.close() diff --git a/test_scheduled_event_sink.py b/test_scheduled_event_sink.py new file mode 100644 index 0000000..ec2b4c7 --- /dev/null +++ b/test_scheduled_event_sink.py @@ -0,0 +1,162 @@ +"""Actual observe/outbox delivery into the async receiver; signing is a fixture.""" +import base64 +from dataclasses import asdict, replace +import json +import threading + +import pytest + +from async_scheduled_sink import AsyncScheduledSink +from emitter import Emitter, connect, deliver_pending, observe +from scheduled_event_sink import ScheduledEventSink +from scheduled_receiver import ExecutionScope, run_id +from slot_receiver import SlotLedger + + +@pytest.fixture +def emitted(tmp_path): + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') + definition = Emitter('schedule', ['unused'], 60, 'operator', + event='scheduled', baseline_emits=True) + conn = connect(tmp_path/'emitter.sqlite') + assert observe(conn, definition, [{'item': scope.job, 'verdict': 'UNBLOCKED', + 'event_id': run_id(scope, 600), 'owner': 'operator'}], 0) + event = json.loads(conn.execute('SELECT payload FROM outbox').fetchone()[0]) + yield scope, conn, event + conn.close() + + +class RecordingReceiver: + def __init__(self, scope): + self.scope, self.calls = scope, [] + + def delivery_scope(self): + return {'scope': asdict(self.scope)} + + def deliver(self, binding): + self.calls.append(binding) + + +def encode(value): + raw = json.dumps(value, sort_keys=True, separators=(',', ':')).encode() + return 'scheduled-v1-' + base64.urlsafe_b64encode(raw).decode().rstrip('=') + + +def test_real_emitter_outbox_waits_for_signed_receiver_and_pins_scope(emitted, tmp_path): + scope, conn, event = emitted + entered, release, published = (threading.Event() for _ in range(3)) + calls = [] + + class Port: + def admit(self, scope, slot): pass + def started(self, run, scope): pass + def finished(self, run, scope, outcome): pass + def publish(self, run, scope, outcome): published.set() + + def execute(slot): + calls.append(slot) + entered.set() + assert release.wait(5) + return {'outcome': 'succeeded', 'exit_code': 0} + + inner = AsyncScheduledSink(SlotLedger(tmp_path/'slots.sqlite'), scope, + lambda slot: Port(), execute) + sink = ScheduledEventSink(inner, 'schedule', 'operator') + try: + assert deliver_pending(conn, sink, 0, 'schedule') == (0, 1) + assert entered.wait(5) + frozen, receipt = conn.execute('SELECT delivery_scope,delivered_at FROM outbox').fetchone() + assert json.loads(frozen) == sink.delivery_scope() and receipt is None + assert deliver_pending(conn, sink, 30, 'schedule') == (0, 1) + assert calls == [600] and not published.is_set() + release.set() + with inner._lock: + future = next(iter(inner._pending.values())) + future.result(timeout=5) + assert deliver_pending(conn, sink, 90, 'schedule') == (1, 0) + assert published.is_set() and calls == [600] + assert conn.execute('SELECT delivered_at FROM outbox').fetchone() == (90,) + assert event['event_id'] == run_id(scope, 600) + finally: + release.set() + sink.close() + + +def test_delivery_time_and_generation_cannot_change_frozen_slot(emitted): + scope, _, event = emitted + receiver = RecordingReceiver(scope) + sink = ScheduledEventSink(receiver, 'schedule', 'operator') + event.update(observed_at=9876543210, generation=123) + sink.deliver(event) + assert receiver.calls == [{**asdict(scope), 'scheduled_at': 600}] + + +@pytest.mark.parametrize('field,value', [('event', 'unblocked'), ('emitter', 'other'), + ('item', 'urn:example:foreign'), ('recipient', 'other'), ('command', 'anything'), + ('generation', True), ('generation', -1), ('observed_at', True), + ('observed_at', float('nan')), ('observed_at', float('inf')), + ('evidence', 'x'*8193), ('work_kind', []), ('event_id', 'scheduled-v1-!'), + ('event_id', 'scheduled-v1-'+'a'*24000)]) +def test_misdirected_or_malformed_envelope_never_reaches_receiver(emitted, field, value): + scope, _, event = emitted + receiver = RecordingReceiver(scope) + sink = ScheduledEventSink(receiver, 'schedule', 'operator') + event[field] = value + with pytest.raises(ValueError): + sink.deliver(event) + assert receiver.calls == [] + + +@pytest.mark.parametrize('field', ['job', 'revision', 'host', 'manifest_sha256', 'agent', 'workflow']) +def test_each_foreign_scope_field_refuses_before_receiver(emitted, field): + scope, _, event = emitted + receiver = RecordingReceiver(scope) + other = 'b'*64 if field == 'manifest_sha256' else 'urn:example:foreign' + event['event_id'] = run_id(replace(scope, **{field: other}), 600) + with pytest.raises(ValueError): + ScheduledEventSink(receiver, 'schedule', 'operator').deliver(event) + assert receiver.calls == [] + + +@pytest.mark.parametrize('change', ['duplicate', 'padding', 'version_bool', 'slot_bool', + 'slot_huge', 'extra', 'array', 'missing', 'nested']) +def test_noncanonical_or_unbounded_run_binding_refuses(emitted, change): + scope, _, event = emitted + receiver = RecordingReceiver(scope) + binding = {**asdict(scope), 'scheduled_at': 600, 'version': 1} + if change == 'duplicate': + raw = json.dumps(binding, sort_keys=True, separators=(',', ':')) + raw = raw.replace('"scheduled_at":600', '"scheduled_at":600,"scheduled_at":600') + event['event_id'] = 'scheduled-v1-'+base64.urlsafe_b64encode(raw.encode()).decode().rstrip('=') + elif change == 'padding': + event['event_id'] += '=' + elif change == 'nested': + raw = '['*1500 + '0' + ']'*1500 + event['event_id'] = 'scheduled-v1-'+base64.urlsafe_b64encode(raw.encode()).decode().rstrip('=') + else: + if change == 'version_bool': + binding['version'] = True + elif change == 'slot_bool': + binding['scheduled_at'] = True + elif change == 'slot_huge': + binding['scheduled_at'] = 10**100 + elif change == 'extra': + binding['command'] = 'anything' + elif change == 'array': + binding = [binding] + elif change == 'missing': + del binding['version'] + event['event_id'] = encode(binding) + with pytest.raises(ValueError): + ScheduledEventSink(receiver, 'schedule', 'operator').deliver(event) + assert receiver.calls == [] + + +def test_missing_frozen_fields_are_not_filled_from_current_configuration(emitted): + scope, _, event = emitted + receiver = RecordingReceiver(scope) + del event['event_id'] + with pytest.raises(ValueError): + ScheduledEventSink(receiver, 'schedule', 'operator').deliver(event) + assert receiver.calls == [] From 18bf754ae80f7feb5c2ea96fdae3929c9cf0c155 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Thu, 15 Oct 2026 07:50:07 -0400 Subject: [PATCH 4/7] feat(schedule): project pinned declarations into current UTC slots --- docs/demos/scheduled-projection.log | 10 ++ docs/demos/scheduled-projection.timing | 5 + docs/scheduled-projection.md | 45 ++++++++ pyproject.toml | 2 +- scheduled_projection.py | 106 +++++++++++++++++ scripts/demo-scheduled-projection.py | 52 +++++++++ test_scheduled_projection.py | 154 +++++++++++++++++++++++++ 7 files changed, 373 insertions(+), 1 deletion(-) create mode 100644 docs/demos/scheduled-projection.log create mode 100644 docs/demos/scheduled-projection.timing create mode 100644 docs/scheduled-projection.md create mode 100644 scheduled_projection.py create mode 100644 scripts/demo-scheduled-projection.py create mode 100644 test_scheduled_projection.py diff --git a/docs/demos/scheduled-projection.log b/docs/demos/scheduled-projection.log new file mode 100644 index 0000000..90b1b9a --- /dev/null +++ b/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" ] +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"] diff --git a/docs/demos/scheduled-projection.timing b/docs/demos/scheduled-projection.timing new file mode 100644 index 0000000..519fafe --- /dev/null +++ b/docs/demos/scheduled-projection.timing @@ -0,0 +1,5 @@ +0.032912 41 +0.000014 43 +0.001719 50 +0.070044 67 +0.000009 124 diff --git a/docs/scheduled-projection.md b/docs/scheduled-projection.md new file mode 100644 index 0000000..65680c1 --- /dev/null +++ b/docs/scheduled-projection.md @@ -0,0 +1,45 @@ +# Pinned fixed-interval projection + +`scheduled_projection.ScheduledProjection` consumes the versioned declaration +catalogue produced by the graph reader. Its constructor pins one execution +scope, catalogue graph, manifest target IRI, owner, anchor, interval, timezone +and effective UTC time from trusted local admission. The manifest digest stays +inside `ExecutionScope`; a target IRI is not a command or a digest substitute. + +Evaluation requires the declared identity to be an explicit verified membership +control. A missing, duplicate, malformed or changed declaration raises; it never +means an empty healthy catalogue. The caller must treat failure as UNKNOWN and +retain its durable deadlines, verdicts and delivery outbox. Disabled declarations +remain explicit BLOCKED records. Their next check is null, so change-feed routing +and bounded catalogue reconciliation must observe subsequent enablement. + +The result uses the existing incremental adapter envelope: one scoped record +and its `next_checks` deadline. An already-due eligible slot gets the canonical +signed-run identifier, `UNBLOCKED` and the pinned owner. Configure its emitter +with `event: scheduled`, `baseline_emits: true`, and no generation-based identity +rewriting. Repeated observations deduplicate in the actual emitter. A later +slot advances even while earlier receipts remain pending; success acceptance +and missed-run alerts are separate responsibilities. + +Only the current aligned slot is projected, never historical catch-up work. +The first permitted slot is the anchor-aligned occurrence at or after the +explicit effective time. UTC arithmetic preserves fixed elapsed seconds through +DST; it does not implement local calendar recurrence. The caller supplies a +trusted current clock and must preserve UNKNOWN when that clock is uncertain. +There is no persistent clock watermark or backwards-clock policy in this module. + +This is a library, not a configured adapter process or service. It performs no +query, write, child execution, receipt verification, graph profile installation +or cron change. The configured catalogue transport, change subscription, sink +factory, bounded executor, deadline-success verifier and activation remain +separate integration work. It does not clear the child-survival or runtime gates. + +`python scripts/demo-scheduled-projection.py` demonstrates actual emitter outbox +deduplication and refusal without executing anything. The fixture's runtime +admission fields remain UNKNOWN. `test_scheduled_projection.py` covers activation +boundaries, missing identity retention, pinned scope refusal, disabled state, +time bounds and a real DST transition. + +Replay the output-only [recording](demos/scheduled-projection.log) with its +[timing](demos/scheduled-projection.timing) using `scriptreplay`. On base0b4a0eb5 +the same demo reports the projection absent and its receiver control present. diff --git a/pyproject.toml b/pyproject.toml index 6de0bc1..acbfcf5 100644 --- a/pyproject.toml +++ b/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", - "graph_sink", "incremental", "once", "scheduled_event_sink", "scheduled_receiver", "shuttle_process_port", "slot_receiver", + "graph_sink", "incremental", "once", "scheduled_event_sink", "scheduled_projection", "scheduled_receiver", "shuttle_process_port", "slot_receiver", ] [tool.pytest.ini_options] diff --git a/scheduled_projection.py b/scheduled_projection.py new file mode 100644 index 0000000..8832ca3 --- /dev/null +++ b/scheduled_projection.py @@ -0,0 +1,106 @@ +"""Pure projection of one pinned declaration; no reads, writes or execution. + +The caller supplies a fresh, bounded catalogue from the declaration reader. +Errors must remain UNKNOWN in the caller, preserving its durable queue. This +module neither caches a failed catalogue nor grants executable admission. +""" +from __future__ import annotations + +from dataclasses import dataclass +import math +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from scheduled_receiver import ExecutionScope, run_id + +MAX_TIME = 253402300799 + + +def _text(value): + return type(value) is str and 0 < len(value) <= 2048 + + +@dataclass(frozen=True) +class ScheduledProjection: + scope: ExecutionScope + graph: str + target: str + owner: str + anchor_s: int + interval_s: int + timezone: str + effective_at: int + + def __post_init__(self): + self.scope.validate() + if not all(_text(v) for v in (self.graph, self.target, self.owner, self.timezone)): + raise ValueError('bounded pinned graph, target, owner and timezone required') + if len(self.owner) > 256: + raise ValueError('owner must fit the pinned receiver recipient') + for value in (self.anchor_s, self.effective_at): + if type(value) is not int or not 0 <= value <= MAX_TIME: + raise ValueError('bounded integer UTC anchors required') + if type(self.interval_s) is not int or not 0 < self.interval_s <= MAX_TIME: + raise ValueError('bounded positive fixed interval required') + try: + ZoneInfo(self.timezone) + except (ValueError, ZoneInfoNotFoundError) as exc: + raise ValueError('valid pinned IANA timezone required') from exc + + def evaluate(self, catalogue: dict, now: float) -> dict: + """One current eligible slot, never a catch-up execution list. + + UTC arithmetic intentionally preserves fixed elapsed seconds through + DST. Calendar recurrence is not supported. effective_at is trusted + local activation policy, not a delivery timestamp or graph command. + """ + if (type(now) not in (int, float) or not 0 <= now <= MAX_TIME or + not math.isfinite(now)): + raise ValueError('bounded current UTC time required') + root = {'schema_version', 'source', 'graph', 'membership', 'jobs', + 'executable_admission', 'signed_run_status', 'firing_eligibility'} + if (type(catalogue) is not dict or set(catalogue) != root or + type(catalogue['schema_version']) is not int or + catalogue['schema_version'] != 1 or catalogue['source'] != 'declarations_only' or + catalogue['graph'] != self.graph or any(catalogue[k] != 'UNKNOWN' for k in + ('executable_admission', 'signed_run_status', 'firing_eligibility'))): + raise ValueError('pinned versioned declaration catalogue required') + membership, jobs = catalogue['membership'], catalogue['jobs'] + if (type(membership) is not dict or set(membership) != {'status', 'required'} or + membership['status'] != 'required_identities_verified' or + type(membership['required']) is not list or + not 0 < len(membership['required']) <= 256 or + not all(_text(v) for v in membership['required']) or + type(jobs) is not list or not 0 < len(jobs) <= 256): + raise ValueError('bounded catalogue and positive membership controls required') + required = membership['required'] + identities = [j.get('iri') if type(j) is dict else None for j in jobs] + if (not all(_text(v) for v in identities) or len(set(identities)) != len(identities) or + len(set(required)) != len(required) or self.scope.job not in required or + not set(required) <= set(identities)): + raise ValueError('missing or duplicate required job identity') + job = jobs[identities.index(self.scope.job)] + fields = {'iri', 'name', 'revision', 'owner', 'enabled', 'host', 'target', + 'event', 'schedule', 'interval_s', 'anchor_s', 'timezone'} + pins = {'revision': self.scope.revision, 'host': self.scope.host, + 'target': self.target, 'owner': self.owner, 'interval_s': self.interval_s, + 'anchor_s': self.anchor_s, 'timezone': self.timezone} + if (set(job) != fields or type(job['enabled']) is not bool or + type(job['anchor_s']) is not int or type(job['interval_s']) is not int or + not all(_text(job[k]) for k in ('name', 'event', 'schedule')) or + any(job[k] != v for k, v in pins.items())): + raise ValueError('declaration differs from pinned schedule and execution scope') + first = self.anchor_s + max(0, (self.effective_at - self.anchor_s + + self.interval_s - 1) // self.interval_s) * self.interval_s + record = {'item': self.scope.job, 'owner': self.owner, 'verdict': 'BLOCKED'} + next_check = first if first <= MAX_TIME else None + if not job['enabled']: + next_check = None + elif now >= first: + slot = self.anchor_s + (math.floor(now) - self.anchor_s) // self.interval_s * self.interval_s + record.update(verdict='UNBLOCKED', event_id=run_id(self.scope, slot), + evidence='Current eligible UTC slot; historical slots are not replayed.') + next_check = slot + self.interval_s + if next_check > MAX_TIME: + next_check = None + return {'version': 1, 'scope': [self.scope.job], 'records': [record], + 'next_checks': {self.scope.job: next_check}} diff --git a/scripts/demo-scheduled-projection.py b/scripts/demo-scheduled-projection.py new file mode 100644 index 0000000..987db04 --- /dev/null +++ b/scripts/demo-scheduled-projection.py @@ -0,0 +1,52 @@ +"""Pure pinned projection and actual emitter outbox; no network or execution.""" +import importlib.util +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 emitter import Emitter, connect, observe # noqa: E402 +from scheduled_receiver import ExecutionScope, run_id # noqa: E402 + +scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') +print('Receiver identity positive control:', run_id(scope, 600).startswith('scheduled-v1-')) +present = importlib.util.find_spec('scheduled_projection') is not None +print('Pinned schedule projection available:', present) +if not present: + sys.exit(0) +from scheduled_projection import ScheduledProjection # noqa: E402 + +p = ScheduledProjection(scope, 'urn:example:catalogue', 'urn:example:manifest', + 'operator', 600, 600, 'UTC', 1250) +catalogue = {'schema_version': 1, 'source': 'declarations_only', 'graph': p.graph, + 'executable_admission': 'UNKNOWN', 'signed_run_status': 'UNKNOWN', + 'firing_eligibility': 'UNKNOWN', + 'membership': {'status': 'required_identities_verified', 'required': [scope.job]}, + 'jobs': [{'iri': scope.job, 'name': 'fixture', 'revision': scope.revision, + 'owner': p.owner, 'enabled': True, 'host': scope.host, 'target': p.target, + 'event': 'urn:example:event', 'schedule': 'urn:example:schedule', + 'interval_s': 600, 'anchor_s': 600, 'timezone': 'UTC'}]} +print('Effective 1250; first eligible aligned slot:', p.evaluate(catalogue, 1250)['next_checks'][scope.job]) +with tempfile.TemporaryDirectory() as root: + conn = connect(Path(root)/'outbox.sqlite') + e = Emitter('fixture-schedule', ['unused'], 60, p.owner, + event='scheduled', baseline_emits=True) + try: + for now in (1801, 1802, 2401): + out = p.evaluate(catalogue, now) + observe(conn, e, out['records'], now, scope=set(out['scope'])) + print('Three observations; two current-slot events:', conn.execute( + 'SELECT COUNT(*) FROM outbox').fetchone()[0]) + print('Next deadline:', out['next_checks'][scope.job]) + catalogue['jobs'] = [] + try: + p.evaluate(catalogue, 3001) + except ValueError: + print('Missing required identity: REFUSED; retained outbox rows:', conn.execute( + 'SELECT COUNT(*) FROM outbox').fetchone()[0]) + print('Execution callbacks: 0; no sink or production schedule enabled.') + finally: + conn.close() diff --git a/test_scheduled_projection.py b/test_scheduled_projection.py new file mode 100644 index 0000000..1ad6ddf --- /dev/null +++ b/test_scheduled_projection.py @@ -0,0 +1,154 @@ +from copy import deepcopy +from dataclasses import replace +from datetime import datetime +import json +from zoneinfo import ZoneInfo + +import pytest + +from emitter import Emitter, connect, observe +from scheduled_projection import MAX_TIME, ScheduledProjection +from scheduled_receiver import ExecutionScope, run_id + + +@pytest.fixture +def projection(): + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') + return ScheduledProjection(scope, 'urn:example:catalogue', 'urn:example:manifest', + 'operator', 600, 600, 'UTC', 1250) + + +def catalogue(p): + return {'schema_version': 1, 'source': 'declarations_only', 'graph': p.graph, + 'membership': {'status': 'required_identities_verified', 'required': [p.scope.job]}, + 'executable_admission': 'UNKNOWN', 'signed_run_status': 'UNKNOWN', + 'firing_eligibility': 'UNKNOWN', 'jobs': [{ + 'iri': p.scope.job, 'name': 'fixture', 'revision': p.scope.revision, + 'owner': p.owner, 'enabled': True, 'host': p.scope.host, 'target': p.target, + 'event': 'urn:example:event', 'schedule': 'urn:example:schedule', + 'interval_s': p.interval_s, 'anchor_s': p.anchor_s, 'timezone': p.timezone}]} + + +def test_boundaries_and_no_historical_catchup(projection): + p = projection + for now in (0, 600, 1249, 1250, 1799.999): + out = p.evaluate(catalogue(p), now) + assert out['records'][0]['verdict'] == 'BLOCKED' + assert out['next_checks'] == {p.scope.job: 1800} + for now, slot in ((1800, 1800), (2399.999, 1800), (2400, 2400), (600000, 600000)): + out = p.evaluate(catalogue(p), now) + assert len(out['records']) == 1 + assert out['records'][0]['event_id'] == run_id(p.scope, slot) + assert out['next_checks'] == {p.scope.job: slot+600} + + +def test_actual_emitter_deduplicates_slot_and_advances_without_receipt(tmp_path, projection): + p = projection + conn = connect(tmp_path/'outbox.sqlite') + e = Emitter('scheduled-fixture', ['unused'], 60, p.owner, + event='scheduled', baseline_emits=True) + try: + for now in (1801, 1802, 2401): + out = p.evaluate(catalogue(p), now) + observe(conn, e, out['records'], now, scope=set(out['scope'])) + rows = conn.execute('SELECT event_id,payload,delivered_at FROM outbox ORDER BY created').fetchall() + assert [r[0] for r in rows] == [run_id(p.scope, 1800), run_id(p.scope, 2400)] + assert all(json.loads(r[1])['recipient'] == p.owner and r[2] is None for r in rows) + before = conn.execute('SELECT * FROM verdicts').fetchall() + broken = catalogue(p) + broken['jobs'] = [] + with pytest.raises(ValueError): + out = p.evaluate(broken, 3001) + observe(conn, e, out['records'], 3001, scope=set(out['scope'])) + assert conn.execute('SELECT * FROM verdicts').fetchall() == before + assert conn.execute('SELECT COUNT(*) FROM outbox').fetchone()[0] == 2 + finally: + conn.close() + + +@pytest.mark.parametrize('field,value', [ + ('revision', 'other'), ('host', 'urn:example:foreign'), ('target', 'urn:example:other'), + ('owner', 'other'), ('anchor_s', True), ('anchor_s', 0), ('interval_s', True), + ('interval_s', 60), ('timezone', 'Europe/London'), ('enabled', 1), ('event', ''), + ('schedule', None), ('name', ''), ('iri', 'urn:example:missing'), +]) +def test_changed_declaration_refuses(projection, field, value): + data = catalogue(projection) + data['jobs'][0][field] = value + with pytest.raises(ValueError): + projection.evaluate(data, 1800) + + +@pytest.mark.parametrize('mutation', [ + lambda d: d.update(schema_version=True), lambda d: d.update(source='local_toml'), + lambda d: d.update(graph='urn:example:other'), lambda d: d.update(extra=True), + lambda d: d.update(executable_admission='ADMITTED'), lambda d: d.update(jobs=[]), + lambda d: d.update(jobs=d['jobs']*2), lambda d: d.update(jobs=[None]), + lambda d: d.update(membership={'status': 'UNKNOWN', 'required': []}), + lambda d: d['membership'].update(required=[]), + lambda d: d['membership'].update(required=['urn:example:missing']), + lambda d: d['membership'].update(required=d['membership']['required']*2), + lambda d: d['jobs'][0].update(extra=True), +]) +def test_failed_membership_and_protocol_never_mean_empty_healthy(projection, mutation): + data = deepcopy(catalogue(projection)) + mutation(data) + with pytest.raises(ValueError): + projection.evaluate(data, 1800) + + +def test_disabled_is_explicit_not_missing(projection): + data = catalogue(projection) + data['jobs'][0]['enabled'] = False + out = projection.evaluate(data, 600000) + assert out['records'] == [{'item': projection.scope.job, 'owner': projection.owner, + 'verdict': 'BLOCKED'}] + assert out['next_checks'] == {projection.scope.job: None} + + +def test_fixed_elapsed_intervals_across_dst_and_many_anchors(projection): + for anchor in (0, 17, 1700000000): + for interval in (1, 7, 600, 86400): + p = replace(projection, anchor_s=anchor, interval_s=interval, + effective_at=anchor+interval+1, timezone='America/New_York') + for k in range(2, 20): + now = anchor+k*interval + out = p.evaluate(catalogue(p), now) + assert out['records'][0]['event_id'] == run_id(p.scope, now) + assert out['next_checks'][p.scope.job] == now+interval + + +def test_dst_transition_preserves_elapsed_seconds_not_wall_clock(projection): + zone = ZoneInfo('America/New_York') + anchor = int(datetime(2023, 11, 4, 12, tzinfo=zone).timestamp()) + p = replace(projection, anchor_s=anchor, interval_s=86400, + effective_at=anchor, timezone=zone.key) + slot = anchor+86400 + assert datetime.fromtimestamp(slot, zone).hour == 11 # fall-back day + out = p.evaluate(catalogue(p), slot) + assert out['records'][0]['event_id'] == run_id(p.scope, slot) + assert out['next_checks'][p.scope.job] == slot+86400 + + +@pytest.mark.parametrize('now', [-1, True, float('nan'), float('inf'), MAX_TIME+1, '1800']) +def test_clock_bounds(projection, now): + with pytest.raises(ValueError): + projection.evaluate(catalogue(projection), now) + + +@pytest.mark.parametrize('field,value', [ + ('anchor_s', True), ('anchor_s', -1), ('effective_at', MAX_TIME+1), + ('interval_s', 0), ('interval_s', True), ('interval_s', MAX_TIME+1), + ('timezone', 'Not/AZone'), ('timezone', '/etc/passwd'), ('owner', ''), ('owner', 'x'*257), +]) +def test_pinned_configuration_bounds(projection, field, value): + with pytest.raises(ValueError): + replace(projection, **{field: value}) + + +def test_last_representable_slot_has_no_overflowing_deadline(projection): + p = replace(projection, anchor_s=MAX_TIME, interval_s=600, effective_at=MAX_TIME) + out = p.evaluate(catalogue(p), MAX_TIME) + assert out['records'][0]['event_id'] == run_id(p.scope, MAX_TIME) + assert out['next_checks'] == {p.scope.job: None} From f4a755d9c4344d3eede234eb9429355b3c6dc360 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Thu, 15 Oct 2026 08:38:25 -0400 Subject: [PATCH 5/7] Add bounded fixed-command outcome adapter for scheduled receiver --- docs/demos/fixed-outcome-executor.log | 10 ++ docs/demos/fixed-outcome-executor.timing | 6 + docs/fixed-outcome-executor.md | 39 ++++++ fixed_outcome_executor.py | 159 ++++++++++++++++++++++ pyproject.toml | 2 +- scripts/demo-fixed-outcome-executor.py | 35 +++++ test_fixed_outcome_executor.py | 164 +++++++++++++++++++++++ 7 files changed, 414 insertions(+), 1 deletion(-) create mode 100644 docs/demos/fixed-outcome-executor.log create mode 100644 docs/demos/fixed-outcome-executor.timing create mode 100644 docs/fixed-outcome-executor.md create mode 100644 fixed_outcome_executor.py create mode 100644 scripts/demo-fixed-outcome-executor.py create mode 100644 test_fixed_outcome_executor.py diff --git a/docs/demos/fixed-outcome-executor.log b/docs/demos/fixed-outcome-executor.log new file mode 100644 index 0000000..300e28a --- /dev/null +++ b/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" ] +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"] diff --git a/docs/demos/fixed-outcome-executor.timing b/docs/demos/fixed-outcome-executor.timing new file mode 100644 index 0000000..9775484 --- /dev/null +++ b/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 diff --git a/docs/fixed-outcome-executor.md b/docs/fixed-outcome-executor.md new file mode 100644 index 0000000..46c69ae --- /dev/null +++ b/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. diff --git a/fixed_outcome_executor.py b/fixed_outcome_executor.py new file mode 100644 index 0000000..de6d168 --- /dev/null +++ b/fixed_outcome_executor.py @@ -0,0 +1,159 @@ +"""Fixed trusted command with a bounded, unsigned outcome descriptor. + +Manifest/artifact admission and durable slot claims belong to the signed receiver. +Nothing in a delivery can select argv, environment, directory or a descriptor name. +""" +from __future__ import annotations + +import json +import math +import os +from pathlib import Path +import re +import selectors +import signal +import subprocess +import time +from types import MappingProxyType + +MAX_RESULT = 4096 + + +class ExecutionUnknown(RuntimeError): + """No confirmed outcome; preserve the receiver claim and never rerun it.""" + + +def _pairs(pairs): + out = {} + for key, value in pairs: + if key in out: + raise ValueError('duplicate result field') + out[key] = value + return out + + +def _constant(_): + raise ValueError('non-finite JSON result') + + +def decode_result(raw: bytes, actual_code: int) -> dict: + """Validate the child channel against the actual process exit, not its logs.""" + if (type(raw) is not bytes or not 0 < len(raw) <= MAX_RESULT or + type(actual_code) is not int or not 0 <= actual_code <= 255): + raise ExecutionUnknown('bounded result and actual normal process exit required') + try: + value = json.loads(raw, object_pairs_hook=_pairs, parse_constant=_constant) + required = {'schema_version', 'outcome', 'exit_code'} + if (type(value) is not dict or not required <= value.keys() or + value.keys() - required - {'reason'} or + type(value['schema_version']) is not int or value['schema_version'] != 1 or + type(value['exit_code']) is not int or value['exit_code'] != actual_code or + value['outcome'] not in ('succeeded', 'failed', 'skipped-by-hold') or + (value['outcome'] == 'failed') != (actual_code != 0) or + ('reason' in value and (type(value['reason']) is not str or + not 0 < len(value['reason']) <= 256))): + raise ValueError('inconsistent process result') + return {'outcome': value['outcome'], 'exit_code': actual_code} + except (ValueError, UnicodeError, RecursionError): + raise ExecutionUnknown('invalid or inconsistent child result') from None + + +class FixedOutcomeExecutor: + def __init__(self, argv, *, cwd, env: dict, result_fd_env: str, + timeout_s: float, output_limit: int = 1048576): + if (type(argv) not in (list, tuple) or not 0 < len(argv) <= 64 or + any(type(a) is not str or not a or '\0' in a or len(a) > 8192 for a in argv) or + sum(map(len, argv)) > 131072 or not os.path.isabs(argv[0])): + raise ValueError('bounded fixed argv with an absolute executable required') + cwd = os.fspath(cwd) + if type(cwd) is not str or not os.path.isabs(cwd) or not Path(cwd).is_dir(): + raise ValueError('existing absolute trusted working directory required') + if (type(result_fd_env) is not str or + not re.fullmatch(r'[A-Z][A-Z0-9_]{0,124}_FD', result_fd_env) or + type(env) is not dict or result_fd_env in env or len(env) > 256 or + any(type(k) is not str or not re.fullmatch(r'[A-Za-z_][A-Za-z0-9_]{0,127}', k) or + type(v) is not str or '\0' in v or len(v) > 32768 for k, v in env.items())): + raise ValueError('bounded explicit environment and unused result descriptor name required') + if sum(len(k)+len(v) for k, v in env.items()) > 131072: + raise ValueError('environment exceeds bound') + if (type(timeout_s) not in (int, float) or not 0 < timeout_s <= 86400 or + not math.isfinite(timeout_s) or type(output_limit) is not int or + not 0 < output_limit <= 8388608): + raise ValueError('bounded timeout and output limit required') + self._argv, self._cwd = tuple(argv), cwd + self._env = MappingProxyType(dict(env)) + self._fd_env, self._timeout, self._output_limit = result_fd_env, timeout_s, output_limit + + def __call__(self, scheduled_at: int) -> dict: + if type(scheduled_at) is not int or not 0 <= scheduled_at <= 253402300799: + raise ValueError('bounded integer UTC slot required') + # The slot is identity metadata, never an argument interpolation source. + read_fd, write_fd = os.pipe() + proc = None + selector = selectors.DefaultSelector() + deadline = time.monotonic() + self._timeout + try: + proc = subprocess.Popen(self._argv, cwd=self._cwd, + env={**self._env, self._fd_env: str(write_fd)}, stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=False, + pass_fds=(write_fd,), start_new_session=True) + os.close(write_fd) + write_fd = None + for fd, kind in ((read_fd, 'result'), (proc.stdout.fileno(), 'output')): + os.set_blocking(fd, False) + selector.register(fd, selectors.EVENT_READ, kind) + result, output = bytearray(), 0 + while selector.get_map(): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise ExecutionUnknown('command or inherited descriptors exceeded timeout') + for key, _ in selector.select(min(remaining, 0.1)): + try: + chunk = os.read(key.fd, 65536) + except BlockingIOError: + continue + if not chunk: + selector.unregister(key.fd) + elif key.data == 'result': + result.extend(chunk) + if len(result) > MAX_RESULT: + raise ExecutionUnknown('child result exceeded limit') + else: + output += len(chunk) + if output > self._output_limit: + raise ExecutionUnknown('combined child output exceeded limit') + remaining = deadline - time.monotonic() + if remaining <= 0: + raise ExecutionUnknown('command exceeded timeout') + code = proc.wait(timeout=remaining) + # A normal exit is not completion if ordinary descendants remain. + try: + os.killpg(proc.pid, 0) + except ProcessLookupError: + pass + else: + raise ExecutionUnknown('command left descendants in its process group') + return decode_result(bytes(result), code) + except (OSError, subprocess.SubprocessError): + raise ExecutionUnknown('command invocation or completion unavailable') from None + finally: + selector.close() + os.close(read_fd) + if write_fd is not None: + os.close(write_fd) + if proc is not None: + cleanup_failed = False + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass + except OSError: + cleanup_failed = True + try: + proc.wait(timeout=2) + except subprocess.TimeoutExpired: + cleanup_failed = True + finally: + proc.stdout.close() + if cleanup_failed: + raise ExecutionUnknown('child cleanup unconfirmed') from None diff --git a/pyproject.toml b/pyproject.toml index acbfcf5..9af3df7 100644 --- a/pyproject.toml +++ b/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", - "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", "shuttle_process_port", "slot_receiver", ] [tool.pytest.ini_options] diff --git a/scripts/demo-fixed-outcome-executor.py b/scripts/demo-fixed-outcome-executor.py new file mode 100644 index 0000000..418f1ea --- /dev/null +++ b/scripts/demo-fixed-outcome-executor.py @@ -0,0 +1,35 @@ +"""Local temporary child processes only; no signing, network or production job.""" +import importlib.util +import json +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 scheduled_receiver import ExecutionScope, run_id # noqa: E402 + +scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') +print('Receiver identity positive control:', run_id(scope, 600).startswith('scheduled-v1-')) +present = importlib.util.find_spec('fixed_outcome_executor') is not None +print('Fixed outcome executor available:', present) +if not present: + sys.exit(0) +from fixed_outcome_executor import ExecutionUnknown, FixedOutcomeExecutor # noqa: E402 + +with tempfile.TemporaryDirectory() as root: + for outcome, code in [('succeeded', 0), ('skipped-by-hold', 0), ('failed', 7), (None, 0)]: + packet = json.dumps(dict(schema_version=1, outcome=outcome, exit_code=code)).encode() + program = ('import os,sys; ' + f'os.write(int(os.environ["RESULT_FD"]), {packet!r}); sys.exit({code})') + if outcome is None: + program = 'pass' + call = FixedOutcomeExecutor([sys.executable, '-c', program], cwd=root, env={}, + result_fd_env='RESULT_FD', timeout_s=2) + try: + print('Child:', call(600)) + except ExecutionUnknown: + print('Exit-zero child without result: UNKNOWN; preserve claim, no retry.') +print('No production schedule, signing key, graph or cron was used.') diff --git a/test_fixed_outcome_executor.py b/test_fixed_outcome_executor.py new file mode 100644 index 0000000..d30dfba --- /dev/null +++ b/test_fixed_outcome_executor.py @@ -0,0 +1,164 @@ +from itertools import permutations +import json +import sys +import time + +import pytest + +from fixed_outcome_executor import ExecutionUnknown, FixedOutcomeExecutor, decode_result + + +def packet(outcome='succeeded', code=0): + return dict(schema_version=1, outcome=outcome, exit_code=code) + + +def executor(tmp_path, code, **options): + return FixedOutcomeExecutor([sys.executable, '-c', code], cwd=tmp_path, env={}, + result_fd_env='RESULT_FD', timeout_s=2, **options) + + +def program(value=None, *, raw=None, code=0, prefix=''): + if raw is None: + raw = json.dumps(packet() if value is None else value).encode() + return ('import os,sys; '+prefix+ + f'os.write(int(os.environ["RESULT_FD"]), {raw!r}); sys.exit({code})') + + +def test_result_invariant_exhausts_normal_exit_domain_and_field_order(): + for code in range(256): + for outcome in ('succeeded', 'failed', 'skipped-by-hold'): + value = packet(outcome, code) + for order in permutations(value): + raw = json.dumps({k: value[k] for k in order}).encode() + if (outcome == 'failed') == (code != 0): + assert decode_result(raw, code) == {'outcome': outcome, 'exit_code': code} + else: + with pytest.raises(ExecutionUnknown): + decode_result(raw, code) + + +@pytest.mark.parametrize('raw,code', [ + (b'', 0), (b'{}', 0), (b'[]', 0), (b'null', 0), (b'\xff', 0), + (b'{"schema_version":1,"schema_version":1,"outcome":"succeeded","exit_code":0}', 0), + (json.dumps(packet()).encode()+b'{}', 0), + (json.dumps(dict(packet(), schema_version=True)).encode(), 0), + (json.dumps(dict(packet(), exit_code=False)).encode(), 0), + (json.dumps(dict(packet(), exit_code=0.0)).encode(), 0), + (json.dumps(dict(packet(), reason=None)).encode(), 0), + (json.dumps(dict(packet(), reason='')).encode(), 0), + (json.dumps(dict(packet(), reason='x'*257)).encode(), 0), + (json.dumps(dict(packet(), extra=True)).encode(), 0), + (b'{"schema_version":1,"outcome":"succeeded","exit_code":NaN}', 0), + (json.dumps(packet()).encode(), 1), (json.dumps(packet()).encode(), -9), + (json.dumps(packet()).encode(), True), (b' '*4097, 0), (b'['*1500+b']'*1500, 0), +]) +def test_unavailable_or_inconsistent_result_is_unknown(raw, code): + with pytest.raises(ExecutionUnknown): + decode_result(raw, code) + + +@pytest.mark.parametrize('outcome,code', [('succeeded', 0), ('failed', 7), ('skipped-by-hold', 0)]) +def test_actual_process_exit_and_hold_are_distinct(tmp_path, outcome, code): + value = dict(packet(outcome, code), reason='fixture') + assert executor(tmp_path, program(value, code=code))(600) == {'outcome': outcome, 'exit_code': code} + + +def test_fixed_argv_and_explicit_environment_are_not_delivery_interpolation(tmp_path, monkeypatch): + monkeypatch.setenv('EXECUTOR_FIXTURE_SECRET', 'synthetic-not-a-credential') + literal = ['${HOME}', '$$', 'a b', "q'uote"] + code = ('import os,sys,json; assert "EXECUTOR_FIXTURE_SECRET" not in os.environ; ' + f'assert sys.argv[1:] == {literal!r}; ' + 'os.write(int(os.environ["RESULT_FD"]),json.dumps(' + f'{packet()!r}).encode())') + argv, env = [sys.executable, '-c', code, *literal], {} + call = FixedOutcomeExecutor(argv, cwd=tmp_path, env=env, + result_fd_env='RESULT_FD', timeout_s=2) + argv[:] = ['not-the-command'] + env['EXECUTOR_FIXTURE_SECRET'] = 'also-synthetic' + assert call(600) == {'outcome': 'succeeded', 'exit_code': 0} + + +@pytest.mark.parametrize('code', [ + 'pass', # exit zero without the explicit result is not success + program(packet('failed', 1), code=0), + program(raw=b'not-json'), + program(raw=b'x'*4097), + program(prefix='os.write(1,b"x"*4096); '), + program(prefix='os.write(2,b"x"*4096); '), +]) +def test_real_process_missing_mismatched_and_bounded_channels(tmp_path, code): + with pytest.raises(ExecutionUnknown): + executor(tmp_path, code, output_limit=1024)(600) + + +def test_timeout_stops_ordinary_descendants_before_their_side_effect(tmp_path): + marker = tmp_path/'late-side-effect' + code = ('import os,time; pid=os.fork(); ' + f'(time.sleep(0.7),open({str(marker)!r},"w").write("late")) if pid==0 else time.sleep(5)') + call = FixedOutcomeExecutor([sys.executable, '-c', code], cwd=tmp_path, env={}, + result_fd_env='RESULT_FD', timeout_s=0.15) + started = time.monotonic() + with pytest.raises(ExecutionUnknown): + call(600) + assert time.monotonic()-started < 3 + time.sleep(0.8) + assert not marker.exists() + + +def test_valid_result_cannot_acknowledge_a_still_running_parent(tmp_path): + code = program(prefix='import time; ', code=0).replace('; sys.exit(0)', '; time.sleep(5)') + call = FixedOutcomeExecutor([sys.executable, '-c', code], cwd=tmp_path, env={}, + result_fd_env='RESULT_FD', timeout_s=0.15) + with pytest.raises(ExecutionUnknown): + call(600) + + +@pytest.mark.parametrize('valid', [True, False]) +def test_durable_receiver_never_repeats_actual_child_after_publication_or_channel_failure(tmp_path, valid): + from scheduled_receiver import ExecutionScope, receive + from slot_receiver import SlotLedger, SlotUnknown + from dataclasses import asdict + + counter = tmp_path/'child-calls' + prefix = (f'from pathlib import Path; p=Path({str(counter)!r}); ' + 'p.write_text(str(int(p.read_text())+1) if p.exists() else "1"); ') + raw = json.dumps(packet('skipped-by-hold')).encode() if valid else b'not-json' + call = executor(tmp_path, program(raw=raw, prefix=prefix)) + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') + ledger = SlotLedger(tmp_path/'slots.sqlite') + class Port: + publications = 0 + def admit(self, *_): pass + def started(self, *_): pass + def finished(self, *_): pass + def publish(self, *_): + self.publications += 1 + if self.publications == 1: + raise ConnectionError('synthetic lost publication') + port = Port() + binding = {**asdict(scope), 'scheduled_at': 600} + with pytest.raises(ConnectionError if valid else ExecutionUnknown): + receive(ledger, scope, 600, binding, port, lambda: call(600)) + if valid: + assert receive(ledger, scope, 600, binding, port, lambda: call(600))['outcome'] == 'skipped-by-hold' + assert port.publications == 2 + else: + with pytest.raises(SlotUnknown): + receive(ledger, scope, 600, binding, port, lambda: call(600)) + assert counter.read_text() == '1' + + +@pytest.mark.parametrize('key,value', [ + ('argv', 'not-an-argv'), ('argv', ['relative']), ('argv', [sys.executable, 'a\0b']), + ('env', {'RESULT_FD': '3'}), ('env', {'BAD-KEY': 'x'}), ('env', {'X': 'a\0b'}), + ('result_fd_env', 'PATH'), ('result_fd_env', 'lower_fd'), + ('timeout_s', 0), ('timeout_s', True), ('timeout_s', float('inf')), + ('output_limit', True), ('output_limit', 0), ('output_limit', 8388609), +]) +def test_invalid_config_refuses_before_any_child(tmp_path, key, value): + args = dict(argv=[sys.executable, '-c', 'pass'], cwd=tmp_path, env={}, + result_fd_env='RESULT_FD', timeout_s=1) + args[key] = value + with pytest.raises(ValueError): + FixedOutcomeExecutor(**args) From 305c541c7a9fdf20073a8fce1c9f8722b2670e0d Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Thu, 15 Oct 2026 09:20:06 -0400 Subject: [PATCH 6/7] Wire opt-in scheduled sinks with pinned recipes and owned worker cleanup --- chaski.py | 72 ++++- docs/demos/scheduled-sink-factory.log | 11 + docs/demos/scheduled-sink-factory.timing | 6 + docs/scheduled-sink-factory.md | 55 ++++ pyproject.toml | 2 +- scheduled_sink_factory.py | 102 +++++++ scripts/demo-scheduled-sink-factory.py | 48 ++++ test_scheduled_sink_factory.py | 346 +++++++++++++++++++++++ 8 files changed, 628 insertions(+), 14 deletions(-) create mode 100644 docs/demos/scheduled-sink-factory.log create mode 100644 docs/demos/scheduled-sink-factory.timing create mode 100644 docs/scheduled-sink-factory.md create mode 100644 scheduled_sink_factory.py create mode 100644 scripts/demo-scheduled-sink-factory.py create mode 100644 test_scheduled_sink_factory.py diff --git a/chaski.py b/chaski.py index 3408693..5dce90c 100644 --- a/chaski.py +++ b/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: }} 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() diff --git a/docs/demos/scheduled-sink-factory.log b/docs/demos/scheduled-sink-factory.log new file mode 100644 index 0000000..0ab7a91 --- /dev/null +++ b/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" ] +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"] diff --git a/docs/demos/scheduled-sink-factory.timing b/docs/demos/scheduled-sink-factory.timing new file mode 100644 index 0000000..14a358a --- /dev/null +++ b/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 diff --git a/docs/scheduled-sink-factory.md b/docs/scheduled-sink-factory.md new file mode 100644 index 0000000..ccab63b --- /dev/null +++ b/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. diff --git a/pyproject.toml b/pyproject.toml index 9af3df7..0b18a16 100644 --- a/pyproject.toml +++ b/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] diff --git a/scheduled_sink_factory.py b/scheduled_sink_factory.py new file mode 100644 index 0000000..2fcb510 --- /dev/null +++ b/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 diff --git a/scripts/demo-scheduled-sink-factory.py b/scripts/demo-scheduled-sink-factory.py new file mode 100644 index 0000000..3cc4be3 --- /dev/null +++ b/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() diff --git a/test_scheduled_sink_factory.py b/test_scheduled_sink_factory.py new file mode 100644 index 0000000..1b2b8ba --- /dev/null +++ b/test_scheduled_sink_factory.py @@ -0,0 +1,346 @@ +"""Opt-in factory with actual subprocesses/outbox; Shuttle CLI is a test double.""" + +from dataclasses import asdict, replace +from pathlib import Path +import hashlib +import json +import sys + +import pytest +import yaml + +import chaski +import incremental +from emitter import Emitter, connect, deliver_pending, observe +from scheduled_receiver import run_id +from scheduled_sink_factory import execution_manifest_sha256 +from shuttle_process_port import PortRefused +from test_shuttle_process_port import setup as port_fixture + + +@pytest.fixture +def configured(tmp_path): + _, scope, trace, config, executable = port_fixture.__wrapped__(tmp_path) + # A real child with an owned counter, never the production program. + counter = tmp_path / "child-calls" + program = ( + "import os,json;from pathlib import Path;" + f'p=Path({str(counter)!r});p.write_text(str(int(p.read_text())+1) if p.exists() else "1");' + 'os.write(int(os.environ["RESULT_FD"]),json.dumps(dict(' + 'schema_version=1,outcome="skipped-by-hold",exit_code=0)).encode())' + ) + spec = dict( + scope=asdict(scope), + ledger=str(tmp_path / "slots.sqlite"), + shuttle=dict( + executable=str(executable), + config=str(config), + config_sha256=hashlib.sha256(config.read_bytes()).hexdigest(), + timeout_s=5, + ), + execute=dict( + argv=[sys.executable, "-c", program], + cwd=str(tmp_path), + env={}, + result_fd_env="RESULT_FD", + timeout_s=2, + output_limit=1048576, + ), + workers=1, + capacity=2, + ) + scope = replace(scope, manifest_sha256=execution_manifest_sha256(spec["execute"])) + spec["scope"] = asdict(scope) + definition = dict( + label="fixture", + command=["unused"], + schedule="PT1H", + owner="operator", + trigger="changes", + event="scheduled", + baseline_emits=True, + sink={"scheduled": spec}, + ) + path = tmp_path / "emitters.yml" + + def load(): + path.write_text(yaml.safe_dump({"emitters": [definition]})) + return chaski.load_emitters(path)[0] + + return load, definition, spec, scope, trace, counter, config, executable + + +def test_actual_outbox_factory_waits_for_terminal_process_receipt(configured, tmp_path): + load, _, spec, scope, trace, counter, _, _ = configured + e, d = load() + sink = chaski.make_sink(e, d, "http://unused.invalid") + spec["execute"]["argv"][:] = [sys.executable, "-c", "raise SystemExit(9)"] + conn = connect(tmp_path / "outbox.sqlite") + try: + observe(conn, e, [dict(item=scope.job, owner=e.owner, verdict="UNBLOCKED", event_id=run_id(scope, 600))], 600) + assert deliver_pending(conn, sink, 600, e.label) == (0, 1) + assert conn.execute("SELECT delivered_at FROM outbox").fetchone() == (None,) + with sink.receiver._lock: + future = next(iter(sink.receiver._pending.values())) + future.result(timeout=5) + assert deliver_pending(conn, sink, 1600, e.label) == (1, 0) + assert counter.read_text() == "1" + events = [json.loads(x) for x in trace.read_text().splitlines()] + assert [x["action"] for x in events] == ["admit", "started", "finished", "publish"] + assert events[2]["packet"]["outcome"] == {"outcome": "skipped-by-hold", "exit_code": 0} + assert deliver_pending(conn, sink, 2600, e.label) == (0, 0) + finally: + sink.close() + conn.close() + assert spec["execute"]["env"] == {} # no descriptor or ambient environment leaked into policy + + +def test_lost_publication_receipt_retries_publication_without_child_replay(configured, tmp_path): + load, _, spec, scope, trace, counter, config, executable = configured + config.write_text(json.dumps({"trace": str(trace), "mode": "badpublish"})) + spec["shuttle"]["config_sha256"] = hashlib.sha256(config.read_bytes()).hexdigest() + executable.write_text( + executable.read_text().replace( + "if mode == 'fail': sys.exit(3)", + "if mode == 'fail' or (mode == 'badpublish' and action == 'publish' and " + "sum(json.loads(x)['action']=='publish' for x in " + "Path(config['trace']).read_text().splitlines())==1): sys.exit(3)", + ) + ) + e, d = load() + sink = chaski.make_sink(e, d, "http://unused.invalid") + conn = connect(tmp_path / "outbox.sqlite") + try: + observe(conn, e, [dict(item=scope.job, owner=e.owner, verdict="UNBLOCKED", event_id=run_id(scope, 600))], 600) + assert deliver_pending(conn, sink, 600, e.label) == (0, 1) + with sink.receiver._lock: + future = next(iter(sink.receiver._pending.values())) + with pytest.raises(PortRefused): + future.result(timeout=5) + assert deliver_pending(conn, sink, 1600, e.label) == (0, 1) # consume actual worker error + assert deliver_pending(conn, sink, 2600, e.label) == (0, 1) # publication replay submitted + with sink.receiver._lock: + future = next(iter(sink.receiver._pending.values())) + future.result(timeout=5) + assert deliver_pending(conn, sink, 3600, e.label) == (1, 0) + assert counter.read_text() == "1" + events = [json.loads(x)["action"] for x in trace.read_text().splitlines()] + assert events == ["admit", "started", "finished", "publish", "admit", "publish"] + finally: + sink.close() + conn.close() + + +@pytest.mark.parametrize("change", ["changed", "symlink", "fifo", "writable", "ledger-symlink"]) +def test_local_custody_refuses_before_ledger_or_any_process(configured, tmp_path, change): + import os + + load, _, spec, _, trace, counter, config, _ = configured + e, d = load() + if change == "changed": + config.write_text("{}") + elif change == "writable": + config.chmod(0o666) + elif change == "ledger-symlink": + Path(spec["ledger"]).symlink_to(tmp_path / "not-a-ledger") + else: + config.unlink() + if change == "symlink": + config.symlink_to(tmp_path / "elsewhere") + else: + os.mkfifo(config) + with pytest.raises((ValueError, OSError)): + chaski.make_sink(e, d, "http://unused.invalid") + assert not trace.exists() and not counter.exists() + assert not (tmp_path / "not-a-ledger").exists() + if change != "ledger-symlink": + assert not (tmp_path / "slots.sqlite").exists() + + +@pytest.mark.parametrize( + "field,value", + [ + ("workers", True), + ("workers", 0), + ("capacity", 65), + ("ledger", "relative"), + ("scope", {}), + ("execute", {}), + ("shuttle", {}), + ("extra", 1), + ], +) +def test_bad_configuration_refuses_during_load_without_creating_state(configured, tmp_path, field, value): + load, _, spec, _, trace, counter, _, _ = configured + spec[field] = value + with pytest.raises(chaski.RuleError): + load() + assert not (tmp_path / "slots.sqlite").exists() and not trace.exists() and not counter.exists() + + +@pytest.mark.parametrize( + "field,value", + [ + ("event", "unblocked"), + ("baseline_emits", False), + ("generation_event_ids", True), + ("trigger", "schedule"), + ("label", "invalid label"), + ("key", "foreign"), + ("from", "OK"), + ("to", "BAD"), + ("resolve_recovery", True), + ], +) +def test_incompatible_emitter_policy_refuses(configured, field, value): + load, definition, *_ = configured + definition[field] = value + with pytest.raises(chaski.RuleError): + load() + + +def test_consumer_closes_every_sink_and_database_before_releasing_lease(tmp_path, monkeypatch): + import emitter + from consumer_lease import ConsumerLease, ConsumerLeaseError + + state, inbox = tmp_path / "cursor.json", tmp_path / "emitter.sqlite" + events = [] + + class Sink: + def __init__(self, name): + self.name = name + + def close(self): + with pytest.raises(ConsumerLeaseError): + with ConsumerLease(state, inbox): + pass + events.append(self.name) + + class Conn: + def close(self): + events.append("database") + + monkeypatch.setattr(chaski, "load_rules", lambda _: []) + monkeypatch.setattr( + chaski, + "load_emitters", + lambda _: [(Emitter("one", ["unused"], 60, "operator"), {}), (Emitter("two", ["unused"], 60, "operator"), {})], + ) + monkeypatch.setattr(chaski, "make_sink", lambda e, *_: Sink(e.label)) + monkeypatch.setattr(emitter, "connect", lambda _: Conn()) + + class Runner: + def __init__(self, e, *_): + self.emitter = e + + monkeypatch.setattr(emitter, "Runner", Runner) + monkeypatch.setattr(chaski, "_serve_consumer", lambda *_: events.append("serve") or 0) + assert ( + chaski.main( + [ + "--quipu", + "http://unused.invalid", + "--rules", + str(tmp_path / "rules.yml"), + "--state", + str(state), + "--emitters", + str(tmp_path / "emitters.yml"), + "--emitter-db", + str(inbox), + "--once", + ] + ) + == 0 + ) + assert events == ["serve", "two", "one", "database"] + with ConsumerLease(state, inbox): + pass + + +def test_shutdown_failure_still_drains_other_owned_sinks(): + calls = [] + + class Sink: + def __init__(self, name): + self.name = name + + def close(self): + calls.append(self.name) + if self.name == "two": + raise RuntimeError("synthetic close failure") + + with pytest.raises(RuntimeError, match="synthetic"): + chaski._close_sinks([Sink("one"), Sink("two")]) + assert calls == ["two", "one"] + + +@pytest.mark.parametrize( + "field,value", + [ + ("argv", [sys.executable, "-c", "pass"]), + ("env", {"EXAMPLE": "changed"}), + ("timeout_s", 3), + ("output_limit", 1024), + ], +) +def test_changed_literal_recipe_cannot_keep_old_manifest_digest(configured, tmp_path, field, value): + load, _, spec, _, trace, counter, _, _ = configured + spec["execute"][field] = value + with pytest.raises(chaski.RuleError, match="manifest digest"): + load() + assert not trace.exists() and not counter.exists() and not (tmp_path / "slots.sqlite").exists() + + +def test_partial_startup_failure_closes_constructed_sinks_and_database(tmp_path, monkeypatch): + import emitter + + events = [] + + class Conn: + def close(self): + events.append("database") + + class Sink: + def __init__(self, name): + self.name = name + + def close(self): + events.append(self.name) + + class Runner: + def __init__(self, e, *_): + self.emitter = e + + def failed(*_): + raise RuntimeError("synthetic adapter startup failure") + + monkeypatch.setattr(chaski, "load_rules", lambda _: []) + monkeypatch.setattr( + chaski, + "load_emitters", + lambda _: [ + (Emitter("one", ["unused"], 60, "operator"), {}), + (Emitter("two", ["unused"], 60, "operator", change_driven=True), {}), + ], + ) + monkeypatch.setattr(chaski, "make_sink", lambda e, *_: Sink(e.label)) + monkeypatch.setattr(emitter, "connect", lambda _: Conn()) + monkeypatch.setattr(emitter, "Runner", Runner) + monkeypatch.setattr(incremental, "ChangeRunner", failed) + args = type( + "Args", + (), + dict( + quipu="http://unused.invalid", + rules=tmp_path / "rules.yml", + state=tmp_path / "cursor.json", + event_poll=60, + emitters=tmp_path / "emitters.yml", + emitter_db=tmp_path / "emitter.sqlite", + emitter_write_interval=0, + ), + )() + with pytest.raises(RuntimeError, match="adapter startup"): + chaski.run_consumer(args) + assert events == ["two", "one", "database"] From 4edc2e0b1927ebbc4ae05faf6374c7ba76737026 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Thu, 15 Oct 2026 09:36:18 -0400 Subject: [PATCH 7/7] Adapt pinned ST catalogues to incremental schedule deadlines --- docs/demos/scheduled-catalogue-adapter.log | 13 + docs/demos/scheduled-catalogue-adapter.timing | 9 + docs/scheduled-catalogue-adapter.md | 58 ++++ pyproject.toml | 2 +- scheduled_catalogue_adapter.py | 216 +++++++++++++++ scripts/demo-scheduled-catalogue-adapter.py | 52 ++++ test_scheduled_catalogue_adapter.py | 250 ++++++++++++++++++ 7 files changed, 599 insertions(+), 1 deletion(-) create mode 100644 docs/demos/scheduled-catalogue-adapter.log create mode 100644 docs/demos/scheduled-catalogue-adapter.timing create mode 100644 docs/scheduled-catalogue-adapter.md create mode 100644 scheduled_catalogue_adapter.py create mode 100644 scripts/demo-scheduled-catalogue-adapter.py create mode 100644 test_scheduled_catalogue_adapter.py diff --git a/docs/demos/scheduled-catalogue-adapter.log b/docs/demos/scheduled-catalogue-adapter.log new file mode 100644 index 0000000..b7a7430 --- /dev/null +++ b/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" ] +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"] diff --git a/docs/demos/scheduled-catalogue-adapter.timing b/docs/demos/scheduled-catalogue-adapter.timing new file mode 100644 index 0000000..20bf8b5 --- /dev/null +++ b/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 diff --git a/docs/scheduled-catalogue-adapter.md b/docs/scheduled-catalogue-adapter.md new file mode 100644 index 0000000..69d76e3 --- /dev/null +++ b/docs/scheduled-catalogue-adapter.md @@ -0,0 +1,58 @@ +# Incremental schedule catalogue transport + +`python -m scheduled_catalogue_adapter --config FILE --config-sha256 SHA` +implements the existing `--describe` / `--changes` protocol. The configuration +is an absolute, bounded, owned regular JSON file with no final symlink or group/ +other write permission. Each invocation checks the exact bytes before use. + +Version 1 configuration has exactly `version`, `projection`, and `reader`. +`projection` contains the documented `ScheduledProjection` fields, including +the six-field scope, graph, target, owner, UTC anchor/interval, IANA timezone +and effective activation time. `reader` contains exactly `executable`, `root`, +`env`, `timeout_s`, and `output_limit`. Paths are absolute; environment is explicit +and bounded; timeout is at most 60 seconds and stdout at most 8 MiB. Stderr is +discarded, never copied into an error or event. No caller-supplied reader argv +is accepted: the only constructed operation is: + +```text +ST --root ROOT work jobs --graph GRAPH --expect-job JOB --json list +``` + +This uses the installed typed declaration reader without executing a TOML job, +opening its history, or using graph-provided command text. The executable and +its import tree must still come from the approved immutable deployment; hashing +adapter configuration does not prove their artifact identity. + +`--describe` returns the pinned graph and all declaration/schedule dependencies, +without invoking ST. Its configuration digest participates in the consumer's +subscription identity so a local policy refresh reconciles immediately. +Matching change routing statically queues the single pinned job, without a +reader call. A due evaluation reads and validates the current catalogue and +emits the current eligible slot plus the next deadline. Discovery also validates +the catalogue before returning membership. Idle ticks perform no reader call. + +Input, output, time, JSON keys and JSON numbers are bounded/validated. A missing +identity, nonzero reader exit, timeout, changed scope, malformed result or output +overflow exits 2 with no protocol result. The existing incremental consumer then +retains due work and the previous outbox; it cannot interpret failure as an +empty healthy catalogue. Required membership and runtime admission remain +distinct. The declaration's runtime verdicts must still be explicitly UNKNOWN. + +Set the outer emitter adapter timeout above reader timeout plus two seconds of +cleanup and interpreter startup. An external kill is not confirmed cleanup. +Owned process groups provide ordinary cleanup, not kernel containment; service +cgroup/supervisor custody remains required. This adapter supports fixed elapsed +intervals only. Calendar schedules such as a local Monday wall-clock time need +calendar support before migration. + +Keep an emitter label bound to its admitted job identity. Changing the job key +requires a new label or the separately reviewed state-key migration protocol; +do not erase old deadlines, cursors, claims or pending outbox rows. Configuration +refresh is not authority to reinterpret earlier deliveries. + +`python scripts/demo-scheduled-catalogue-adapter.py` runs the actual incremental +consumer and reader subprocess against a temporary typed-catalogue fixture. +It shows two initial reads, no extra idle reads, one new read on the next slot, +and retained deadline/outbox after deleting the required identity. The fake +reader asserts exact list-only argv. No graph, job execution, signing key, +profile or schedule is activated. The demo reuses development test fixtures. diff --git a/pyproject.toml b/pyproject.toml index 0b18a16..e1059b8 100644 --- a/pyproject.toml +++ b/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", "scheduled_sink_factory", "shuttle_process_port", "slot_receiver", + "fixed_outcome_executor", "graph_sink", "incremental", "once", "scheduled_catalogue_adapter", "scheduled_event_sink", "scheduled_projection", "scheduled_receiver", "scheduled_sink_factory", "shuttle_process_port", "slot_receiver", ] [tool.pytest.ini_options] diff --git a/scheduled_catalogue_adapter.py b/scheduled_catalogue_adapter.py new file mode 100644 index 0000000..41d95ff --- /dev/null +++ b/scheduled_catalogue_adapter.py @@ -0,0 +1,216 @@ +"""Pinned, read-only ST catalogue transport for Chaski's incremental protocol.""" +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import os +from pathlib import Path +import re +import selectors +import signal +import stat +import subprocess +import sys +import time + +from scheduled_projection import MAX_TIME, ScheduledProjection +from scheduled_receiver import ExecutionScope + +SCHEMA = 'https://schema.org/' +QUECHUA = 'https://scbrown.github.io/quechua/ns#' +TYPE = 'http://www.w3.org/1999/02/22-rdf-syntax-ns#type' +ATTRIBUTES = [TYPE, *[SCHEMA+x for x in ('name', 'identifier', 'location', 'target', + 'potentialAction', 'eventSchedule', 'repeatFrequency', 'scheduleTimezone', 'startDate')], + QUECHUA+'sourceKind', QUECHUA+'owner', QUECHUA+'enabled'] + + +def _pairs(pairs): + out = {} + for key, value in pairs: + if key in out: + raise ValueError('duplicate JSON key') + out[key] = value + return out + + +def _constant(_): + raise ValueError('non-finite JSON value') + + +def _float(value): + value = float(value) + if not math.isfinite(value): + raise ValueError('non-finite JSON value') + return value + + +def _decode(raw): + try: + return json.loads(raw, object_pairs_hook=_pairs, parse_constant=_constant, parse_float=_float) + except (ValueError, UnicodeError, RecursionError): + raise ValueError('invalid bounded JSON') from None + + +def _keys(value, keys): + if type(value) is not dict or set(value) != set(keys): + raise ValueError('complete supported adapter object required') + + +def load_config(path, digest): + if (not Path(path).is_absolute() or type(digest) is not str or + not re.fullmatch('[0-9a-f]{64}', digest)): + raise ValueError('absolute pinned configuration required') + fd = os.open(path, 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('configuration custody unavailable') + raw = source.read(65537) + if len(raw) > 65536 or hashlib.sha256(raw).hexdigest() != digest: + raise ValueError('configuration changed') + config = _decode(raw) + _keys(config, ('version', 'projection', 'reader')) + if type(config['version']) is not int or config['version'] != 1: + raise ValueError('adapter configuration version unsupported') + _keys(config['projection'], ('scope', 'graph', 'target', 'owner', 'anchor_s', 'interval_s', + 'timezone', 'effective_at')) + projection = dict(config['projection']) + projection['scope'] = ExecutionScope(**projection['scope']) + pinned = ScheduledProjection(**projection) + reader = config['reader'] + _keys(reader, ('executable', 'root', 'env', 'timeout_s', 'output_limit')) + for key in ('executable', 'root'): + if type(reader[key]) is not str or '\0' in reader[key] or not Path(reader[key]).is_absolute(): + raise ValueError('absolute fixed reader paths required') + if not Path(reader['root']).is_dir(): + raise ValueError('existing reader root required') + env = reader['env'] + if (type(env) is not dict or len(env) > 64 or any( + type(k) is not str or not re.fullmatch(r'[A-Za-z_][A-Za-z0-9_]{0,127}', k) or + type(v) is not str or '\0' in v or len(v) > 8192 for k, v in env.items()) or + sum(len(k)+len(v) for k, v in env.items()) > 65536): + raise ValueError('bounded explicit reader environment required') + timeout, limit = reader['timeout_s'], reader['output_limit'] + if (type(timeout) not in (int, float) or not 0 < timeout <= 60 or not math.isfinite(timeout) or + type(limit) is not int or not 0 < limit <= 8388608): + raise ValueError('bounded reader timeout and output required') + return pinned, reader + + +def read_catalogue(projection, reader): + """The only invocation is ST's explicit graph, required-identity JSON list.""" + argv = [reader['executable'], '--root', reader['root'], 'work', 'jobs', + '--graph', projection.graph, '--expect-job', projection.scope.job, '--json', 'list'] + proc = None + selector = selectors.DefaultSelector() + deadline = time.monotonic()+reader['timeout_s'] + try: + proc = subprocess.Popen(argv, cwd=reader['root'], env=dict(reader['env']), + stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, + start_new_session=True) + os.set_blocking(proc.stdout.fileno(), False) + selector.register(proc.stdout, selectors.EVENT_READ) + result = bytearray() + while selector.get_map(): + remaining = deadline-time.monotonic() + if remaining <= 0: + raise ValueError('catalogue reader timed out') + for key, _ in selector.select(min(remaining, .1)): + try: + chunk = os.read(key.fd, 65536) + except BlockingIOError: + continue + if not chunk: + selector.unregister(key.fileobj) + else: + result.extend(chunk) + if len(result) > reader['output_limit']: + raise ValueError('catalogue reader output exceeded limit') + remaining = deadline-time.monotonic() + if remaining <= 0 or proc.wait(timeout=remaining) != 0: + raise ValueError('catalogue reader did not confirm a complete read') + try: + os.killpg(proc.pid, 0) + except ProcessLookupError: + pass + else: + raise ValueError('catalogue reader left descendants') + return _decode(bytes(result)) + finally: + selector.close() + if proc is not None: + failed = False + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass + except OSError: + failed = True + try: + proc.wait(timeout=2) + except subprocess.TimeoutExpired: + failed = True + finally: + proc.stdout.close() + if failed: + raise ValueError('catalogue reader cleanup unconfirmed') from None + + +def describe(projection, config_sha256): + return dict(version=1, graphs=[projection.graph], attributes=ATTRIBUTES, + types=[SCHEMA+x for x in ('Action', 'Event', 'Schedule')], + configuration_sha256=config_sha256) + + +def changes(projection, reader, request): + if type(request) is not dict: + raise ValueError('incremental request required') + now = request.get('now') + if (type(now) not in (int, float) or not 0 <= now <= MAX_TIME or not math.isfinite(now)): + raise ValueError('bounded trusted current time required') + if set(request) == {'now', 'route'}: + rows = request['route'] + if (type(rows) is not list or not 0 < len(rows) <= 256 or any( + type(r) is not dict or r.get('graph') != projection.graph or + r.get('attribute') not in ATTRIBUTES or type(r.get('entity')) is not str or + not 0 < len(r['entity']) <= 2048 for r in rows)): + raise ValueError('bounded matching graph changes required') + return dict(version=1, items=[projection.scope.job]) # static routing; no read + if set(request) == {'now', 'discover'} and request['discover'] is True: + projection.evaluate(read_catalogue(projection, reader), now) + return dict(version=1, items=[projection.scope.job]) + _keys(request, ('now', 'items', 'changes')) + if request['items'] != [projection.scope.job] or request['changes'] != []: + raise ValueError('only the fixed catalogue key may be evaluated') + return projection.evaluate(read_catalogue(projection, reader), now) + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--config', required=True) + parser.add_argument('--config-sha256', required=True) + mode = parser.add_mutually_exclusive_group(required=True) + mode.add_argument('--describe', action='store_true') + mode.add_argument('--changes', action='store_true') + args = parser.parse_args(argv) + try: + projection, reader = load_config(args.config, args.config_sha256) + if args.describe: + result = describe(projection, args.config_sha256) + else: + raw = sys.stdin.buffer.read(65537) + if len(raw) > 65536: + raise ValueError('incremental request exceeded bound') + result = changes(projection, reader, _decode(raw)) + except (ValueError, TypeError, OSError, subprocess.SubprocessError): + print('catalogue adapter UNKNOWN; no result', file=sys.stderr) + return 2 + print(json.dumps(result, allow_nan=False)) + return 0 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/scripts/demo-scheduled-catalogue-adapter.py b/scripts/demo-scheduled-catalogue-adapter.py new file mode 100644 index 0000000..6c8d606 --- /dev/null +++ b/scripts/demo-scheduled-catalogue-adapter.py @@ -0,0 +1,52 @@ +"""Real incremental consumer/reader process with a temporary catalogue 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 scheduled_projection import ScheduledProjection # noqa: E402 + +print('Projection positive control:', callable(ScheduledProjection.evaluate)) +present = importlib.util.find_spec('scheduled_catalogue_adapter') is not None +print('Incremental catalogue adapter available:', present) +if not present: + sys.exit(0) +from emitter import Emitter, connect # noqa: E402 +from incremental import ChangeRunner # noqa: E402 +from test_scheduled_catalogue_adapter import count, setup # noqa: E402 + +logging.getLogger('chaski.emitter').setLevel(logging.ERROR) +logging.getLogger('chaski.changes').setLevel(logging.ERROR) +with tempfile.TemporaryDirectory() as root: + p, _, _, report, counter, command, _ = setup.__wrapped__(Path(root)) + conn = connect(Path(root)/'outbox.sqlite') + class PendingSink: + def delivery_scope(self): + return {'fixture': 'pending'} + def deliver(self, *_): + raise ConnectionError('synthetic pending receipt') + definition = Emitter('fixture', command(), 3600, p.owner, event='scheduled', + baseline_emits=True, change_driven=True) + runner = ChangeRunner(definition, conn, PendingSink()) + try: + print('Describe reader calls:', count(counter)) + runner.tick(1801) + print('Bootstrap discover/evaluate reader calls:', count(counter)) + runner.tick(1802) + runner.tick(2399) + print('After idle ticks, reader calls:', count(counter)) + runner.tick(2400) + print('After next due slot, reader calls:', count(counter)) + report.write_text(json.dumps({'schema_version': 1, 'jobs': []})) + runner.tick(3000) + print('Missing identity UNKNOWN count:', runner.runs['unknown']) + print('Retained next deadline:', conn.execute('SELECT due FROM adapter_deadlines').fetchone()[0]) + print('Retained queued slots:', conn.execute('SELECT COUNT(*) FROM outbox').fetchone()[0]) + print('Reader is a fixture; no graph, production job, key, profile or cron used.') + finally: + conn.close() diff --git a/test_scheduled_catalogue_adapter.py b/test_scheduled_catalogue_adapter.py new file mode 100644 index 0000000..d622724 --- /dev/null +++ b/test_scheduled_catalogue_adapter.py @@ -0,0 +1,250 @@ +"""Actual adapter/reader processes and durable incremental state; no graph calls.""" + +from dataclasses import asdict +import hashlib +import json +from pathlib import Path +import subprocess +import sys +import time + +import pytest + +import scheduled_catalogue_adapter as adapter +from emitter import Emitter, connect +from incremental import ChangeRunner +from test_scheduled_projection import catalogue, projection as projection_fixture + + +@pytest.fixture +def setup(tmp_path): + projection = projection_fixture.__wrapped__() + report, counter = tmp_path / "catalogue.json", tmp_path / "reader-calls" + report.write_text(json.dumps(catalogue(projection))) + reader = tmp_path / "st-reader-fixture" + reader.write_text( + "#!" + + sys.executable + + "\n" + + """import json,os,sys,time +from pathlib import Path +assert 'ADAPTER_TEST_SENTINEL' not in os.environ +assert sys.argv[1:]==['--root',os.environ['FIXTURE_ROOT'],'work','jobs','--graph', + 'urn:example:catalogue','--expect-job','urn:example:job','--json','list'] +p=Path(os.environ['FIXTURE_COUNTER']);p.write_text(str(int(p.read_text())+1) if p.exists() else '1') +mode=os.environ.get('FIXTURE_MODE','ok') +if mode=='sleep':time.sleep(10) +if mode=='fail':print('SYNTHETIC SECRET ERROR',file=sys.stderr);sys.exit(2) +if mode=='large':os.write(1,b'x'*10000) +else:sys.stdout.buffer.write(Path(os.environ['FIXTURE_REPORT']).read_bytes()) +""" + ) + reader.chmod(0o700) + config = dict( + version=1, + projection=asdict(projection), + reader=dict( + executable=str(reader), + root=str(tmp_path), + env=dict(FIXTURE_ROOT=str(tmp_path), FIXTURE_COUNTER=str(counter), FIXTURE_REPORT=str(report)), + timeout_s=2, + output_limit=1048576, + ), + ) + config_file = tmp_path / "adapter.json" + + def command(): + config_file.write_text(json.dumps(config)) + config_file.chmod(0o600) + return [ + sys.executable, + str(Path(adapter.__file__)), + "--config", + str(config_file), + "--config-sha256", + hashlib.sha256(config_file.read_bytes()).hexdigest(), + ] + + def run(request=None): + argv = command() + (["--describe"] if request is None else ["--changes"]) + raw = json.dumps(request).encode() if not isinstance(request, bytes) else request + return subprocess.run(argv, input=raw if request is not None else None, capture_output=True, timeout=5) + + return projection, config, config_file, report, counter, command, run + + +def count(counter): + return int(counter.read_text()) if counter.exists() else 0 + + +def test_describe_and_static_route_do_not_invoke_reader(setup, monkeypatch): + p, _, _, _, counter, _, run = setup + monkeypatch.setenv("ADAPTER_TEST_SENTINEL", "synthetic-not-a-credential") + r = run() + assert r.returncode == 0 + assert json.loads(r.stdout)["graphs"] == [p.graph] and count(counter) == 0 + r = run(dict(now=1801, route=[dict(graph=p.graph, attribute=adapter.SCHEMA + "name", entity="urn:example:any")])) + assert r.returncode == 0 and json.loads(r.stdout)["items"] == [p.scope.job] and count(counter) == 0 + r = run(dict(now=1801, discover=True)) + assert r.returncode == 0 and count(counter) == 1 + r = run(dict(now=1801, items=[p.scope.job], changes=[])) + assert r.returncode == 0 and count(counter) == 2 + out = json.loads(r.stdout) + assert out["records"][0]["verdict"] == "UNBLOCKED" + assert out["next_checks"] == {p.scope.job: 2400} + + +@pytest.mark.parametrize( + "payload", + [ + {}, + dict(now=True, discover=True), + dict(now=1801, discover=False), + dict(now=1801, items=["urn:example:foreign"], changes=[]), + dict(now=1801, items=["urn:example:job"], changes=[], command=["anything"]), + dict(now=1801, route=[]), + dict(now=1801, route=[dict(graph="urn:foreign", attribute="name", entity="x")]), + b'{"now":1801,"now":1802,"discover":true}', + b'{"now":NaN,"discover":true}', + ( + b'{"now":1801,"route":[{"graph":"urn:example:catalogue",' + b'"attribute":"https://schema.org/name","entity":"urn:example:any","value":1e999}]}' + ), + b"x" * 65537, + ], +) +def test_malformed_or_foreign_request_refuses_without_reader(setup, payload): + *_, counter, _, run = setup + r = run(payload) + assert r.returncode == 2 and not r.stdout and count(counter) == 0 + + +@pytest.mark.parametrize("mode", ["fail", "sleep", "large"]) +def test_reader_failure_timeout_and_output_bound_never_emit_catalogue(setup, mode): + p, config, _, _, counter, _, run = setup + config["reader"]["env"]["FIXTURE_MODE"] = mode + if mode == "sleep": + config["reader"]["timeout_s"] = 0.1 + if mode == "large": + config["reader"]["output_limit"] = 1024 + started = time.monotonic() + r = run(dict(now=1801, discover=True)) + assert r.returncode == 2 and not r.stdout and count(counter) == 1 + assert b"SYNTHETIC SECRET" not in r.stderr and time.monotonic() - started < 3 + + +@pytest.mark.parametrize( + "body", [b"{}", b"[]", b"not-json", b'{"schema_version":1,"schema_version":1}', b'{"schema_version":NaN}'] +) +def test_malformed_reader_result_is_unknown(setup, body): + _, _, _, report, counter, _, run = setup + report.write_bytes(body) + r = run(dict(now=1801, discover=True)) + assert r.returncode == 2 and not r.stdout and count(counter) == 1 + + +@pytest.mark.parametrize("change", ["changed", "writable", "symlink", "fifo"]) +def test_pinned_config_custody_refuses_before_reader(setup, tmp_path, change): + import os + + _, _, config_file, _, counter, command, _ = setup + argv = command() + ["--describe"] + if change == "changed": + config_file.write_text("{}") + elif change == "writable": + config_file.chmod(0o666) + else: + config_file.unlink() + if change == "symlink": + config_file.symlink_to(tmp_path / "missing") + else: + os.mkfifo(config_file) + r = subprocess.run(argv, capture_output=True, timeout=3) + assert r.returncode == 2 and not r.stdout and count(counter) == 0 + + +@pytest.mark.parametrize("key,value", [("version", True), ("projection", {}), ("reader", {}), ("extra", 1)]) +def test_unsupported_configuration_is_unknown(setup, key, value): + _, config, _, _, counter, _, run = setup + config[key] = value + r = run() + assert r.returncode == 2 and not r.stdout and count(counter) == 0 + + +def test_actual_incremental_idle_deadline_and_missing_identity_retention(setup, tmp_path): + p, _, _, report, counter, command, _ = setup + definition = Emitter( + "fixture", command(), 3600, p.owner, event="scheduled", baseline_emits=True, change_driven=True + ) + conn = connect(tmp_path / "outbox.sqlite") + + class PendingSink: + def delivery_scope(self): + return {"fixture": "unacknowledged"} + + def deliver(self, *_): + raise ConnectionError("synthetic pending receipt") + + runner = ChangeRunner(definition, conn, PendingSink()) + try: + assert count(counter) == 0 + runner.tick(1801) + assert count(counter) == 2 and conn.execute("select count(*) from outbox").fetchone() == (1,) + runner.tick(1802) + runner.tick(2399) + assert count(counter) == 2 + runner.tick(2400) + assert count(counter) == 3 and conn.execute("select count(*) from outbox").fetchone() == (2,) + before = conn.execute("select * from verdicts").fetchall() + broken = catalogue(p) + broken["jobs"] = [] + report.write_text(json.dumps(broken)) + runner.tick(3000) + assert count(counter) == 4 and conn.execute("select count(*) from outbox").fetchone() == (2,) + assert conn.execute("select * from verdicts").fetchall() == before + assert conn.execute("select due from adapter_deadlines").fetchone() == (3000,) + runner.tick(3001) + assert count(counter) == 4 + report.write_text(json.dumps(catalogue(p))) + runner.tick(3060) + assert count(counter) == 5 and conn.execute("select count(*) from outbox").fetchone() == (3,) + assert conn.execute("select due from adapter_deadlines").fetchone() == (3600,) + finally: + conn.close() + + +def test_local_configuration_change_reconciles_without_graph_subscription_change(setup, tmp_path): + p, config, _, _, counter, command, _ = setup + conn = connect(tmp_path / "outbox.sqlite") + + class PendingSink: + def delivery_scope(self): + return {"fixture": "pending"} + + def deliver(self, *_): + raise ConnectionError("synthetic pending receipt") + + try: + first = Emitter("fixture", command(), 3600, p.owner, event="scheduled", baseline_emits=True, change_driven=True) + runner = ChangeRunner(first, conn, PendingSink()) + runner.tick(1801) + assert count(counter) == 2 + config["reader"]["timeout_s"] = 3 + updated = Emitter( + "fixture", command(), 3600, p.owner, event="scheduled", baseline_emits=True, change_driven=True + ) + replacement = ChangeRunner(updated, conn, PendingSink()) + assert replacement.subscription["graphs"] == runner.subscription["graphs"] + assert replacement.subscription["configuration_sha256"] != runner.subscription["configuration_sha256"] + assert conn.execute("SELECT last_success FROM adapter_reconcile").fetchone() is None + replacement.tick(1802) + assert count(counter) == 4 and conn.execute("SELECT COUNT(*) FROM outbox").fetchone() == (1,) + finally: + conn.close() + + +def test_projection_configuration_requires_an_object_not_pairs(setup): + _, config, _, _, counter, _, run = setup + config["projection"] = list(config["projection"].items()) + result = run() + assert result.returncode == 2 and not result.stdout and count(counter) == 0