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

@@ -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}"

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"]

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

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

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

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 <private-file>` 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.

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
+              }
+            ]
+          }
+        ]
+      }
+    }
+  ]
+}

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
+}

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

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)

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.