Deliver signed scheduled work asynchronously with pinned emitter envelopes

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)

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

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

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

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

docs/scheduled-receiver.md

@@ -1,5 +1,77 @@
 # 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.
+
+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.

pyproject.toml

@@ -30,9 +30,9 @@ 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",
+    "graph_sink", "incremental", "once", "scheduled_event_sink", "scheduled_receiver", "shuttle_process_port", "slot_receiver",
 ]
 
 [tool.pytest.ini_options]

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

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