Project pinned schedule slots and validate fixed child outcomes

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

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

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

docs/demos/scheduled-projection.timing

@@ -0,0 +1,5 @@
+0.032912 41
+0.000014 43
+0.001719 50
+0.070044 67
+0.000009 124

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.

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.

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

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