Bridge admitted scheduled slots to signed-run ports

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" <not executed on terminal>]
+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"]

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

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.

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]

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

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

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)