feat(receiver): bridge signed phases through the fixed Shuttle CLI

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

docs/demos/shuttle-process.timing

@@ -0,0 +1,2 @@
+0.244914 175
+0.000139 53

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 <phase>`, 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.

pyproject.toml

@@ -33,7 +33,7 @@ chaski-state-key-migrate = "state_key_migration: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", "state_key_migration",
+    "graph_sink", "incremental", "once", "scheduled_receiver", "shuttle_process_port", "slot_receiver", "state_key_migration",
 ]
 
 [tool.pytest.ini_options]

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')

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)

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'