From 4a59b234b9d3b3bef1b4b9c6382a0543066f6ab9 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 4 Oct 2026 22:36:05 +0200 Subject: [PATCH] Implement private State Hub decision dispatch with durable replay Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f --- .forgejo/workflows/memo-dispatch-image.yaml | 62 ++++++ Containerfile.memo-dispatch | 13 ++ api/services/informed_decision_contract.py | 13 +- api/services/memo_dispatch.py | 174 +++++++++++++++ api/services/memo_forwarding.py | 15 +- deploy/informed-decision/README.md | 42 ++++ .../dispatcher.candidate.json | 207 ++++++++++++++++++ .../2026-10-04-memo-dispatch/container.json | 16 ++ scripts/smoke_memo_dispatch.py | 63 ++++++ tests/test_memo_dispatch.py | 126 +++++++++++ ...STATE-WP-0096-private-decision-dispatch.md | 78 +++++++ 11 files changed, 801 insertions(+), 8 deletions(-) create mode 100644 .forgejo/workflows/memo-dispatch-image.yaml create mode 100644 Containerfile.memo-dispatch create mode 100644 api/services/memo_dispatch.py create mode 100644 deploy/informed-decision/README.md create mode 100644 deploy/informed-decision/dispatcher.candidate.json create mode 100644 docs/evidence/2026-10-04-memo-dispatch/container.json create mode 100644 scripts/smoke_memo_dispatch.py create mode 100644 tests/test_memo_dispatch.py create mode 100644 workplans/STATE-WP-0096-private-decision-dispatch.md diff --git a/.forgejo/workflows/memo-dispatch-image.yaml b/.forgejo/workflows/memo-dispatch-image.yaml new file mode 100644 index 0000000..76f43ed --- /dev/null +++ b/.forgejo/workflows/memo-dispatch-image.yaml @@ -0,0 +1,62 @@ +name: Build and Publish Private Memo Dispatcher + +# Modelled on activity-core/.forgejo/workflows/image.yaml -- the fleet's +# canonical image-publish pattern. Images are built by CI from a tarball of +# the pushed commit, never from a workstation working tree, so the artifact's +# provenance is a forge revision. +# +# `examples/**` is a build-trigger path on purpose: the policy packages are +# COPYed into the image (see Containerfile) and there is no hot reload, so a +# policy change is a new image. + +on: + push: + branches: + - main + paths: + - ".forgejo/workflows/memo-dispatch-image.yaml" + - "Containerfile.memo-dispatch" + - "api/services/informed_decision_contract.py" + - "api/services/memo_forwarding.py" + - "api/services/memo_dispatch.py" + workflow_dispatch: + +env: + REGISTRY: forgejo.coulomb.social + IMAGE_NAME: coulomb/state-hub-memo-dispatch + DOCKER_HOST: tcp://127.0.0.1:2375 + +jobs: + build-and-push: + runs-on: container-build + steps: + - name: Build and push image + env: + REGISTRY_USER: ${{ secrets.REGISTRY_USER }} + REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }} + run: | + set -eu + REF="${GITHUB_SHA:-main}" + SHORT="${REF:0:7}" + mkdir -p buildctx "${HOME}/bin" + wget -qO /tmp/repo.tar.gz \ + "https://forgejo.coulomb.social/${GITHUB_REPOSITORY}/archive/${SHORT}.tar.gz" + tar xzf /tmp/repo.tar.gz -C buildctx --strip-components=1 + wget -qO- https://download.docker.com/linux/static/stable/x86_64/docker-27.3.1.tgz \ + | tar xz --strip-components=1 -C "${HOME}/bin" docker/docker + export PATH="${HOME}/bin:${PATH}" + echo "${REGISTRY_TOKEN}" | docker login "${REGISTRY}" -u "${REGISTRY_USER}" --password-stdin + IMAGE="${REGISTRY}/${IMAGE_NAME}" + docker build -f buildctx/Containerfile.memo-dispatch -t "${IMAGE}:latest" -t "${IMAGE}:main-${SHORT}" buildctx + docker push "${IMAGE}:latest" + docker push "${IMAGE}:main-${SHORT}" + echo "pushed ${IMAGE}:latest and ${IMAGE}:main-${SHORT}" + + - name: Report immutable digest + run: | + set -eu + export PATH="${HOME}/bin:${PATH}" + IMAGE="${REGISTRY}/${IMAGE_NAME}" + SHORT="${GITHUB_SHA:0:7}" + # Deployments pin by digest, never by tag -- print it for the rollout step. + docker inspect --format='{{index .RepoDigests 0}}' "${IMAGE}:main-${SHORT}" diff --git a/Containerfile.memo-dispatch b/Containerfile.memo-dispatch new file mode 100644 index 0000000..6f5ca2e --- /dev/null +++ b/Containerfile.memo-dispatch @@ -0,0 +1,13 @@ +# Standalone private queue worker; no State Hub API or database credential. +FROM python:3.12-alpine@sha256:b64631e04e4920160c50fbe8d8df828f7f35f06f425cb44aa09bca53e708a35a +RUN apk add --no-cache 'libuuid>=2.42.3-r1' \ + && addgroup -S -g 10001 dispatcher \ + && adduser -S -u 10001 -G dispatcher -H dispatcher \ + && rm -rf /usr/local/bin/pip* /usr/local/lib/python3.12/site-packages/pip* \ + /usr/local/lib/python3.12/site-packages/setuptools* /usr/local/lib/python3.12/site-packages/pkg_resources +WORKDIR /app +COPY api/services/informed_decision_contract.py api/services/memo_forwarding.py api/services/memo_dispatch.py /app/api/services/ +ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 +USER 10001:10001 +ENTRYPOINT ["python", "-m", "api.services.memo_dispatch", "--queue", "/data/private"] +CMD ["run"] diff --git a/api/services/informed_decision_contract.py b/api/services/informed_decision_contract.py index 8c1f0f2..e993ba2 100644 --- a/api/services/informed_decision_contract.py +++ b/api/services/informed_decision_contract.py @@ -39,7 +39,7 @@ ENVELOPE_KEYS = { "kind", "version", "idempotency_key", "memo_class", "review_group", "drafted_by", "serves", "approval_intent", "memo", "packet", } -ENVELOPE_REQUIRED = ENVELOPE_KEYS - {"drafted_by"} +ENVELOPE_REQUIRED = ENVELOPE_KEYS - {"drafted_by", "serves"} INTENT_KEYS = {"action", "principal", "purpose", "target"} TARGET_KEYS = {"id", "type", "system"} PACKET_KEYS = {"item_id", "media_type", "content"} @@ -234,10 +234,15 @@ def validate_envelope(envelope: Any) -> dict: memo_id = _memo_id(memo.get("id")) if "question" in memo: _one_question(memo["question"]) - decision_id = _serves(envelope["serves"]) key = _text(envelope["idempotency_key"], pattern=_KEY_RE, code="idempotency_key_invalid") - if key != stable_idempotency_key(decision_id, memo_id): - raise ContractError("idempotency_key_not_stable") + if "serves" in envelope: + decision_id = _serves(envelope["serves"]) + if key != stable_idempotency_key(decision_id, memo_id): + raise ContractError("idempotency_key_not_stable") + elif not (memo_id.startswith("review-") and _UUID_RE.fullmatch(memo_id[7:]) and key == memo_id): + # Lifecycle service supplies its retained snapshot id/key. Never invent + # a Hub decision relationship or rewrite that immutable envelope. + raise ContractError("lifecycle_handoff_key_invalid") _slug(envelope["memo_class"], "memo_class") _slug(envelope["review_group"], "review_group") drafted_by = envelope.get("drafted_by") diff --git a/api/services/memo_dispatch.py b/api/services/memo_dispatch.py new file mode 100644 index 0000000..0037f45 --- /dev/null +++ b/api/services/memo_dispatch.py @@ -0,0 +1,174 @@ +"""Private, explicitly queued State Hub memo dispatch. No API route or human act.""" +from contextlib import contextmanager +import argparse +import fcntl +import json +import os +from pathlib import Path +import signal +import stat +import threading +import urllib.error +import urllib.request + +from api.services.informed_decision_contract import MAX_ENVELOPE_BYTES, validate_envelope +from api.services.memo_forwarding import MemoForwarder + +ORIGIN = 'http://informed-decision-workload.informed-decision.svc.cluster.local:8081' +TOKEN = '/var/run/secrets/state-hub/informed-decision/intake-token' +MAX_RESPONSE = 65536 + + +class Refused(RuntimeError): + pass + + +def read_json(path, limit): + with Path(path).open('rb') as stream: + raw = stream.read(limit + 1) + if len(raw) > limit: + raise Refused('input_too_large') + return json.loads(raw) + + +def token_file(path): + # Reopen on every attempt: projected files rotate by symlink replacement. + with Path(path).open('rb') as stream: + raw = stream.read(16385) + value = raw.decode('ascii').strip() + if not value or len(raw) > 16384 or any(c.isspace() for c in value): + raise Refused('token_unavailable') + return value + + +class NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, *args, **kwargs): + return None + + +def transport(method, url, body, headers): + if method != 'POST' or url != ORIGIN + '/v1/memos': + raise Refused('route_refused') + raw = json.dumps(body, allow_nan=False, separators=(',', ':')).encode() + if len(raw) > MAX_ENVELOPE_BYTES: + raise Refused('input_too_large') + opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect()) + request = urllib.request.Request(url, data=raw, headers=headers, method=method) + try: + try: + response = opener.open(request, timeout=5) + except urllib.error.HTTPError as error: + response = error + with response: + status = response.code + raw = response.read(MAX_RESPONSE + 1) + if len(raw) > MAX_RESPONSE: + raise OSError('transport_unavailable') + try: + body = json.loads(raw) + except (ValueError, UnicodeError): + body = {} + return status, body + except Exception: + # Never propagate URLs, response bodies, credentials or transport errors. + raise OSError('transport_unavailable') from None + + +class Dispatcher: + def __init__(self, directory, *, send=transport, read_token=lambda: token_file(TOKEN), clock=None): + self.directory = Path(directory) + if not self.directory.is_absolute(): + raise Refused('absolute_queue_path_required') + self.directory.mkdir(mode=0o700, parents=False, exist_ok=True) + info = self.directory.lstat() + if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700: + raise Refused('private_queue_required') + # SQLite files inherit 0600; reject existing unsafe custody, including links. + self.path = self.directory / 'forward.sqlite' + for name in ('forward.sqlite', 'forward.sqlite-wal', 'forward.sqlite-shm', 'dispatch.lock'): + path = self.directory / name + if path.exists() or path.is_symlink(): + info = path.lstat() + if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o600: + raise Refused('private_queue_required') + self.send, self.read_token, self.clock = send, read_token, clock + with self.lock(): + previous = os.umask(0o077) + try: + self.ledger = MemoForwarder(self.path, transport=send, + intake_base_url=ORIGIN, hub_base_url='http://state-hub.state-hub.svc.cluster.local:8000', + token_reader=read_token, **({'clock': clock} if clock else {})) + finally: + os.umask(previous) + + @contextmanager + def lock(self): + fd = os.open(self.directory / 'dispatch.lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + previous = os.umask(0o077) + try: + yield + finally: + os.umask(previous) + except BlockingIOError: + raise Refused('queue_busy') from None + finally: + os.close(fd) + + def enqueue(self, envelope): + validate_envelope(envelope) + if envelope['memo_class'] != 'decision-record' or envelope['review_group'] != 'net-kingdom-admins': + raise Refused('routing_refused') + with self.lock(): + return self.ledger.submit(envelope, dispatch=False) + + def once(self): + # Hold across remote attempt + local receipt to serialize other processes. + # If killed after POST, restart replays the same envelope/key. Intake owns + # native create uncertainty and refuses automatic native-approval repost. + with self.lock(): + return self.ledger.run_due() + + def counts(self): + with self.lock(), self.ledger._connect() as db: + return dict(db.execute('SELECT state,COUNT(*) FROM pose_ledger GROUP BY state')) + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--queue', required=True, type=Path) + commands = parser.add_subparsers(dest='command', required=True) + commands.add_parser('enqueue').add_argument('--envelope', required=True, type=Path) + for name in ('once', 'run', 'status'): + commands.add_parser(name) + args = parser.parse_args(argv) + try: + dispatcher = Dispatcher(args.queue) + if args.command == 'enqueue': + result = dispatcher.enqueue(read_json(args.envelope, MAX_ENVELOPE_BYTES)) + print(json.dumps({'status': result.status, 'detail': result.detail})) + elif args.command == 'once': + result = dispatcher.once() + print(json.dumps({'status': result.status if result else 'idle'})) + elif args.command == 'status': + print(json.dumps(dispatcher.counts())) + else: + stopped = threading.Event() + for sig in (signal.SIGTERM, signal.SIGINT): + signal.signal(sig, lambda *_: stopped.set()) + while not stopped.is_set(): + try: + dispatcher.once() + except Refused as exc: + if str(exc) != 'queue_busy': + raise + stopped.wait(2) + return 0 + except Exception: + print('{"error":"dispatch_refused"}') + return 1 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/api/services/memo_forwarding.py b/api/services/memo_forwarding.py index d8e9311..7cb9ed8 100644 --- a/api/services/memo_forwarding.py +++ b/api/services/memo_forwarding.py @@ -1,6 +1,7 @@ -"""Dev-only pose, replay, and conflict fixture. +"""Pose, replay, and conflict ledger with injected transport. -Not imported by the API runtime and not scheduled. It never reads a token +Not imported by the API runtime. The private dispatcher composes this ledger +under a process lock; dev tests retain the injectable adapter. It never reads a token file, mints a credential, or dials a default host. Tests inject the transport and, when a pose should carry a header, a token callable. The callable's value is not retained. @@ -48,7 +49,7 @@ class ForwardResult: detail: str = "" -class DevIntakeFixture: +class MemoForwarder: def __init__( self, path: str | Path, @@ -102,7 +103,7 @@ class DevIntakeFixture: """ ) - def submit(self, envelope: dict) -> ForwardResult: + def submit(self, envelope: dict, *, dispatch: bool = True) -> ForwardResult: checked = validate_envelope(envelope) key = checked["idempotency_key"] body_digest = submission_digest(checked) @@ -146,6 +147,8 @@ class DevIntakeFixture: early = ("pending", row["last_error"] or "waiting") if early is not None: return self._result(key, *early) + if not dispatch: + return self._result(key, "pending", "queued") return self._attempt(key) def run_due(self) -> ForwardResult | None: @@ -438,3 +441,7 @@ def _dump(value: dict) -> str: def _iso(timestamp: float) -> str: return datetime.fromtimestamp(timestamp, timezone.utc).isoformat() + + +# Compatibility for the original injectable development fixture. +DevIntakeFixture = MemoForwarder diff --git a/deploy/informed-decision/README.md b/deploy/informed-decision/README.md new file mode 100644 index 0000000..1cfcf28 --- /dev/null +++ b/deploy/informed-decision/README.md @@ -0,0 +1,42 @@ +# Private memo dispatch candidate + +STATE-WP-0096 / INFD-WP-0007. The founder selected the State Hub workload. +`python -m api.services.memo_dispatch` is a private worker, separate from the +State Hub API process. Its queue contains explicitly submitted memo envelopes; +it does not scan decisions or lend its credential to a public Hub endpoint. + +An admitted agent registers, refines and publishes using the lifecycle service, +then requests review. The returned immutable `envelope` is the queue input; no +projection or regeneration is needed. Queue the envelope using `enqueue +--envelope ` in the worker's authorized workload context. This +command validates and retains it without a network side effect. The running +worker forwards it to the fixed intake endpoint using its rotating projected +token. `status` prints counts; `once` performs at most one due attempt. Native +intake receipts retain approval pins in private custody. No credential, content +or upstream error body is printed. Browser review uses the returned lifecycle +outcome link, never a worker-constructed approval. + +The original dev fixture remains injectable. The production wrapper adds private +0700/0600 storage, symlink refusal, a lock across remote attempt/local receipt, +5-second transport timeout, bounded JSON, disabled proxies and no redirects. +A lost reply replays the same envelope/key. Intake's unresolved-create refusal +blocks the queue row; it never creates another native approval automatically. +Concurrent enqueue may report queue_busy while dispatch is in flight; retry the +same explicit enqueue after the bounded attempt. SIGTERM stops between attempts. + +The worker uses the selected existing State Hub service account. It has no +Service/Ingress, database credential, browser route or approval scopes. Only the +intake-audience token is mounted: lifecycle API clients must use their separately +admitted lifecycle-audience token. Token mounts are read-only without subPath. +The worker image does not change the Hub API image or PostgreSQL schema. + +All execution that can write this private queue or run commands as this workload +is trusted to pose memos under its existing admission. Kubernetes exec/filesystem +custody is the boundary. Adding a public enqueue API needs independent caller +identity and authorization; it is not silently enabled here. + +Live deployment is still a distinct reviewed change. Deploy after CCR-2026-0032 +requester delivery, the INFD lifecycle service and policy candidates. Retain the +PVC and rows on withdrawal; rollback stops the worker but does not undo submitted +memos, native approvals or human acts. Native human acceptance and audit receipts +remain STATE-WP-0096-T03 / INFD-WP-0007-T07. diff --git a/deploy/informed-decision/dispatcher.candidate.json b/deploy/informed-decision/dispatcher.candidate.json new file mode 100644 index 0000000..40d7403 --- /dev/null +++ b/deploy/informed-decision/dispatcher.candidate.json @@ -0,0 +1,207 @@ +{ + "apiVersion": "v1", + "kind": "List", + "items": [ + { + "apiVersion": "v1", + "kind": "PersistentVolumeClaim", + "metadata": { + "name": "state-hub-memo-dispatch", + "namespace": "state-hub", + "annotations": { + "state-hub.coulomb.social/admission": "proposal; STATE-WP-0096 / INFD-WP-0007 approved rollout required" + } + }, + "spec": { + "accessModes": [ + "ReadWriteOnce" + ], + "storageClassName": "local-path", + "resources": { + "requests": { + "storage": "1Gi" + } + } + } + }, + { + "apiVersion": "apps/v1", + "kind": "Deployment", + "metadata": { + "name": "state-hub-memo-dispatch", + "namespace": "state-hub", + "annotations": { + "state-hub.coulomb.social/admission": "proposal; STATE-WP-0096 / INFD-WP-0007 approved rollout required" + } + }, + "spec": { + "replicas": 1, + "strategy": { + "type": "Recreate" + }, + "selector": { + "matchLabels": { + "app.kubernetes.io/name": "state-hub", + "app.kubernetes.io/component": "memo-dispatch" + } + }, + "template": { + "metadata": { + "labels": { + "app.kubernetes.io/name": "state-hub", + "app.kubernetes.io/component": "memo-dispatch" + } + }, + "spec": { + "serviceAccountName": "state-hub", + "automountServiceAccountToken": false, + "terminationGracePeriodSeconds": 15, + "securityContext": { + "runAsNonRoot": true, + "runAsUser": 10001, + "runAsGroup": 10001, + "fsGroup": 10001, + "fsGroupChangePolicy": "OnRootMismatch", + "seccompProfile": { + "type": "RuntimeDefault" + } + }, + "containers": [ + { + "name": "dispatcher", + "image": "forgejo.coulomb.social/coulomb/state-hub-memo-dispatch@sha256:a3b311ab99f08ab51a2919e72537ffb1176aad2448fa892bf1bf826e1a11161a", + "securityContext": { + "allowPrivilegeEscalation": false, + "readOnlyRootFilesystem": true, + "capabilities": { + "drop": [ + "ALL" + ] + } + }, + "resources": { + "requests": { + "cpu": "5m", + "memory": "24Mi" + }, + "limits": { + "cpu": "100m", + "memory": "96Mi" + } + }, + "volumeMounts": [ + { + "name": "queue", + "mountPath": "/data" + }, + { + "name": "intake-caller", + "mountPath": "/var/run/secrets/state-hub/informed-decision", + "readOnly": true + } + ] + } + ], + "volumes": [ + { + "name": "queue", + "persistentVolumeClaim": { + "claimName": "state-hub-memo-dispatch" + } + }, + { + "name": "intake-caller", + "projected": { + "defaultMode": 288, + "sources": [ + { + "serviceAccountToken": { + "audience": "informed-decision-intake", + "expirationSeconds": 600, + "path": "intake-token" + } + } + ] + } + } + ] + } + } + } + }, + { + "apiVersion": "networking.k8s.io/v1", + "kind": "NetworkPolicy", + "metadata": { + "name": "state-hub-memo-dispatch", + "namespace": "state-hub", + "annotations": { + "state-hub.coulomb.social/admission": "proposal; STATE-WP-0096 / INFD-WP-0007 approved rollout required" + } + }, + "spec": { + "podSelector": { + "matchLabels": { + "app.kubernetes.io/name": "state-hub", + "app.kubernetes.io/component": "memo-dispatch" + } + }, + "policyTypes": [ + "Ingress", + "Egress" + ], + "ingress": [], + "egress": [ + { + "to": [ + { + "namespaceSelector": { + "matchLabels": { + "kubernetes.io/metadata.name": "kube-system" + } + }, + "podSelector": { + "matchLabels": { + "k8s-app": "kube-dns" + } + } + } + ], + "ports": [ + { + "protocol": "UDP", + "port": 53 + }, + { + "protocol": "TCP", + "port": 53 + } + ] + }, + { + "to": [ + { + "namespaceSelector": { + "matchLabels": { + "kubernetes.io/metadata.name": "informed-decision" + } + }, + "podSelector": { + "matchLabels": { + "app.kubernetes.io/name": "informed-decision" + } + } + } + ], + "ports": [ + { + "protocol": "TCP", + "port": 8081 + } + ] + } + ] + } + } + ] +} diff --git a/docs/evidence/2026-10-04-memo-dispatch/container.json b/docs/evidence/2026-10-04-memo-dispatch/container.json new file mode 100644 index 0000000..c797bc9 --- /dev/null +++ b/docs/evidence/2026-10-04-memo-dispatch/container.json @@ -0,0 +1,16 @@ +{ + "status": "passed", + "image_id": "sha256:a3b311ab99f08ab51a2919e72537ffb1176aad2448fa892bf1bf826e1a11161a", + "checks": [ + "non-root read-only worker starts without external network", + "explicit enqueue persists a lifecycle envelope without inventing a Hub relationship", + "SIGTERM stops without forced kill", + "restart retains the exact pending envelope; missing custody does not drop it", + "private queue and database modes enforced" + ], + "fixture_only": true, + "external_network": "none", + "native_identity_proven": false, + "envelope_sha256": "5be460cc7e1b099733e1ee2275a589147d4f96ab29f78c81c81f14d770aa5dc3", + "cleanup_complete": true +} diff --git a/scripts/smoke_memo_dispatch.py b/scripts/smoke_memo_dispatch.py new file mode 100644 index 0000000..7ec312f --- /dev/null +++ b/scripts/smoke_memo_dispatch.py @@ -0,0 +1,63 @@ +"""Installed dispatcher proof, isolated synthetic queue, no network or credential.""" +import argparse +import hashlib +import json +from pathlib import Path +import subprocess +import time +import uuid + + +def main(): + parser=argparse.ArgumentParser(description=__doc__) + parser.add_argument('--image',required=True) + parser.add_argument('--receipt',type=Path,required=True) + args=parser.parse_args() + name='memo-proof-'+uuid.uuid4().hex[:12] + def docker(*words,check=True): + return subprocess.run(['docker',*words],check=check,capture_output=True,text=True,timeout=45) + checks=[] + try: + docker('volume','create',name) + docker('run','--rm','--network','none','--user','0:0','--mount','type=volume,src='+name+',dst=/data', + '--entrypoint','python',args.image,'-c',"import os;os.chown('/data',10001,10001);os.chmod('/data',0o700)") + docker('run','-d','--name',name,'--network','none','--read-only','--cap-drop','ALL', + '--security-opt','no-new-privileges','--mount','type=volume,src='+name+',dst=/data',args.image) + for _ in range(50): + r=docker('exec',name,'python','-m','api.services.memo_dispatch','--queue','/data/private','status',check=False) + if r.returncode==0:break + time.sleep(.1) + else:raise AssertionError('worker failed startup') + checks.append('non-root read-only worker starts without external network') + ident='review-'+str(uuid.uuid4()) + envelope={'kind':'informed-decision.memo-intake','version':1,'idempotency_key':ident, + 'memo_class':'decision-record','review_group':'net-kingdom-admins', + 'memo':{'id':ident,'question':'Synthetic fixture only?'}, + 'packet':[{'item_id':'proof','media_type':'text/plain','content':'Fixture only.'}], + 'approval_intent':{'action':'review','principal':'pending-human-session','purpose':'Synthetic test', + 'target':{'id':'fixture','type':'decision-memo','system':'informed-decision'}}} + source="from api.services.memo_dispatch import Dispatcher; d=Dispatcher('/data/private'); assert d.enqueue("+repr(envelope)+").status=='pending'; assert d.counts()=={'pending':1}" + docker('exec',name,'python','-c',source) + checks.append('explicit enqueue persists a lifecycle envelope without inventing a Hub relationship') + digest_code="from api.services.memo_dispatch import Dispatcher;import hashlib;d=Dispatcher('/data/private');row=d.ledger.pose("+repr(ident)+");print(hashlib.sha256(row['envelope_json'].encode()).hexdigest())" + before=docker('exec',name,'python','-c',digest_code).stdout.strip() + docker('stop','--time','10',name) + state=json.loads(docker('inspect',name).stdout)[0]['State'];assert state['ExitCode']==0 + checks.append('SIGTERM stops without forced kill') + docker('start',name) + after=docker('exec',name,'python','-c',digest_code).stdout.strip();assert before==after + checks.append('restart retains the exact pending envelope; missing custody does not drop it') + docker('exec',name,'python','-c',"import os,stat;assert os.getuid()==10001;assert stat.S_IMODE(os.stat('/data/private').st_mode)==0o700;assert stat.S_IMODE(os.stat('/data/private/forward.sqlite').st_mode)==0o600") + checks.append('private queue and database modes enforced') + image=json.loads(docker('image','inspect',args.image).stdout)[0] + receipt={'status':'passed','image_id':image['Id'],'checks':checks,'fixture_only':True, + 'external_network':'none','native_identity_proven':False,'envelope_sha256':before} + finally: + docker('rm','-f',name,check=False) + docker('volume','rm',name,check=False) + receipt['cleanup_complete']=docker('inspect',name,check=False).returncode!=0 and docker('volume','inspect',name,check=False).returncode!=0 + assert receipt['cleanup_complete'] + args.receipt.write_text(json.dumps(receipt,indent=2)+'\n') + print(json.dumps({'status':'passed','checks':len(checks),'cleanup_complete':True})) + +if __name__=='__main__':main() diff --git a/tests/test_memo_dispatch.py b/tests/test_memo_dispatch.py new file mode 100644 index 0000000..3bc6c7a --- /dev/null +++ b/tests/test_memo_dispatch.py @@ -0,0 +1,126 @@ +import json +from pathlib import Path +import stat +import subprocess +import sys + +import pytest + +from api.services.memo_dispatch import Dispatcher, Refused, token_file, transport, ORIGIN +from tests.test_memo_forwarding import envelope, receipt, Clock + + +def draft(): + return envelope(memo_class='decision-record') + + +def test_queue_is_explicit_and_receipt_survives_restart(tmp_path): + calls = [] + def send(*args): + calls.append(args) + return 201, receipt() + queue = tmp_path/'private' + worker = Dispatcher(queue, send=send, read_token=lambda: 'synthetic-token') + assert worker.enqueue(draft()).status == 'pending' + assert calls == [] + assert worker.once().status == 'delivered' + assert len(calls) == 1 + assert calls[0][0:2] == ('POST', ORIGIN+'/v1/memos') + reopened = Dispatcher(queue, send=send, read_token=lambda: 'synthetic-token') + assert reopened.enqueue(draft()).status == 'replay' + assert reopened.once() is None + assert reopened.counts() == {'delivered': 1} + assert stat.S_IMODE((queue/'forward.sqlite').stat().st_mode) == 0o600 + assert b'synthetic-token' not in (queue/'forward.sqlite').read_bytes() + + +def test_rotation_retry_and_uncertain_native_create_blocks(tmp_path): + clock = Clock() + tokens = tmp_path/'token';tokens.write_text('first') + seen=[] + def send(method,url,body,headers): + seen.append(headers['Authorization']) + if len(seen)==1: raise OSError('hidden response and token') + return 409, {'error':'approval_create_unresolved'} + worker=Dispatcher(tmp_path/'private',send=send,read_token=lambda:token_file(tokens),clock=clock) + worker.enqueue(draft()) + assert worker.once().status=='pending' + tokens.write_text('second');clock.now+=10 + assert worker.once().status=='blocked' + clock.now+=1000 + assert worker.once() is None + assert seen==['Bearer first','Bearer second'] + assert 'hidden response' not in worker.path.read_bytes().decode(errors='ignore') + + +def test_lock_serializes_separate_processes(tmp_path): + worker=Dispatcher(tmp_path/'private') + with worker.lock(): + result=subprocess.run([sys.executable,'-m','api.services.memo_dispatch','--queue',str(worker.directory),'once'],capture_output=True,text=True) + assert result.returncode==1 + assert json.loads(result.stdout)=={'error':'dispatch_refused'} + + +def test_unsafe_custody_and_wrong_routing_refused(tmp_path): + unsafe=tmp_path/'unsafe';unsafe.mkdir(mode=0o755) + with pytest.raises(Refused): Dispatcher(unsafe) + private=tmp_path/'private';private.mkdir(mode=0o700) + target=tmp_path/'target';target.write_text('secret') + (private/'forward.sqlite').symlink_to(target) + with pytest.raises(Refused): Dispatcher(private) + worker=Dispatcher(tmp_path/'queue') + with pytest.raises(Refused): worker.enqueue(envelope()) + assert worker.counts()=={} + + +def test_transport_refuses_arbitrary_destinations_before_io(): + with pytest.raises(Refused): transport('POST','https://other.example/v1/memos',{}, {'Authorization':'Bearer hidden'}) + + +def test_cli_errors_never_emit_input_content(tmp_path): + path=tmp_path/'bad';path.write_text('synthetic-secret') + result=subprocess.run([sys.executable,'-m','api.services.memo_dispatch','--queue',str(tmp_path/'private'),'enqueue','--envelope',str(path)],capture_output=True,text=True) + assert result.returncode==1 and 'synthetic-secret' not in result.stdout+result.stderr + + +def test_retained_lifecycle_handoff_has_no_invented_hub_relationship(tmp_path): + from api.services.informed_decision_contract import ContractError, validate_envelope + body=draft();body.pop('serves') + body['memo']['id']='review-12345678-1234-1234-1234-123456789abc' + body['idempotency_key']=body['memo']['id'] + assert validate_envelope(body)==body + worker=Dispatcher(tmp_path/'private') + assert worker.enqueue(body).status=='pending' + assert 'serves' not in json.loads(worker.ledger.pose(body['idempotency_key'])['envelope_json']) + body['idempotency_key']='changed-key' + with pytest.raises(ContractError): validate_envelope(body) + + +def test_transport_suppresses_proxy_redirect_and_oversized_response(monkeypatch): + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + import threading + from api.services import memo_dispatch + calls=[] + mode=['redirect'] + class Handler(BaseHTTPRequestHandler): + def log_message(self,*args): pass + def do_POST(self): + calls.append(self.path) + self.rfile.read(int(self.headers['Content-Length'])) + if mode[0]=='redirect': + self.send_response(307);self.send_header('Location','/stolen');self.end_headers() + else: + self.send_response(200);self.end_headers();self.wfile.write(b'x'*(memo_dispatch.MAX_RESPONSE+1)) + server=ThreadingHTTPServer(('127.0.0.1',0),Handler) + thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start() + origin=f'http://127.0.0.1:{server.server_port}' + monkeypatch.setattr(memo_dispatch,'ORIGIN',origin) + monkeypatch.setenv('http_proxy','http://127.0.0.1:1') + try: + assert transport('POST',origin+'/v1/memos',{}, {'Authorization':'Bearer synthetic'})[0]==307 + assert calls==['/v1/memos'] + mode[0]='oversized' + with pytest.raises(OSError,match='transport_unavailable'): + transport('POST',origin+'/v1/memos',{}, {}) + finally: + server.shutdown();server.server_close();thread.join(timeout=2) diff --git a/workplans/STATE-WP-0096-private-decision-dispatch.md b/workplans/STATE-WP-0096-private-decision-dispatch.md new file mode 100644 index 0000000..87e97df --- /dev/null +++ b/workplans/STATE-WP-0096-private-decision-dispatch.md @@ -0,0 +1,78 @@ +--- +id: STATE-WP-0096 +type: workplan +title: "Private decision memo dispatch for the selected State Hub workload" +domain: infotech +repo: state-hub +status: active +owner: codex +topic_slug: netkingdom +created: "2026-10-04" +updated: "2026-10-04" +related: [INFD-WP-0005, INFD-WP-0007] +--- + +Founder requested continued production delivery and selected the existing +system:serviceaccount:state-hub:state-hub identity. No new public privileged +forwarding endpoint: the existing Hub API has no suitable authenticated-caller +boundary for delegating its projected token to arbitrary requests. + +## Compose the durable private queue worker + +```task +id: STATE-WP-0096-T01 +status: done +priority: high +``` + +Explicit local enqueue, durable replay/receipt ledger, single-attempt process +lock, rotating intake-audience token and fixed bounded no-proxy/no-redirect +transport. No API startup hook, auto-discovery, human binding or hub-resolution +mutation. Preserve native approval uncertainty: the intake service decides replay +versus unresolved refusal. Only reviewed local workload execution queues content. + +## Package and validate the concrete deployment candidate + +```task +id: STATE-WP-0096-T02 +status: progress +priority: high +``` + +Standalone non-root image using only standard-library dependencies; isolated +single-replica Recreate worker/PVC, existing exact State Hub service account, +projected audience token and exact intake network peer. Build/scan, local +restart/rotation/concurrency proof and server dry-run. Explicit queue entry means +new decisions require no infrastructure rollout. Source assertions do not become +caller identity. + +## Approved deployment and native agent-human-agent acceptance + +```task +id: STATE-WP-0096-T03 +status: todo +priority: high +``` + +Under the concrete INFD-WP-0007 rollout go, deploy the worker after requester +custody and intake/PDP prerequisites. Queue a retained lifecycle review envelope, +verify stable receipt and protected human review. Human alone accepts; retrieve +native outcome and audit evidence. Unresolved/blocked attempts require review, +never automatic clearing. No automatic Hub decision scanning or public enqueue +route is included in this initial workload composition. + + +## Implementation evidence — 2026-10-04 + +The initial cross-service test exposed an overly narrow Hub envelope contract: +lifecycle review snapshots do not necessarily name a Hub decision. The parser +now accepts exactly the retained review-UUID id/key when serves is absent, while +preserving the stable Hub-derived key requirement whenever serves is supplied. +No relationship, receipt or approval is invented. + +The real Approval Engine/Audit Core component test exercises private queueing, +lost response after native creation, restart, same-key receipt replay (201 then +200), human-only review and independent evidence. Identity/custody remain fixtures. +Five installed-container checks prove private storage, explicit enqueue, restart +and SIGTERM with no network. The three deployment objects pass server dry-run; +no live worker or token projection has been created.