From 82d600461b2d516071b127b749dc48275eb779b1 Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Thu, 15 Oct 2026 03:47:09 -0400 Subject: [PATCH] feat: bridge admitted scheduled slots to signed-run ports --- docs/demos/scheduled-receiver.log | 14 +++ docs/demos/scheduled-receiver.timing | 8 ++ docs/scheduled-receiver.md | 53 ++++++++++ pyproject.toml | 2 +- scheduled_receiver.py | 108 ++++++++++++++++++++ scripts/demo-scheduled-receiver.py | 46 +++++++++ test_scheduled_receiver.py | 145 +++++++++++++++++++++++++++ 7 files changed, 375 insertions(+), 1 deletion(-) create mode 100644 docs/demos/scheduled-receiver.log create mode 100644 docs/demos/scheduled-receiver.timing create mode 100644 docs/scheduled-receiver.md create mode 100644 scheduled_receiver.py create mode 100644 scripts/demo-scheduled-receiver.py create mode 100644 test_scheduled_receiver.py diff --git a/docs/demos/scheduled-receiver.log b/docs/demos/scheduled-receiver.log new file mode 100644 index 0000000..9e66678 --- /dev/null +++ b/docs/demos/scheduled-receiver.log @@ -0,0 +1,14 @@ +Script started on 2026-10-15 03:46:33-04:00 [COMMAND="stty -onlcr; python scripts/demo-scheduled-receiver.py" ] +Library demonstration; no real key, signature, process or graph write. +First delivery: +fixture admission control passed +fixture signed-start persistence callback +SYNTHETIC CHILD CALLBACK +fixture signed-terminal persistence callback: skipped-by-hold +fixture export/read-back callback +Identical redelivery: +fixture admission control passed +fixture export/read-back callback +{"child_callbacks": 1, "same_result": true, "outcome": "skipped-by-hold"} + +Script done on 2026-10-15 03:46:33-04:00 [COMMAND_EXIT_CODE="0"] diff --git a/docs/demos/scheduled-receiver.timing b/docs/demos/scheduled-receiver.timing new file mode 100644 index 0000000..7ffc0d3 --- /dev/null +++ b/docs/demos/scheduled-receiver.timing @@ -0,0 +1,8 @@ +0.045559 87 +0.000062 33 +0.009478 67 +0.000006 62 +0.009869 56 +0.000054 33 +0.000148 34 +0.000004 74 diff --git a/docs/scheduled-receiver.md b/docs/scheduled-receiver.md new file mode 100644 index 0000000..319eb3c --- /dev/null +++ b/docs/scheduled-receiver.md @@ -0,0 +1,53 @@ +# Signed scheduled delivery bridge + +`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. + +The port checks host, immutable manifest, declaration eligibility, existing +local key and registered workflow scope. Only after admission does the ledger +commit a unique slot claim. The port must durably persist signed `started` +before execution, then the actual signed terminal outcome. Success, failure +and hold are distinct outcomes; contradictory exit codes refuse. A crash or +exception after claiming leaves UNKNOWN and never authorizes a second child. + +The complete job, revision, host, manifest digest, agent, workflow and scheduled +UTC slot are encoded reversibly into the run identifier covered by the +signature. Delivery must repeat that exact binding, including value types. +This avoids relying on an unsigned hash lookup to interpret signed scope. + +The terminal result is committed before publication. If publication or its +acknowledgement fails, redelivery rechecks admission and retries publication +only. It never calls the child or signs a second execution. The ledger must +survive rollback. Changed manifest/scope for a claimed slot is a conflict. + +A concrete port must export and independently read back the exact signed +started and terminal records against the admitted registered public key. A +successful callback return or local ledger result alone is not governed +acceptance. Registered-key integrity does not establish an isolated human or +agent identity on a shared Unix account. Transport authentication, manifest +compilation, schedule projection, maintenance policy and deadline monitoring +remain separate integration requirements. No port or runtime activation is +provided by this library module. + +Tests use an explicitly synthetic port double to check sequencing, scope +refusal, crash/exception uncertainty and publication-only replay. They do not +claim cryptographic or live graph verification. A concrete adapter needs its +own actual-signature and export/read-back tests before it may execute a job. + +The local receipt's exit code is metadata. Shuttle's current signature format +covers the transition's state and step, not arbitrary receipt payload. A port +must not describe an unsigned exit-code field as cryptographically bound. + +Demo: [recording](demos/scheduled-receiver.log) and +[timing](demos/scheduled-receiver.timing), reproduced with +`python scripts/demo-scheduled-receiver.py`. On base707e0d9 this bridge module +is absent; its existing ledger control already suppresses a duplicate callback. +The proposal adds the displayed started/terminal/publication ordering and a +reversible run scope. The synthetic port is explicitly a double: no real keys, +signatures, child processes or graph writes occur. The one child callback on +two deliveries is observable; duplicate delivery invokes only admission and +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. diff --git a/pyproject.toml b/pyproject.toml index cd63e26..4d639aa 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", "slot_receiver", + "graph_sink", "incremental", "once", "scheduled_receiver", "slot_receiver", ] [tool.pytest.ini_options] diff --git a/scheduled_receiver.py b/scheduled_receiver.py new file mode 100644 index 0000000..9ec4a01 --- /dev/null +++ b/scheduled_receiver.py @@ -0,0 +1,108 @@ +"""Signed-run bridge for an already admitted scheduled delivery. + +The trusted port must persist and verify signed records; this module never +interprets graph-supplied commands or claims that a callback is authentication. +No scheduler, listener or automatic retry of child execution is started here. +""" +from __future__ import annotations + +import base64 +import json +from dataclasses import asdict, dataclass +from typing import Callable, Protocol + +from slot_receiver import SlotLedger, slot_id + + +@dataclass(frozen=True) +class ExecutionScope: + """Immutable binding supplied by a trusted manifest admission layer.""" + job: str + revision: str + host: str + manifest_sha256: str + agent: str + workflow: str + + def validate(self): + if any(type(value) is not str or not value or len(value) > 2048 + for value in asdict(self).values()): + raise ValueError("scope requires bounded nonempty string fields") + if (len(self.manifest_sha256) != 64 or + any(c not in "0123456789abcdef" for c in self.manifest_sha256)): + raise ValueError("manifest digest must be lowercase SHA256") + + +def run_id(scope: ExecutionScope, scheduled_at: int) -> str: + """Reversible scope inside the identifier covered by Shuttle's signature. + + A hash alone would leave the verifier dependent on an unsigned lookup. + Base64url avoids both the canonical signature delimiter and IRI escaping. + """ + scope.validate() + slot_id(scope.job, scope.revision, scheduled_at) # validates UTC slot type + binding = {**asdict(scope), "scheduled_at": scheduled_at, "version": 1} + raw = json.dumps(binding, sort_keys=True, separators=(",", ":")).encode() + return "scheduled-v1-" + base64.urlsafe_b64encode(raw).decode().rstrip("=") + + +class SignedRunPort(Protocol): + """Trusted adapter contract, not data supplied by an incoming delivery. + + admit must verify the local host, current manifest, eligible declaration, + registered workflow/key scope and existing private-key custody. started + must durably persist a signed queued->running record BEFORE returning. + finished must durably persist the actual signed running->terminal record. + publish must export and independently read back those exact records using + the registered key. An exception is UNKNOWN, never successful acceptance. + All methods must be safe to call for the deterministic run identifier. + """ + def admit(self, scope: ExecutionScope, scheduled_at: int) -> None: ... + def started(self, run: str, scope: ExecutionScope) -> None: ... + def finished(self, run: str, scope: ExecutionScope, outcome: dict) -> None: ... + def publish(self, run: str, scope: ExecutionScope, outcome: dict) -> None: ... + + +def _outcome(value: dict) -> dict: + if not isinstance(value, dict) or set(value) != {"outcome", "exit_code"}: + raise ValueError("execution needs exactly outcome and exit_code") + name, code = value["outcome"], value["exit_code"] + if type(code) is not int or not 0 <= code <= 255: + raise ValueError("exit_code must be an actual bounded process result") + if name not in ("succeeded", "failed", "skipped-by-hold"): + raise ValueError("unknown execution outcome") + if (name == "failed") != (code != 0): + raise ValueError("outcome and process exit code disagree") + return {"outcome": name, "exit_code": code} + + +def receive(ledger: SlotLedger, scope: ExecutionScope, scheduled_at: int, + delivery: dict, port: SignedRunPort, execute: Callable[[], dict]) -> dict: + """Execute at most once, then retry only publication on redelivery. + + Delivery must repeat the complete admitted scope and slot exactly. The + ledger persists before started, and the terminal result persists before + publish. A crash or terminal-signing failure leaves the claim UNKNOWN; + it never authorizes rerunning the child. Admission is checked even for a + duplicate; a revoked identity cannot use replay to obtain acceptance. + """ + run = run_id(scope, scheduled_at) + binding = {**asdict(scope), "scheduled_at": scheduled_at} + if (not isinstance(delivery, dict) or + json.dumps(delivery, sort_keys=True, allow_nan=False) != + json.dumps(binding, sort_keys=True, allow_nan=False)): + raise ValueError("delivery differs from the admitted scope or slot") + port.admit(scope, scheduled_at) + + def action(): + port.started(run, scope) + outcome = _outcome(execute()) + port.finished(run, scope, dict(outcome)) + return {"run": run, **outcome} + + result = ledger.run(slot_id(scope.job, scope.revision, scheduled_at), binding, action) + outcome = _outcome({k: result[k] for k in ("outcome", "exit_code")}) + if result.get("run") != run: + raise ValueError("stored run identity does not match the admitted scope") + port.publish(run, scope, outcome) + return result diff --git a/scripts/demo-scheduled-receiver.py b/scripts/demo-scheduled-receiver.py new file mode 100644 index 0000000..1f60d48 --- /dev/null +++ b/scripts/demo-scheduled-receiver.py @@ -0,0 +1,46 @@ +"""Synthetic port double: demonstrate sequencing and publication-only replay.""" +from dataclasses import asdict +import json +from pathlib import Path +import sys +import tempfile + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from scheduled_receiver import ExecutionScope, receive # noqa: E402 +from slot_receiver import SlotLedger # noqa: E402 + + +class Port: + def admit(self, scope, slot): + print('fixture admission control passed') + + def started(self, run, scope): + print('fixture signed-start persistence callback') + + def finished(self, run, scope, outcome): + print('fixture signed-terminal persistence callback:', outcome['outcome']) + + def publish(self, run, scope, outcome): + print('fixture export/read-back callback') + + +with tempfile.TemporaryDirectory() as directory: + ledger = SlotLedger(Path(directory) / 'slots.db') + scope = ExecutionScope('urn:example:job', 'revision-1', 'host-a', 'a'*64, + 'fixture-agent', 'urn:example:workflow') + delivery = {**asdict(scope), 'scheduled_at': 120} + executions = [] + + def execute(): + executions.append(1) + print('SYNTHETIC CHILD CALLBACK') + return {'outcome': 'skipped-by-hold', 'exit_code': 0} + + print('Library demonstration; no real key, signature, process or graph write.') + print('First delivery:') + first = receive(ledger, scope, 120, delivery, Port(), execute) + print('Identical redelivery:') + second = receive(ledger, scope, 120, delivery, Port(), execute) + print(json.dumps({'child_callbacks': len(executions), 'same_result': first == second, + 'outcome': second['outcome']})) + assert len(executions) == 1 and first == second diff --git a/test_scheduled_receiver.py b/test_scheduled_receiver.py new file mode 100644 index 0000000..60530a0 --- /dev/null +++ b/test_scheduled_receiver.py @@ -0,0 +1,145 @@ +"""Bridge sequencing and replay controls; the signing port is a test double.""" +import base64 +from dataclasses import asdict +import json +import tempfile +from pathlib import Path +import unittest + +from scheduled_receiver import ExecutionScope, receive, run_id +from slot_receiver import SlotLedger, SlotUnknown + + +class Port: + def __init__(self): + self.calls = [] + self.fail = None + + def record(self, name): + self.calls.append(name) + if self.fail == name: + raise RuntimeError(name) + + def admit(self, scope, slot): + self.record('admit') + + def started(self, run, scope): + self.record('started') + + def finished(self, run, scope, outcome): + self.record('finished:' + outcome['outcome']) + + def publish(self, run, scope, outcome): + self.record('publish') + + +class ReceiverTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.ledger = SlotLedger(Path(self.temp.name) / 'ledger.db') + self.scope = ExecutionScope('urn:example:job', 'rev1', 'host-a', 'a'*64, + 'fixture', 'urn:example:workflow') + self.slot = 120 + self.delivery = {**asdict(self.scope), 'scheduled_at': self.slot} + self.port = Port() + self.executions = 0 + + def execute(self): + self.port.record('execute') + self.executions += 1 + return {'outcome': 'succeeded', 'exit_code': 0} + + def receive(self, execute=None): + return receive(self.ledger, self.scope, self.slot, self.delivery, + self.port, execute or self.execute) + + def test_order_and_duplicate_publication_without_duplicate_execution(self): + first = self.receive() + self.assertEqual(['admit', 'started', 'execute', 'finished:succeeded', 'publish'], self.port.calls) + self.assertEqual(first, self.receive()) + self.assertEqual(1, self.executions) + self.assertEqual(['admit', 'publish'], self.port.calls[-2:]) + + def test_lost_publication_reply_retries_only_publication(self): + self.port.fail = 'publish' + with self.assertRaises(RuntimeError): + self.receive() + self.port.fail = None + self.assertEqual('succeeded', self.receive()['outcome']) + self.assertEqual(1, self.executions) + self.assertEqual(1, self.port.calls.count('started')) + + def test_failed_started_cannot_execute_or_automatically_retry(self): + self.port.fail = 'started' + with self.assertRaises(RuntimeError): + self.receive() + self.port.fail = None + with self.assertRaises(SlotUnknown): + self.receive() + self.assertEqual(0, self.executions) + + def test_failed_terminal_signing_does_not_repeat_child(self): + self.port.fail = 'finished:succeeded' + with self.assertRaises(RuntimeError): + self.receive() + self.port.fail = None + with self.assertRaises(SlotUnknown): + self.receive() + self.assertEqual(1, self.executions) + self.assertNotIn('publish', self.port.calls) + + def test_admission_refusal_precedes_claim_and_work(self): + self.port.fail = 'admit' + with self.assertRaises(RuntimeError): + self.receive() + self.assertEqual(['admit'], self.port.calls) + self.port.fail = None + self.receive() # refusal never created a claim + self.assertEqual(1, self.executions) + + def test_every_delivery_binding_field_is_checked_before_admission(self): + for field in self.delivery: + with self.subTest(field=field): + delivery = dict(self.delivery) + delivery[field] = 'wrong' + with self.assertRaises(ValueError): + receive(self.ledger, self.scope, self.slot, delivery, self.port, self.execute) + self.assertEqual([], self.port.calls) + with self.assertRaises(ValueError): + receive(self.ledger, self.scope, 1, {**asdict(self.scope), 'scheduled_at': True}, + self.port, self.execute) + + def test_scope_is_recoverable_from_signed_run_name(self): + run = run_id(self.scope, self.slot) + encoded = run.removeprefix('scheduled-v1-') + raw = base64.urlsafe_b64decode(encoded + '='*(-len(encoded) % 4)) + self.assertEqual({**self.delivery, 'version': 1}, json.loads(raw)) + self.assertNotIn('|', run) + self.assertNotEqual(run, run_id(self.scope, self.slot + 60)) + + def test_actual_hold_and_failure_are_distinct_terminal_results(self): + for name, code in [('skipped-by-hold', 0), ('failed', 7)]: + with self.subTest(name=name): + self.slot += 60 + self.delivery['scheduled_at'] = self.slot + result = self.receive(lambda: {'outcome': name, 'exit_code': code}) + self.assertEqual(name, result['outcome']) + self.assertIn('finished:' + name, self.port.calls) + + def test_invalid_child_result_stays_unknown_without_publication(self): + with self.assertRaises(ValueError): + self.receive(lambda: {'outcome': 'succeeded', 'exit_code': 7}) + with self.assertRaises(SlotUnknown): + self.receive() + self.assertNotIn('publish', self.port.calls) + + def test_child_exception_never_authorizes_retry(self): + def fail(): + self.executions += 1 + raise RuntimeError('child outcome unknown') + with self.assertRaises(RuntimeError): + self.receive(fail) + with self.assertRaises(SlotUnknown): + self.receive() + self.assertEqual(1, self.executions)