Merge pull request 'docs(verbiage): TL;DR prologues per comms-etiquette v1.0' (#2) from coo/verbiage-1 into main

Reviewed-on: #2

.ciagent/CHECKPOINT.json

@@ -1,8 +0,0 @@
-{
-  "phase": 1,
-  "stage": "execute",
-  "milestone": "v0.2",
-  "phase_role": "execution",
-  "attempts": 0,
-  "updated_at": "2026-10-10T17:00:00Z"
-}

.ciagent/REQUIREMENTS.md

@@ -108,6 +108,6 @@ None defined. Future candidates held as out-of-scope decisions, not requirements
 | REQ-18 | 1 | complete |
 | REQ-19 | 4 | complete |
 | REQ-20 | 4 | complete |
-| REQ-21 | 1 | pending |
-| REQ-22 | 1 | pending |
-| REQ-23 | 2 | pending |
+| REQ-21 | 1 | complete |
+| REQ-22 | 1 | complete |
+| REQ-23 | 2 | complete |

.ciagent/ROADMAP.md

@@ -22,7 +22,7 @@ Autonomy: **guided** — escalate on every decision gate; threshold 0.85; 1 revi
 - **Requirements covered**: REQ-21, REQ-22
 - **Success criteria**: artifact builds reproducibly from a clean checkout (two consecutive same-environment builds → byte-identical sha256); `python3 coreci-broker.pyz` boots with behavior byte-identical to `python3 coreci_broker.py` (v0.1 suite green through the packaged path); UI pages + client module served from the archive; first-boot semantics preserved on second boot; the `v0.2.1` release carries the `.pyz` attached by `scripts/release.py` (D-020 starts here; `v0.2.0` carries none — G-021)
 - **Depends on**: 0
-- **Status**: not_started
+- **Status**: complete (shipped as `v0.2.1` — 325 cumulative tests green, verify four-layer PASS, .pyz attached by script, asset id 396)
 
 ## Phase 2 — Docker + Image Publish
 
@@ -30,7 +30,7 @@ Autonomy: **guided** — escalate on every decision gate; threshold 0.85; 1 revi
 - **Requirements covered**: REQ-23 (+ REQ-21/REQ-22 docs)
 - **Success criteria**: daemonless publish client stub-verified end-to-end (no live-forge writes, no container runtime); Dockerfile/.dockerignore consistency tests green; this phase's release (`v0.2.2`) carries the `.pyz` attached alongside the source archive with notes; bounded-retry → `release_pending` paths covered by tests
 - **Depends on**: 1
-- **Status**: not_started
+- **Status**: complete (shipped as `v0.2.2` — 341 cumulative tests green, verify four-layer PASS, .pyz attached by script, asset id 397)
 
 ## Phase 3 — Final Review + Ship
 
@@ -38,7 +38,7 @@ Autonomy: **guided** — escalate on every decision gate; threshold 0.85; 1 revi
 - **Requirements covered**: REQ-21, REQ-22, REQ-23 (acceptance)
 - **Success criteria**: all v0.1 + v0.2 acceptance green; security review PASS; `v0.2.3` on `main` with the `.pyz` attached + notes on the gitea release; docker image published as a gitea package (or escalated per the bounded-retry contract + closure runbook, never blocking merge); REQUIREMENTS/ROADMAP closed out; checkpoint cleared
 - **Depends on**: 2
-- **Status**: not_started
+- **Status**: complete (shipped as `v0.2.3` — milestone release; 356 cumulative tests green, security review PASS-WITH-FINDINGS remediated, `.pyz` attached + notes, image published once)
 
 ## Post-Milestone Notes (carry-forward from v0.1, GRILL G-008)
 

.dockerignore

@@ -0,0 +1,17 @@
+# Build-context hygiene (G-015): no secrets, state, or build artifacts
+# may enter an image build context.
+.git
+.env
+.env.*
+data/
+*.db
+*.log
+dist/
+__pycache__/
+tests/
+.ciagent/
+# Nested matches too (P3-03, CWE-693): the root-anchored `__pycache__/`
+# above misses cli/__pycache__/*.pyc — build artifacts must stay out of
+# the context wherever they sit in the tree.
+**/__pycache__/
+**/*.pyc

Dockerfile

@@ -0,0 +1,42 @@
+# coreci-broker docker image (v0.2, REQ-23).
+#
+# Equivalence mapping to the daemonless-published image
+# (scripts/publish_image.py — the pipeline's source of truth):
+#   same base family: python:3.11-slim (D-022 pinned minor)
+#   same artifact:    coreci_broker.py + coreci_broker_client.py +
+#                     coreci_broker_ui/ + cli/__main__.py + _cli_version.py,
+#                     packed by the same `python3 -m zipapp` invocation
+#                     (--python "/usr/bin/env python3" --compress)
+#   same entrypoint:  ["python3", "/app/coreci-broker.pyz"]
+#   same state anchor: WORKDIR /state (volume-backed; first boot generates
+#                     .env, broker.db, broker.audit.log there)
+# Documented differences (equivalence required, byte-identity NOT):
+#   - docker-build layering (COPY + RUN) vs the daemonless client's
+#     verbatim base layers + one deterministic broker layer
+#   - version provenance: ARG VERSION here vs build-time _cli_version.py
+#     injection in scripts/build_cli.py
+#   - no normalize pass here (docker rebuilds are not digest-pinned)
+FROM python:3.11-slim
+
+# Staging mirrors scripts/build_cli.py EXACTLY (P3-03, CWE-636): Docker
+# copies directory CONTENTS flattened, so the staging layout must come
+# from the destination paths — root sources at /app/src/, UI pages under
+# /app/src/coreci_broker_ui/ (cli/__main__.py:80-83 materializes only
+# coreci_broker_ui/* members; a flattened copy leaves /ui/* 404ing), and
+# the shim staged as the archive-root __main__.py (zipapp boots the
+# root __main__.py; the file-to-file rename mirrors build_cli's
+# STAGED_SHIM staging).
+COPY coreci_broker.py coreci_broker_client.py /app/src/
+COPY coreci_broker_ui/ /app/src/coreci_broker_ui/
+COPY cli/__main__.py /app/src/__main__.py
+
+ARG VERSION=dev
+RUN printf 'VERSION = "%s"\n' "$VERSION" > /app/src/_cli_version.py \
+    && python3 -m zipapp /app/src -o /app/coreci-broker.pyz \
+        --python "/usr/bin/env python3" --compress
+
+WORKDIR /state
+
+ENTRYPOINT ["python3", "/app/coreci-broker.pyz"]
+
+EXPOSE 8765

README.md

@@ -1,5 +1,8 @@
 # CoreCI Broker
 
+**TL;DR** — CoreCI Broker — single-file credential broker for autonomous agent tools (Open WebUI workspace tool / standalone). This file explains how to stand it up and who to contact; the machinery detail stays lower under Details.
+
+
 A single-file credential broker for autonomous agent tools on the Open WebUI
 host. Tools authenticate with derived tokens instead of shared secrets, the
 CEO gates access through UI-managed allowlists of app UUIDs and user IDs,
@@ -177,4 +180,43 @@ importable). The AC1–AC14 acceptance matrix is green.
 | `coreci_broker_ui/` | Server-rendered HTML for the four pages + dashboard |
 | `reference_tools/` | Tool-author templates (proxmox-style §11 flow + minimal showcase) |
 | `tests/` | Stdlib-unittest suite incl. the AC1–AC14 acceptance matrix |
-| `.ciagent/` | CI pipeline artifacts (project, requirements, roadmap, decisions) |
+| `.ciagent/` | CI pipeline artifacts (project, requirements, roadmap, decisions) |
+## CLI and Docker (v0.2)
+
+From v0.2.1 onward every gitea release attaches the broker packaged as a
+self-contained CLI zipapp: `coreci-broker-vX.Y.Z.pyz` (its sha256 is in
+the release notes).
+
+**CLI install + usage:**
+
+```bash
+# from the release page assets:
+chmod +x coreci-broker-vX.Y.Z.pyz
+./coreci-broker-vX.Y.Z.pyz --version          # prints the release version
+./coreci-broker-vX.Y.Z.pyz                   # boots the standalone broker
+# or: python3 coreci-broker-vX.Y.Z.pyz       (Python 3.11+)
+```
+
+The artifact bundles the broker, the admin UI pages, and the client
+module — one file serves everything (UI at `/ui/`, client at
+`/static/coreci_broker_client.py`). First boot prints
+`BROKER_ADMIN_PASS=...` exactly once and creates `.env`, `broker.db`,
+and `broker.audit.log` mode 0600 in the state dir; state is anchored by
+`BROKER_ENV_PATH` (relative paths stay CWD-relative). The full
+`BROKER_*` env contract is in the Configuration section above.
+
+**Docker:**
+
+```bash
+docker pull git.coreci.dev/coreci/broker:<tag>     # also :latest
+docker run -p 8765:8765 -v broker-state:/state \
+    git.coreci.dev/coreci/broker:<tag>
+```
+
+State lives on the `/state` volume (`WORKDIR /state`; first boot
+generates `.env`, `broker.db`, and `broker.audit.log there) — mount a
+persistent volume or the admin pass and any registrations are lost with
+the container. The image is built and published once per milestone at
+the milestone release (see `scripts/publish_image.py` — the daemonless
+registry client; the `Dockerfile` documents the docker-user path and
+its equivalence mapping).

scripts/publish_image.py

@@ -0,0 +1,844 @@
+#!/usr/bin/env python3
+"""Daemonless OCI image publish client (D-019; stdlib only).
+
+Usage:
+    python3 scripts/publish_image.py <tag> [--smoke]
+        [--artifact PATH] [--hub-registry URL] [--hub-auth URL]
+        [--gitea-registry URL] [--env-root DIR] [--owner O] [--repo R]
+        [--base-manifest DIGEST] [--timeout SECS]
+
+Flow (REQ-23, RESEARCH §3): pull the pinned ``python:3.11-slim`` base
+from the public registry anonymously (token -> OCI index -> resolve the
+linux/amd64 child manifest by digest -> config + layer blobs; digests
+recorded for provenance; ``--base-manifest`` overrides the tag
+resolution for exact re-publishes) -> authenticate to the gitea
+container registry (``GET /v2/token`` with Basic credentials — the
+gitea token comes only from the sanctioned env files via
+``scripts/_env_secrets.py``; the endpoint answers 200 even anonymously
+with an empty-scope JWT, so the issued token must be non-empty and
+failures must be loud — never treat a 200 alone as authorization) ->
+push: HEAD each blob first (200 -> skip — retries resume cheap; the
+~45.7 MB base layers are re-uploaded once per milestone because the
+registries cannot cross-mount) -> monolithic blob upload
+``POST /v2/<name>/blobs/uploads/?digest=`` (202 + ``Location`` ->
+two-step PUT) with explicit Location-following on 307/308 (stdlib
+urllib does NOT method-preserve on PUT/POST redirects) -> deterministic
+broker layer (sorted tar entries, fixed metadata, member at the stable
+path ``/app/coreci-broker.pyz``, ``gzip.compress(..., mtime=0)``) ->
+new config blob (base config + appended diff_id; Entrypoint
+``["python3", "/app/coreci-broker.pyz"]``, ``Cmd []``, ``WorkingDir
+"/state"``, version label; serialized with sort_keys) -> OCI manifest
+PUT with the base layer descriptors reused verbatim under ``<tag>``
+FIRST, then ``latest`` (D-021).
+
+Exit codes: 0 only after both manifest PUTs confirm 201; 1 after a
+remote failure — bounded retries (3 attempts, 2s/4s backoff) exhausted
+OR a permanent registry answer — printing the machine-readable
+``IMAGE_PENDING <image>:<tag>`` marker for the ship escalation (never
+blocks tag+merge); 2 for local precondition errors (missing artifact,
+token, or config — nothing attempted remotely, no marker). Token values
+are never printed.
+
+Smoke mode (``--smoke``; G-013) — read-only pre-ship verify. NEVER part
+of the default publish flow (the flag dispatches an early return; the
+publish path above never consults it): four GET-only checks against
+the live endpoints, run in order and stopping at the first failure —
+nothing is uploaded, no manifest is PUT, no remote state is mutated.
+
+1. ``GET {gitea}/v2/`` must answer 401 with a Bearer challenge whose
+   realm is the gitea token endpoint (the realm is recorded for 2).
+2. Token->JWT exchange at that recorded realm
+   (``?service=container_registry&scope=repository:{o}/{r}:pull,push``)
+   with Basic auth — the gitea token resolves ONLY from the sanctioned
+   env files via ``scripts/_env_secrets.py`` (never the shell
+   environment); an empty token in the 200 answer is the anonymous-200
+   trap and fails loudly.
+3. ``GET {gitea}/api/v1/repos/{o}/{r}/releases/tags/<tag>`` — 200
+   (exists) and 404 (not yet created) are BOTH acceptable pre-ship and
+   the line says which; the release's asset names print as a fact, and
+   a hard assertion binds ONLY when the tag is exactly ``v0.2.0``
+   (G-021 forge-side confirmation: the phase-0 release predates the
+   CLI and must carry NO ``.pyz`` asset, so any ``.pyz`` in its asset
+   list fails the smoke — the structural half of the guarantee,
+   release.py taking the tag as a positional argument with no v0.2.0
+   code path, is already machine-checked in the suite).
+4. Docker Hub anonymous index resolve of ``python:3.11-slim`` with the
+   full Accept set must yield an index whose linux/amd64 child is
+   resolved; the amd64 child digest is printed — the base-drift fact
+   the P3-05 ship compares against expectations.
+
+Smoke exit codes mirror the publish flow: 0 all four PASS; 1 with one
+clear FAILED line per failure (checks after the failed one never run);
+2 for local preconditions (missing token — nothing attempted
+remotely). Smoke never prints token values and never emits
+``IMAGE_PENDING`` (nothing was attempted, so nothing is pending).
+
+Live gating (G-013): the LIVE smoke executes only at the P3-05 ship
+step, which sets ``PUBLISH_SMOKE=1`` in the environment before invoking
+this mode — this script neither reads nor requires that variable (the
+SHIP GATE owns the gating). The smoke's LOGIC is what the default
+unittest suite exercises (stub-tested, zero network access); the live
+smoke itself never runs in the default suite.
+"""
+
+import argparse
+import base64
+import gzip
+import hashlib
+import http.client
+import io
+import json
+import os
+import re
+import sys
+import tarfile
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+
+_SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__))
+if _SCRIPTS_DIR not in sys.path:
+    sys.path.insert(0, _SCRIPTS_DIR)
+
+import _env_secrets  # sibling import (scripts/ dir on sys.path)
+
+METADATA_TIMEOUT_SECONDS = 30
+BLOB_TIMEOUT_SECONDS = 120
+MAX_ATTEMPTS = 3
+BACKOFF_SECONDS = (2, 4)
+MAX_REDIRECTS = 5
+BASE_REPO = "library/python"
+BASE_TAG = "3.11-slim"
+HUB_AUTH_SERVICE = "registry.docker.io"
+GITEA_AUTH_SERVICE = "container_registry"
+# The smoke-mode G-021 forge-side no-.pyz assertion binds ONLY this
+# tag: the phase-0 release predates the CLI (D-020 starts at v0.2.1).
+SMOKE_G021_TAG = "v0.2.0"
+
+OCI_INDEX = "application/vnd.oci.image.index.v1+json"
+OCI_MANIFEST = "application/vnd.oci.image.manifest.v1+json"
+OCI_CONFIG = "application/vnd.oci.image.config.v1+json"
+OCI_LAYER = "application/vnd.oci.image.layer.v1.tar+gzip"
+DOCKER_LIST = "application/vnd.docker.distribution.manifest.list.v2+json"
+DOCKER_MANIFEST = "application/vnd.docker.distribution.manifest.v2+json"
+ACCEPT_MANIFESTS = ", ".join(
+    (OCI_INDEX, DOCKER_LIST, OCI_MANIFEST, DOCKER_MANIFEST))
+
+
+class PublishError(Exception):
+    """Image-publisher failure base."""
+
+
+class TransientFailure(PublishError):
+    """Connection error or 5xx persisting past the bounded retries."""
+
+
+class RegistryApiError(PublishError):
+    """Permanent registry answer carrying the remote detail."""
+
+
+class LocalError(PublishError):
+    """Local precondition failure (nothing was attempted remotely)."""
+
+
+def repo_root():
+    return os.path.dirname(_SCRIPTS_DIR)
+
+
+def load_release_defaults(root=None):
+    """Read the release block from .ciagent/config.json ({} if absent)."""
+    path = os.path.join(root or repo_root(), ".ciagent", "config.json")
+    try:
+        with open(path, "r", encoding="utf-8") as handle:
+            config = json.load(handle)
+    except (OSError, ValueError):
+        return {}
+    block = config.get("release")
+    return block if isinstance(block, dict) else {}
+
+
+def _error_detail(body):
+    try:
+        payload = json.loads(body.decode("utf-8", "replace"))
+    except ValueError:
+        text = body.decode("utf-8", "replace").strip()
+        return text[:200] if text else "<empty body>"
+    if isinstance(payload, dict):
+        for key in ("errors", "message"):
+            if payload.get(key):
+                return json.dumps(payload[key])[:200]
+    return json.dumps(payload)[:200] if not isinstance(payload, str) else payload[:200]
+
+
+def _header(headers, name):
+    """Case-insensitive response-header lookup — ``_attempt_once``
+    returns a plain dict whose keys keep the wire casing, and registry
+    header casing is not guaranteed (e.g. WWW-Authenticate)."""
+    for key, value in headers.items():
+        if key.lower() == name.lower():
+            return value
+    return None
+
+
+def _parse_json(body, context, status):
+    """Parse a remote JSON body (token / manifest / config answers).
+
+    Malformed bodies raise RegistryApiError — never an uncaught
+    ValueError — so main()'s handler keeps the exit-1 + IMAGE_PENDING
+    marker contract for this failure class (CWE-754).
+    """
+    try:
+        return json.loads(body.decode("utf-8", "replace"))
+    except ValueError as error:
+        raise RegistryApiError(
+            "{} answered malformed JSON ({})".format(context, error),
+            status, body)
+
+
+def _attempt_once(method, url, headers, data, timeout):
+    """One HTTP exchange; HTTPError becomes a completed answer."""
+    request = urllib.request.Request(url, data=data, method=method)
+    for key, value in headers.items():
+        request.add_header(key, value)
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            body = b"" if method == "HEAD" else response.read()
+            return response.status, body, dict(response.headers.items())
+    except urllib.error.HTTPError as error:
+        body = b"" if method == "HEAD" else error.read()
+        return error.code, body, dict((error.headers or {}).items())
+
+
+def _origin(url):
+    """(scheme, netloc) of a URL — two URLs share an origin iff both match."""
+    parts = urllib.parse.urlsplit(url)
+    return (parts.scheme, parts.netloc)
+
+
+def _attempt_following(method, url, headers, data, timeout):
+    """One attempt, following 307/308 redirects explicitly (urllib does
+    NOT method-preserve on PUT/POST — the Location may be relative).
+
+    Cross-origin discipline (CWE-200): when a followed Location resolves
+    to a different origin than the ORIGINAL request's, the Authorization
+    header is stripped from the re-issued request — the registry JWT
+    must never reach another origin (blob-upload Locations are commonly
+    presigned external storage). Same-origin redirects keep it; once
+    stripped it is never re-attached."""
+    current_url = url
+    current_headers = dict(headers)
+    for _ in range(MAX_REDIRECTS + 1):
+        status, body, resp_headers = _attempt_once(
+            method, current_url, current_headers, data, timeout)
+        if status in (307, 308) and resp_headers.get("Location"):
+            next_url = urllib.parse.urljoin(
+                current_url, resp_headers["Location"])
+            if _origin(next_url) != _origin(url):
+                current_headers.pop("Authorization", None)
+            current_url = next_url
+            continue
+        return status, body, resp_headers
+    raise RegistryApiError(
+        "too many redirects following {} {}".format(method, url), 500, b"")
+
+
+def _request(method, url, headers, data, timeout,
+             max_attempts=None, backoff=None):
+    """Bounded-retry HTTP: retries connection errors and 5xx answers.
+
+    Returns (status, body, response_headers) for any answer below 500;
+    a 4xx returns immediately (permanent). Raises TransientFailure once
+    the retry budget is exhausted.
+    """
+    attempts = MAX_ATTEMPTS if max_attempts is None else int(max_attempts)
+    delays = BACKOFF_SECONDS if backoff is None else tuple(backoff)
+    failure = "unknown failure"
+    for attempt in range(1, attempts + 1):
+        try:
+            status, body, resp_headers = _attempt_following(
+                method, url, headers, data, timeout)
+        except (OSError, http.client.HTTPException) as error:
+            status = None
+            failure = "connection error: {}: {}".format(
+                type(error).__name__, error)
+        else:
+            if status < 500:
+                return status, body, resp_headers
+            failure = "HTTP {}: {}".format(status, _error_detail(body))
+        if attempt < attempts and delays:
+            time.sleep(delays[min(attempt - 1, len(delays) - 1)])
+    raise TransientFailure(
+        "{} {} failed after {} attempt(s): {}".format(
+            method, url, attempts, failure))
+
+
+def _hub_token(hub_auth, scope):
+    query = urllib.parse.urlencode(
+        {"service": HUB_AUTH_SERVICE, "scope": scope})
+    status, body, _ = _request(
+        "GET", "{}?{}".format(hub_auth, query), {}, None,
+        METADATA_TIMEOUT_SECONDS)
+    if status != 200:
+        raise RegistryApiError("hub token request failed", status, body)
+    payload = _parse_json(body, "hub token endpoint", status)
+    token = ""
+    if isinstance(payload, dict):
+        token = payload.get("token") or payload.get("access_token") or ""
+    if not token:
+        raise RegistryApiError("hub token endpoint returned no token",
+                               status, body)
+    return token
+
+
+def _fetch_manifest(registry, repository, ref, bearer, accept=ACCEPT_MANIFESTS):
+    url = "{}/v2/{}/manifests/{}".format(registry.rstrip("/"), repository, ref)
+    status, body, _ = _request(
+        "GET", url, {"Authorization": "Bearer {}".format(bearer),
+                     "Accept": accept}, None, METADATA_TIMEOUT_SECONDS)
+    if status != 200:
+        raise RegistryApiError(
+            "manifest fetch {} {}".format(repository, ref), status, body)
+    return _parse_json(
+        body, "manifest fetch {} {}".format(repository, ref), status)
+
+
+def _resolve_amd64_child(index):
+    """linux/amd64 child digest from a base index; malformed index
+    shapes degrade to the not-found RegistryApiError, never a crash."""
+    manifests = index.get("manifests") if isinstance(index, dict) else None
+    if not isinstance(manifests, list):
+        manifests = []
+    for descriptor in manifests:
+        if not isinstance(descriptor, dict):
+            continue
+        platform = descriptor.get("platform")
+        if not isinstance(platform, dict):
+            continue
+        if platform.get("os") == "linux" and platform.get(
+                "architecture") == "amd64":
+            return descriptor.get("digest")
+    raise RegistryApiError(
+        "no linux/amd64 child manifest in the base index", 404,
+        json.dumps(manifests).encode("utf-8", "replace"))
+
+
+def _fetch_blob(registry, repository, digest, bearer, timeout):
+    url = "{}/v2/{}/blobs/{}".format(
+        registry.rstrip("/"), repository, digest)
+    status, body, _ = _request(
+        "GET", url, {"Authorization": "Bearer {}".format(bearer)}, None,
+        timeout)
+    if status != 200:
+        raise RegistryApiError("blob fetch {}".format(digest), status, body)
+    if "sha256:" + hashlib.sha256(body).hexdigest() != digest:
+        raise RegistryApiError("blob digest mismatch {}".format(digest),
+                               500, b"")
+    return body
+
+
+def _child_parts(child):
+    """Validated (config descriptor, layer descriptors) of a base child
+    manifest — missing or malformed keys raise a clear RegistryApiError
+    (never an uncaught KeyError), keeping the marker contract for this
+    failure class (CWE-754)."""
+    def malformed(reason):
+        return RegistryApiError(
+            "malformed base child manifest: {}".format(reason), 500,
+            json.dumps(child).encode("utf-8", "replace"))
+
+    if not isinstance(child, dict):
+        raise malformed("not a JSON object")
+    config = child.get("config")
+    if not isinstance(config, dict) or not config.get("digest"):
+        raise malformed("missing or empty config.digest")
+    layers = child.get("layers")
+    if not isinstance(layers, list):
+        raise malformed("missing or non-list layers")
+    for descriptor in layers:
+        if not isinstance(descriptor, dict) or not descriptor.get("digest"):
+            raise malformed("layer descriptor without a digest")
+    return config, layers
+
+
+def _gitea_jwt(token_url, owner, repo, token, timeout):
+    """Token->JWT exchange at a token endpoint URL (Basic auth).
+
+    ``token_url`` is the FULL endpoint: the registry's ``/v2/token``
+    for the publish flow, or the realm recorded from a ``/v2/``
+    challenge for the smoke (the standard token-auth flow).
+    """
+    query = urllib.parse.urlencode({
+        "service": GITEA_AUTH_SERVICE,
+        "scope": "repository:{}/{}:pull,push".format(owner, repo),
+    })
+    basic = base64.b64encode(
+        "publisher:{}".format(token).encode("utf-8")).decode("ascii")
+    status, body, _ = _request(
+        "GET", "{}?{}".format(token_url, query),
+        {"Authorization": "Basic {}".format(basic)}, None, timeout)
+    if status != 200:
+        raise RegistryApiError("gitea registry token request failed",
+                               status, body)
+    payload = _parse_json(body, "gitea registry token endpoint", status)
+    jwt = (payload.get("token") or "") if isinstance(payload, dict) else ""
+    if not jwt:
+        # The endpoint answers 200 even anonymously with an unusable
+        # (empty-scope) JWT — a 200 proves nothing about push rights.
+        raise RegistryApiError(
+            "gitea registry token endpoint returned an empty token "
+            "(anonymous-200 trap refused; never treat 200 as authorization)",
+            status, body)
+    return jwt
+
+
+def _bearer_challenge(www_authenticate):
+    """Realm recorded from a Bearer WWW-Authenticate challenge.
+
+    Returns None when the header is absent, is not a Bearer challenge,
+    or carries no realm — a challenge without a realm cannot be
+    followed to the token endpoint, so the smoke treats that as no
+    challenge at all.
+    """
+    if not www_authenticate:
+        return None
+    parts = www_authenticate.strip().split(None, 1)
+    if len(parts) < 2 or parts[0].lower() != "bearer":
+        return None
+    match = re.search(r'\brealm\s*=\s*"([^"]*)"', parts[1])
+    return match.group(1) if match else None
+
+
+def _asset_names(release):
+    """Asset display names from a gitea release payload — odd shapes
+    degrade to [] (the read-only smoke never crashes on them)."""
+    if not isinstance(release, dict):
+        return []
+    assets = release.get("assets")
+    if not isinstance(assets, list):
+        return []
+    names = []
+    for asset in assets:
+        if isinstance(asset, dict) and isinstance(asset.get("name"), str):
+            names.append(asset["name"])
+    return names
+
+
+def _blob_exists(gitea_registry, name, digest, jwt, timeout):
+    url = "{}/v2/{}/blobs/{}".format(
+        gitea_registry.rstrip("/"), name, digest)
+    status, _, _ = _request(
+        "HEAD", url, {"Authorization": "Bearer {}".format(jwt)}, None,
+        timeout)
+    return status == 200
+
+
+def _upload_blob(gitea_registry, name, digest, blob, jwt, timeout):
+    base_url = gitea_registry.rstrip("/")
+    url = "{}/v2/{}/blobs/uploads/?digest={}".format(base_url, name, digest)
+    headers = {"Authorization": "Bearer {}".format(jwt),
+               "Content-Type": "application/octet-stream"}
+    status, body, resp_headers = _request(
+        "POST", url, headers, blob, timeout)
+    if status in (200, 201):
+        return
+    if status == 202 and resp_headers.get("Location"):
+        location = urllib.parse.urljoin(url, resp_headers["Location"])
+        if "digest=" not in location:
+            separator = "&" if "?" in location else "?"
+            location = "{}{}digest={}".format(location, separator, digest)
+        status, body, _ = _request(
+            "PUT", location, headers, blob, timeout)
+        if status in (200, 201):
+            return
+    raise RegistryApiError("blob upload {}".format(digest), status, body)
+
+
+def _push_blob_if_missing(gitea_registry, name, digest, blob, jwt, timeout):
+    """HEAD-skip then upload — retries resume cheap."""
+    if _blob_exists(gitea_registry, name, digest, jwt, timeout):
+        return False
+    _upload_blob(gitea_registry, name, digest, blob, jwt, timeout)
+    return True
+
+
+def build_broker_layer(artifact_bytes):
+    """Deterministic layer carrying /app/coreci-broker.pyz.
+
+    Returns (gzip_blob, layer_digest, diff_id): sorted single entry,
+    fixed TarInfo metadata, gzip with mtime=0 — byte-stable rebuilds.
+    """
+    info = tarfile.TarInfo("app/coreci-broker.pyz")
+    info.size = len(artifact_bytes)
+    info.mode = 0o644
+    info.uid = 0
+    info.gid = 0
+    info.uname = ""
+    info.gname = ""
+    info.mtime = 0
+    buffer = io.BytesIO()
+    with tarfile.open(fileobj=buffer, mode="w",
+                      format=tarfile.GNU_FORMAT) as archive:
+        archive.addfile(info, io.BytesIO(artifact_bytes))
+    raw = buffer.getvalue()
+    compressed = gzip.compress(raw, compresslevel=9, mtime=0)
+    layer_digest = "sha256:" + hashlib.sha256(compressed).hexdigest()
+    diff_id = "sha256:" + hashlib.sha256(raw).hexdigest()
+    return compressed, layer_digest, diff_id
+
+
+def build_config_blob(base_config, diff_id, tag):
+    """Base config + appended diff_id + CLI entrypoint/WorkingDir/label.
+
+    Serialized with sort_keys — deterministic config digest.
+    """
+    config = dict(base_config)
+    rootfs = dict(base_config.get("rootfs") or {})
+    rootfs["diff_ids"] = list(rootfs.get("diff_ids") or []) + [diff_id]
+    if "type" not in rootfs:
+        rootfs["type"] = "layers"
+    config["rootfs"] = rootfs
+    inner = dict(base_config.get("config") or {})
+    inner["Entrypoint"] = ["python3", "/app/coreci-broker.pyz"]
+    inner["Cmd"] = []
+    inner["WorkingDir"] = "/state"
+    inner["Labels"] = {"org.opencontainers.image.version": tag}
+    config["config"] = inner
+    return json.dumps(config, sort_keys=True,
+                      separators=(",", ":")).encode("utf-8")
+
+
+def build_manifest_blob(base_layer_descriptors, broker_layer_descriptor,
+                        config_descriptor):
+    manifest = {
+        "schemaVersion": 2,
+        "mediaType": OCI_MANIFEST,
+        "config": config_descriptor,
+        "layers": list(base_layer_descriptors) + [broker_layer_descriptor],
+    }
+    return json.dumps(manifest, sort_keys=True,
+                      separators=(",", ":")).encode("utf-8")
+
+
+def _descriptor(digest, size, media_type):
+    return {"mediaType": media_type, "digest": digest, "size": size}
+
+
+def _put_manifest(gitea_registry, name, tag, manifest_blob, jwt, timeout):
+    url = "{}/v2/{}/manifests/{}".format(
+        gitea_registry.rstrip("/"), name, tag)
+    status, body, _ = _request(
+        "PUT", url, {"Authorization": "Bearer {}".format(jwt),
+                     "Content-Type": OCI_MANIFEST}, manifest_blob, timeout)
+    if status not in (200, 201):
+        raise RegistryApiError("manifest PUT {}".format(tag), status, body)
+    return "sha256:" + hashlib.sha256(manifest_blob).hexdigest()
+
+
+def _load_artifact(path):
+    if not os.path.isfile(path):
+        raise LocalError("artifact not found: {}".format(path))
+    with open(path, "rb") as handle:
+        return handle.read()
+
+
+def smoke(tag, gitea_registry, hub_registry, hub_auth, owner, repo,
+          timeout, env_root):
+    """G-013 read-only live smoke: the four GET-only pre-ship checks.
+
+    Never part of the default publish flow; never mutates remote state
+    (no uploads, no manifest PUTs). The LIVE invocation happens only
+    at the P3-05 ship step, which gates it by setting PUBLISH_SMOKE=1
+    in the environment — this script neither reads nor requires that
+    variable (the SHIP GATE owns the gating). The default unittest
+    suite exercises this LOGIC against stubs (zero network).
+
+    Checks, in order, stopping at the first failure (see the module
+    docstring for the full G-021 tie-in): [1/4] the gitea /v2/ 401
+    Bearer challenge (realm recorded); [2/4] the token->JWT exchange at
+    the recorded realm — the gitea token resolves only from the
+    sanctioned env files via ``_env_secrets.get_token`` (never the
+    shell environment), and an empty token is the anonymous-200 trap
+    refused loudly; [3/4] the gitea API release-by-tag GET — 200
+    (exists) and 404 (not yet created) are both acceptable pre-ship,
+    asset names print as a fact, and the hard no-.pyz assertion binds
+    ONLY the tag v0.2.0 exactly (G-021 forge-side confirmation);
+    [4/4] the Docker Hub anonymous index resolve of python:3.11-slim —
+    the linux/amd64 child digest prints (the drift fact P3-05
+    compares against expectations).
+
+    Returns 0 (all PASS), or 1 after one clear FAILED line (token
+    values are never printed; IMAGE_PENDING is never emitted — nothing
+    was attempted, so nothing is pending). Local preconditions (a
+    missing token) raise ``_env_secrets.SecretsError`` BEFORE any
+    request; main() maps that to exit 2.
+    """
+    gitea_token = _env_secrets.get_token(root=env_root)
+    metadata_timeout = min(METADATA_TIMEOUT_SECONDS, timeout)
+    gitea_base = gitea_registry.rstrip("/")
+
+    # [1/4] gitea /v2/ auth challenge (anonymous; realm recorded).
+    try:
+        status, _, resp_headers = _request(
+            "GET", "{}/v2/".format(gitea_base), {}, None, metadata_timeout)
+    except (TransientFailure, RegistryApiError) as error:
+        print("FAILED [1/4] gitea /v2/ auth challenge: {}".format(error))
+        return 1
+    challenge = _header(resp_headers, "WWW-Authenticate")
+    realm = _bearer_challenge(challenge)
+    if status != 401 or not realm:
+        print("FAILED [1/4] gitea /v2/ auth challenge: expected 401 with "
+              "a Bearer challenge, got HTTP {} (WWW-Authenticate: {!r})"
+              .format(status, challenge))
+        return 1
+    print("PASS [1/4] gitea /v2/ auth challenge: 401 + Bearer challenge "
+          "(realm {})".format(realm))
+
+    # [2/4] token->JWT exchange at the recorded realm (Basic auth).
+    try:
+        _gitea_jwt(realm, owner, repo, gitea_token, metadata_timeout)
+    except (TransientFailure, RegistryApiError) as error:
+        print("FAILED [2/4] gitea token->JWT exchange at {}: {}".format(
+            realm, error))
+        return 1
+    print("PASS [2/4] gitea token->JWT exchange: 200 + non-empty token "
+          "(service {}, scope repository:{}/{}:pull,push)".format(
+              GITEA_AUTH_SERVICE, owner, repo))
+
+    # [3/4] release-by-tag GET (gitea API; 200 exists, 404 pre-ship ok).
+    release_url = "{}/api/v1/repos/{}/{}/releases/tags/{}".format(
+        gitea_base, owner, repo, urllib.parse.quote(tag, safe=""))
+    try:
+        status, body, _ = _request(
+            "GET", release_url,
+            {"Authorization": "token {}".format(gitea_token)}, None,
+            metadata_timeout)
+    except (TransientFailure, RegistryApiError) as error:
+        print("FAILED [3/4] release-by-tag {}: {}".format(tag, error))
+        return 1
+    if status == 404:
+        print("PASS [3/4] release-by-tag {}: not yet created (404) — "
+              "acceptable pre-ship".format(tag))
+    elif status != 200:
+        print("FAILED [3/4] release-by-tag {}: expected 200 or 404, got "
+              "HTTP {}".format(tag, status))
+        return 1
+    else:
+        try:
+            release = _parse_json(
+                body, "release-by-tag {}".format(tag), status)
+        except RegistryApiError as error:
+            print("FAILED [3/4] release-by-tag {}: {}".format(tag, error))
+            return 1
+        asset_names = _asset_names(release)
+        pyz_assets = [name for name in asset_names
+                      if name.endswith(".pyz")]
+        if tag == SMOKE_G021_TAG and pyz_assets:
+            print("FAILED [3/4] release-by-tag {}: G-021 violation — the "
+                  "phase-0 release must carry no .pyz asset, found: {}"
+                  .format(tag, ", ".join(pyz_assets)))
+            return 1
+        listing = ", ".join(asset_names) if asset_names else "(none)"
+        note = (" — G-021 confirmed: no .pyz asset"
+                if tag == SMOKE_G021_TAG else "")
+        print("PASS [3/4] release-by-tag {}: exists (assets: {}){}".format(
+            tag, listing, note))
+
+    # [4/4] Docker Hub index resolve (anonymous; amd64 drift print).
+    try:
+        hub_bearer = _hub_token(
+            hub_auth, "repository:{}:pull".format(BASE_REPO))
+        index = _fetch_manifest(hub_registry, BASE_REPO, BASE_TAG,
+                                hub_bearer)
+        child_digest = _resolve_amd64_child(index)
+        if not child_digest:
+            raise RegistryApiError(
+                "amd64 child descriptor carries no digest", 500, b"")
+    except (TransientFailure, RegistryApiError) as error:
+        print("FAILED [4/4] hub index resolve {}:{}: {}".format(
+            BASE_REPO, BASE_TAG, error))
+        return 1
+    print("PASS [4/4] hub index resolve {}:{} -> linux/amd64 child {}"
+          .format(BASE_REPO, BASE_TAG, child_digest))
+    return 0
+
+
+def publish(tag, artifact_path, hub_registry, hub_auth, gitea_registry,
+            owner, repo, base_manifest, timeout, env_root):
+    """Run the full publish flow; returns the image provenance dict."""
+    artifact_bytes = _load_artifact(artifact_path)
+    gitea_token = _env_secrets.get_token(root=env_root)
+    name = "{}/{}".format(owner, repo)
+    metadata_timeout = min(METADATA_TIMEOUT_SECONDS, timeout)
+
+    hub_bearer = _hub_token(hub_auth, "repository:{}:pull".format(BASE_REPO))
+    if base_manifest:
+        child_digest = base_manifest
+        child = _fetch_manifest(hub_registry, BASE_REPO, base_manifest,
+                                hub_bearer)
+        index_digest = None
+    else:
+        base = _fetch_manifest(hub_registry, BASE_REPO, BASE_TAG, hub_bearer)
+        index_digest = "sha256:" + hashlib.sha256(
+            json.dumps(base, sort_keys=True,
+                       separators=(",", ":")).encode("utf-8")).hexdigest()
+        child_digest = (_resolve_amd64_child(base)
+                        if "manifests" in base else BASE_TAG)
+        child = _fetch_manifest(hub_registry, BASE_REPO, child_digest,
+                                hub_bearer)
+    child_config, base_layer_descriptors = _child_parts(child)
+    base_config_bytes = _fetch_blob(
+        hub_registry, BASE_REPO, child_config["digest"], hub_bearer,
+        timeout)
+    base_config = _parse_json(base_config_bytes, "base config blob", 200)
+    if not isinstance(base_config, dict):
+        raise RegistryApiError("base config blob is not a JSON object",
+                               500, base_config_bytes)
+
+    jwt = _gitea_jwt("{}/v2/token".format(gitea_registry.rstrip("/")),
+                     owner, repo, gitea_token, metadata_timeout)
+
+    pushed = []
+    pushed.append(_push_blob_if_missing(
+        gitea_registry, name, child_config["digest"], base_config_bytes,
+        jwt, timeout))
+    for descriptor in base_layer_descriptors:
+        if _blob_exists(gitea_registry, name, descriptor["digest"], jwt,
+                        timeout):
+            continue
+        blob = _fetch_blob(hub_registry, BASE_REPO, descriptor["digest"],
+                           hub_bearer, timeout)
+        _upload_blob(gitea_registry, name, descriptor["digest"], blob, jwt,
+                     timeout)
+        pushed.append(True)
+
+    layer_blob, layer_digest, diff_id = build_broker_layer(artifact_bytes)
+    config_blob = build_config_blob(base_config, diff_id, tag)
+    config_digest = "sha256:" + hashlib.sha256(config_blob).hexdigest()
+    _push_blob_if_missing(gitea_registry, name, layer_digest, layer_blob,
+                          jwt, timeout)
+    _push_blob_if_missing(gitea_registry, name, config_digest, config_blob,
+                          jwt, timeout)
+
+    manifest_blob = build_manifest_blob(
+        base_layer_descriptors,
+        _descriptor(layer_digest, len(layer_blob), OCI_LAYER),
+        _descriptor(config_digest, len(config_blob), OCI_CONFIG))
+    manifest_digest = _put_manifest(
+        gitea_registry, name, tag, manifest_blob, jwt, timeout)
+    _put_manifest(gitea_registry, name, "latest", manifest_blob, jwt,
+                  timeout)
+
+    return {
+        "base_index_digest": index_digest,
+        "base_child_digest": child_digest if child_digest != BASE_TAG
+        else "tag-resolved",
+        "base_config_digest": child_config["digest"],
+        "base_layers": [descriptor["digest"]
+                        for descriptor in base_layer_descriptors],
+        "broker_layer_digest": layer_digest,
+        "broker_layer_diff_id": diff_id,
+        "config_digest": config_digest,
+        "manifest_digest": manifest_digest,
+        "blobs_uploaded": sum(1 for flag in pushed if flag),
+    }
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(
+        description="Daemonless OCI image publish (REQ-23, D-019)."
+    )
+    parser.add_argument("tag", help="image tag, e.g. v0.2.3")
+    parser.add_argument("--artifact",
+                        help="path to the built .pyz artifact "
+                             "(default: dist/coreci-broker-v<tag>.pyz)")
+    parser.add_argument("--hub-registry",
+                        default="https://registry-1.docker.io",
+                        help="public base-image registry")
+    parser.add_argument("--hub-auth", default="https://auth.docker.io/token",
+                        help="public registry token service")
+    parser.add_argument("--gitea-registry", default="https://git.coreci.dev",
+                        help="gitea registry base URL")
+    parser.add_argument("--owner", help="gitea owner (default: release.owner)")
+    parser.add_argument("--repo", help="gitea repo (default: release.repo)")
+    parser.add_argument("--base-manifest",
+                        help="base child-manifest digest for exact "
+                             "re-publishes (skips tag resolution)")
+    parser.add_argument("--smoke", action="store_true",
+                        help="G-013 read-only live smoke: four GET-only "
+                             "pre-ship checks, NO publish (see the "
+                             "module docstring)")
+    parser.add_argument("--timeout", type=float, default=BLOB_TIMEOUT_SECONDS,
+                        help="per-request timeout for blob operations "
+                             "(seconds; metadata uses the smaller of this "
+                             "and 30)")
+    parser.add_argument("--env-root",
+                        help="root dir for the sanctioned .env* files "
+                             "(default: repo root)")
+    args = parser.parse_args(argv)
+
+    defaults = load_release_defaults()
+    owner = args.owner or defaults.get("owner")
+    repo = args.repo or defaults.get("repo")
+    if not owner or not repo:
+        print("ERROR: missing gitea target (owner/repo); pass the flags or"
+              " extend the .ciagent/config.json release block",
+              file=sys.stderr)
+        return 2
+    blob_timeout = max(1.0, float(args.timeout))
+    if args.smoke:
+        # G-013 read-only live smoke: dispatched BEFORE any
+        # publish-side work and returning directly — the default
+        # publish flow below can never trigger it.
+        try:
+            return smoke(
+                tag=args.tag, gitea_registry=args.gitea_registry,
+                hub_registry=args.hub_registry, hub_auth=args.hub_auth,
+                owner=owner, repo=repo, timeout=blob_timeout,
+                env_root=args.env_root)
+        except _env_secrets.SecretsError as error:
+            print("ERROR: {}".format(error), file=sys.stderr)
+            return 2
+    artifact_path = args.artifact or os.path.join(
+        "dist", "coreci-broker-{}.pyz".format(args.tag))
+    # Reuse the smaller metadata timeout for the registry auth exchange so
+    # the timeout-abort path stays bounded for every request class.
+    try:
+        provenance = publish(
+            tag=args.tag, artifact_path=artifact_path,
+            hub_registry=args.hub_registry, hub_auth=args.hub_auth,
+            gitea_registry=args.gitea_registry, owner=owner, repo=repo,
+            base_manifest=args.base_manifest, timeout=blob_timeout,
+            env_root=args.env_root)
+    except _env_secrets.SecretsError as error:
+        print("ERROR: {}".format(error), file=sys.stderr)
+        return 2
+    except LocalError as error:
+        print("ERROR: {}".format(error), file=sys.stderr)
+        return 2
+    except (TransientFailure, RegistryApiError) as error:
+        print("ERROR: {}".format(error), file=sys.stderr)
+        registry_host = urllib.parse.urlsplit(
+            args.gitea_registry).netloc or args.gitea_registry
+        print("IMAGE_PENDING {}/{}:{}".format(registry_host, owner, repo,
+                                               args.tag))
+        return 1
+
+    print("image: {}/{}:{} (+ latest)".format(
+        args.gitea_registry.rstrip("/"), owner, repo, args.tag))
+    print("base index: {}".format(
+        provenance["base_index_digest"] or "direct (--base-manifest)"))
+    print("base amd64 manifest: {}".format(provenance["base_child_digest"]))
+    print("base config: {}".format(provenance["base_config_digest"]))
+    print("base layers: {}".format(", ".join(provenance["base_layers"])))
+    print("broker layer: {} (diff_id {})".format(
+        provenance["broker_layer_digest"],
+        provenance["broker_layer_diff_id"]))
+    print("config: {}".format(provenance["config_digest"]))
+    print("manifest: {}".format(provenance["manifest_digest"]))
+    print("blobs uploaded this run: {}".format(provenance["blobs_uploaded"]))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_acceptance_v02.py

@@ -0,0 +1,548 @@
+"""v0.2 acceptance matrix (PLAN P3-01; REQ-21/REQ-22/REQ-23 acceptance).
+
+Consolidates the milestone's acceptance as thin, high-value assertions that
+the deliverable holds end-to-end. The deep per-feature logic stays in its own
+files — test_build_cli, test_cli_packaging, test_publish_image,
+test_release_upload, test_dockerfile_consistency — which run green in the
+same cumulative suite (``python3 -m unittest discover -s tests``; suite
+greenness is the runner's job, asserted by nobody in here). This file does
+NOT re-run their cases; the stub roundtrips are smoke-asserted instead via
+module presence + runnable TestSuite sizes (defaultTestLoader).
+
+The matrix (binding order):
+1. boot parity through the artifact — the built ``.pyz`` is booted as a real
+   subprocess child (``[sys.executable, artifact]``, never a bare
+   "python3") with an ephemeral bind + temp state, and v0.1 acceptance
+   parity must hold through the packaged path: admin pass printed exactly
+   once, ``broker.db`` + ``broker.audit.log`` 0600 in the state dir,
+   ``/ui/login`` + ``/ui/dashboard`` 200, ``/static/coreci_broker_client.py``
+   byte-exact vs the repo file, one admin-login + agent-register +
+   state-put/get + secret-read flow (v0.1 request patterns), second boot
+   silent (asserted only after terminate + full drain — G-010);
+2. reproducibility — two consecutive same-environment builds produce
+   byte-identical artifacts (identical sha256, G-009);
+3. self-containment — the archive contains ONLY the staging allowlist
+   (broker + client + UI pages + shim + ``_cli_version.py``): no ``.env``,
+   no ``dist/``, no state, no tests;
+4. stub publish roundtrip + stub release upload green — their test modules
+   import cleanly and expose runnable TestSuites of the expected size;
+5. broker zero-delta — ``coreci_broker.py`` unchanged at its v0.1.5 count
+   of 1390 lines (AC14 under the D-016..D-018 cap of 1400) and carrying no
+   v0.2 packaging references (``_cli_version`` / the shim / the shim's
+   materialized-asset dir — frozen means untouched); plus the AC14
+   no-touch clause machine-checked against the ``v0.1.5`` tag when the
+   working tree carries it (skipped otherwise);
+6. Dockerfile/.dockerignore consistency — module present with its full
+   TestCase set (it runs in the same suite).
+
+The build happens ONCE per class (setUpClass) and every artifact/boot is
+reused across methods where safe; the boot tests dominate and the whole
+file is budgeted to stay well under 30s. Every child is terminated +
+drained exactly once (idempotent ``stop`` via ``addCleanup``); every temp
+dir is ``addCleanup``-registered. Builds use the ``scripts/build_cli.py``
+API (never a shell path) into ``TemporaryDirectory`` dist dirs — no
+committed ``dist/``, no state, no live-forge contact.
+"""
+
+import hashlib
+import importlib.util
+import json
+import os
+import select
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+import urllib.error
+import urllib.request
+import uuid as uuidlib
+import zipfile
+
+REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir))
+TESTS_DIR = os.path.dirname(os.path.abspath(__file__))
+BUILD_CLI_PATH = os.path.join(REPO, "scripts", "build_cli.py")
+BROKER_SOURCE_PATH = os.path.join(REPO, "coreci_broker.py")
+REPO_CLIENT_PATH = os.path.join(REPO, "coreci_broker_client.py")
+VERSION = "v0.2.3"  # the milestone release this acceptance gates (P3-05 ship)
+
+ADMIN_USER = "admin"
+ADMIN_PASS_PREFIX = "BROKER_ADMIN_PASS="
+LISTENING_PREFIX = "coreci-broker listening on http://"
+STATE_FILE_NAMES = (".env", "broker.db", "broker.audit.log")
+
+LOGIN_PATH = "/api/broker/auth/login"
+APPS_PATH = "/api/broker/app_registrations"
+USERS_PATH = "/api/broker/user_allowlist"
+REGISTER_PATH = "/api/broker/agents/register"
+AGENT_STATE_PATH = "/api/broker/agents/proxmox/state"
+SECRETS_PATH = "/api/broker/secrets/coreci/proxmox"
+CLIENT_STATIC_PATH = "/static/coreci_broker_client.py"
+UI_LOGIN_PATH = "/ui/login"
+UI_DASHBOARD_PATH = "/ui/dashboard"
+
+PROXMOX_PATH = "coreci/proxmox"
+PROXMOX_SECRET = {"url": "https://pve.coreci.dev:8006", "token": "pve-tok"}
+
+BOOT_DEADLINE_SECONDS = 30.0
+POLL_SECONDS = 0.1
+DRAIN_TIMEOUT_SECONDS = 20.0
+HTTP_TIMEOUT_SECONDS = 10
+GIT_TIMEOUT_SECONDS = 30
+
+# Frozen-broker sources (PLAN: no task edits these in v0.2) — used only by
+# the AC14 no-touch clause check, never modified.
+FROZEN_SOURCES = ("coreci_broker.py", "coreci_broker_client.py",
+                  "coreci_broker_ui", "reference_tools")
+# v0.2 packaging artifacts the frozen broker must know nothing about.
+FROZEN_MARKER_STRINGS = ("_cli_version", "cli/__main__", "coreci_broker_assets")
+
+# The v0.2 test surface consolidated here — presence/count floors only
+# (the roundtrips themselves run green in their own files, same suite).
+STUB_MODULE_MIN_CASES = {
+    "test_publish_image.py": 9,        # stub publish roundtrip
+    "test_release_upload.py": 15,      # stub release upload + env-secrets
+    "test_dockerfile_consistency.py": 7,
+}
+
+
+def _load_build_cli():
+    spec = importlib.util.spec_from_file_location(
+        "acceptance_build_cli", BUILD_CLI_PATH)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+build_cli = _load_build_cli()
+
+
+def _sha256(path):
+    digest = hashlib.sha256()
+    with open(path, "rb") as handle:
+        for chunk in iter(lambda: handle.read(65536), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+class _BrokerChild:
+    """A subprocess child booted from the built zipapp artifact.
+
+    The merged stdout/stderr pipe is pumped via ``select`` + raw ``os.read``
+    under an explicit deadline until the broker announces its ephemeral
+    listen address. ``stop()`` is the G-010 kill/drain protocol:
+    ``terminate()`` then a full ``communicate()`` drain (kill fallback on
+    drain timeout). It is idempotent and registered with ``addCleanup`` at
+    construction, so no child leaks even when an assertion trips mid-test.
+    Silence and pass-once assertions are made ONLY on ``full_output()``
+    after ``stop()``.
+    """
+
+    def __init__(self, testcase, artifact, env, cwd):
+        self.testcase = testcase
+        self.artifact = artifact
+        self.proc = subprocess.Popen(
+            [sys.executable, artifact],
+            env=env, cwd=cwd,
+            stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+        )
+        self.lines = []
+        self.partial = b""
+        self.drained = b""
+        self.base_url = None
+        self._eof = False
+        self._stopped = False
+        testcase.addCleanup(self.stop)
+        self._await_listening(BOOT_DEADLINE_SECONDS)
+
+    def _pump(self, timeout):
+        if self._eof:
+            return False
+        fd = self.proc.stdout.fileno()
+        ready, _, _ = select.select([fd], [], [], timeout)
+        if not ready:
+            return True
+        try:
+            chunk = os.read(fd, 4096)
+        except OSError:
+            chunk = b""
+        if not chunk:
+            self._eof = True
+            return False
+        self.partial += chunk
+        while b"\n" in self.partial:
+            raw, self.partial = self.partial.split(b"\n", 1)
+            self.lines.append(raw.decode("utf-8", "replace"))
+        return True
+
+    def _await_listening(self, deadline_seconds):
+        deadline = time.monotonic() + deadline_seconds
+        cursor = 0
+        while time.monotonic() < deadline:
+            alive = self._pump(POLL_SECONDS)
+            while cursor < len(self.lines):
+                line = self.lines[cursor]
+                cursor += 1
+                if line.startswith(LISTENING_PREFIX):
+                    self.base_url = line.split(" on ", 1)[1].strip()
+                    return
+            if not alive:
+                break
+        self.stop()
+        self.testcase.fail(
+            "packaged broker did not announce %r within %.0fs "
+            "(artifact=%s exit=%s)\n--- captured child output ---\n%s"
+            % (LISTENING_PREFIX, deadline_seconds, self.artifact,
+               self.proc.returncode, self.full_output())
+        )
+
+    def stop(self):
+        if self._stopped:
+            return
+        self._stopped = True
+        if self.proc.poll() is None:
+            self.proc.terminate()
+        try:
+            remainder, _ = self.proc.communicate(timeout=DRAIN_TIMEOUT_SECONDS)
+        except subprocess.TimeoutExpired:
+            self.proc.kill()
+            remainder, _ = self.proc.communicate()
+        self.drained = remainder or b""
+
+    def full_output(self):
+        text = "\n".join(self.lines)
+        tail = self.partial + self.drained
+        if tail:
+            text += "\n" + tail.decode("utf-8", "replace")
+        return text
+
+    def printed_admin_pass(self):
+        values = [
+            line.split("=", 1)[1]
+            for line in self.lines
+            if line.startswith(ADMIN_PASS_PREFIX)
+        ]
+        self.testcase.assertEqual(
+            1, len(values),
+            "expected exactly one %s line, got: %r" % (ADMIN_PASS_PREFIX, values)
+        )
+        return values[0]
+
+
+class AcceptanceV02Tests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.build_root = tempfile.TemporaryDirectory(prefix="coreci-acc-v02-")
+        cls.addClassCleanup(cls.build_root.cleanup)
+        cls.artifact = build_cli.build(
+            VERSION, repo=REPO, dist_dir=cls.build_root.name
+        )
+        cls.artifact_sha256 = _sha256(cls.artifact)
+
+    # -- harness helpers --------------------------------------------------
+
+    def _temp_dir(self):
+        handle = tempfile.TemporaryDirectory(prefix="coreci-acc-v02-")
+        self.addCleanup(handle.cleanup)
+        return handle.name
+
+    def _child_env(self, **overrides):
+        """A child env with no inherited BROKER_* keys, then overrides."""
+        env = {
+            key: value
+            for key, value in os.environ.items()
+            if not key.startswith("BROKER_")
+        }
+        env.update(overrides)
+        return env
+
+    def _state_env(self, state_dir):
+        """Ephemeral bind + temp state + a seeded file creds source."""
+        creds_path = os.path.join(state_dir, "creds.json")
+        with open(creds_path, "w", encoding="utf-8") as handle:
+            json.dump({PROXMOX_PATH: PROXMOX_SECRET}, handle)
+        return self._child_env(
+            BROKER_BIND="127.0.0.1:0",
+            BROKER_ENV_PATH=os.path.join(state_dir, ".env"),
+            BROKER_DB_PATH=os.path.join(state_dir, "broker.db"),
+            BROKER_AUDIT_LOG_PATH=os.path.join(state_dir, "broker.audit.log"),
+            BROKER_CREDS_SOURCE="file",
+            BROKER_CREDS_FILE=creds_path,
+        )
+
+    def _boot(self, state_dir):
+        """Boot the artifact against state_dir; CWD is a separate temp dir."""
+        return _BrokerChild(
+            self, self.artifact, self._state_env(state_dir), self._temp_dir()
+        )
+
+    def _request(self, base, method, path, payload=None, headers=None):
+        data = None if payload is None else json.dumps(payload).encode("utf-8")
+        request = urllib.request.Request(base + path, data=data, method=method)
+        for key, value in (headers or {}).items():
+            request.add_header(key, value)
+        try:
+            with urllib.request.urlopen(
+                request, timeout=HTTP_TIMEOUT_SECONDS
+            ) as response:
+                return response.status, response.read(), response.headers
+        except urllib.error.HTTPError as error:
+            return error.code, error.read(), error.headers
+
+    def _request_json(self, base, method, path, payload=None, headers=None):
+        status, body, response_headers = self._request(
+            base, method, path, payload, headers
+        )
+        return status, json.loads(body.decode("utf-8")), response_headers
+
+    def _admin_login(self, base, password):
+        """v0.1 pattern: POST login, assert the broker_session cookie."""
+        status, body, headers = self._request_json(
+            base, "POST", LOGIN_PATH,
+            {"username": ADMIN_USER, "password": password},
+        )
+        self.assertEqual(200, status)
+        self.assertEqual({"ok": True, "username": ADMIN_USER}, body)
+        raw_cookie = headers.get("Set-Cookie")
+        self.assertTrue(raw_cookie)
+        self.assertTrue(raw_cookie.startswith("broker_session="))
+        self.assertIn("HttpOnly", raw_cookie)
+        self.assertIn("SameSite=Lax", raw_cookie)
+        return {"Cookie": raw_cookie.split(";", 1)[0]}
+
+    def _load_test_module(self, file_name):
+        """Import a sibling test module cleanly (by path, any invocation)."""
+        path = os.path.join(TESTS_DIR, file_name)
+        stem = os.path.splitext(file_name)[0]
+        spec = importlib.util.spec_from_file_location(
+            "acceptance_probe_" + stem, path)
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        return module
+
+    def _runnable_case_count(self, module):
+        """Flatten defaultTestLoader's suite, counting runnable TestCases."""
+        suite = unittest.defaultTestLoader.loadTestsFromModule(module)
+        stack, count = [suite], 0
+        while stack:
+            item = stack.pop()
+            if isinstance(item, unittest.TestSuite):
+                stack.extend(item)
+            else:
+                self.assertIsInstance(item, unittest.TestCase)
+                count += 1
+        return count
+
+    # -- matrix 1: boot parity through the artifact ----------------------
+
+    def test_boot_parity_through_artifact_second_boot_silent(self):
+        state_dir = self._temp_dir()
+        child = self._boot(state_dir)
+        base = child.base_url
+        printed_pass = child.printed_admin_pass()
+
+        # liveness + full UI served from the packaged artifact
+        status, body, _ = self._request_json(base, "GET", "/healthz")
+        self.assertEqual(200, status)
+        self.assertEqual({"ok": True}, body)
+        for ui_path in (UI_LOGIN_PATH, UI_DASHBOARD_PATH):
+            status, body, headers = self._request(base, "GET", ui_path)
+            self.assertEqual(200, status, ui_path)
+            self.assertTrue(
+                headers.get("Content-Type", "").startswith("text/html"),
+                ui_path,
+            )
+        # the tool-author surface: client module byte-exact vs the repo file
+        status, served, _ = self._request(base, "GET", CLIENT_STATIC_PATH)
+        self.assertEqual(200, status)
+        with open(REPO_CLIENT_PATH, "rb") as handle:
+            self.assertEqual(handle.read(), served)
+
+        # one consolidated v0.1 request-pattern flow:
+        # admin login -> app + user allowlist -> agent register -> state
+        # put/get -> secret read
+        cookie = self._admin_login(base, printed_pass)
+        app_uuid = str(uuidlib.uuid4())
+        status, body, _ = self._request_json(
+            base, "POST", APPS_PATH,
+            {"uuid": app_uuid, "name": "acceptance-tool"}, headers=cookie,
+        )
+        self.assertEqual(201, status)
+        self.assertEqual({"uuid": app_uuid, "status": "active"}, body)
+        status, body, _ = self._request_json(
+            base, "POST", USERS_PATH,
+            {"user_id": "ceo", "email": "ceo@coreci.dev"}, headers=cookie,
+        )
+        self.assertEqual(201, status)
+        status, body, _ = self._request_json(
+            base, "POST", REGISTER_PATH,
+            {"agent_id": "proxmox", "app_registration_uuid": app_uuid,
+             "user_id": "ceo", "tags": ["coreci"]},
+        )
+        self.assertEqual(201, status)
+        app_header = {"Authorization": "App %s %s" % (app_uuid, body["token"])}
+        state_document = {"milestone": "v0.2", "matrix": "acceptance"}
+        status, body, _ = self._request_json(
+            base, "PUT", AGENT_STATE_PATH, state_document, headers=app_header
+        )
+        self.assertEqual(200, status)
+        self.assertEqual({"ok": True}, body)
+        status, body, _ = self._request_json(
+            base, "GET", AGENT_STATE_PATH, headers=app_header
+        )
+        self.assertEqual(200, status)
+        self.assertEqual(state_document, body)
+        status, body, _ = self._request_json(
+            base, "GET", SECRETS_PATH, headers=app_header
+        )
+        self.assertEqual(200, status)
+        self.assertEqual({"path", "data", "version", "updated_at"}, set(body))
+        self.assertEqual(PROXMOX_PATH, body["path"])
+        self.assertEqual(PROXMOX_SECRET, body["data"])
+
+        # v0.1 state-file parity: 0600 in the state dir
+        for name in STATE_FILE_NAMES:
+            path = os.path.join(state_dir, name)
+            self.assertTrue(os.path.isfile(path), path)
+            self.assertEqual(0o600, os.stat(path).st_mode & 0o777, path)
+
+        # pass printed exactly once + pepper never printed — asserted on
+        # the FULLY drained output after terminate (G-010)
+        pepper = None
+        with open(os.path.join(state_dir, ".env"), "r", encoding="utf-8") as handle:
+            for line in handle:
+                if line.startswith("BROKER_INSTANCE_PEPPER="):
+                    pepper = line.split("=", 1)[1].strip()
+        self.assertIsNotNone(pepper)
+        child.stop()
+        first_output = child.full_output()
+        self.assertEqual(1, first_output.count(ADMIN_PASS_PREFIX), first_output)
+        self.assertEqual(1, first_output.count(printed_pass))
+        self.assertNotIn(pepper, first_output)
+
+        # SECOND boot on the same state dir: silent after terminate + drain
+        second = self._boot(state_dir)
+        # the SAME pass still logs in: state persisted across the reboot
+        self._admin_login(second.base_url, printed_pass)
+        second.stop()
+        second_output = second.full_output()
+        self.assertIn(LISTENING_PREFIX, second_output)
+        self.assertNotIn(ADMIN_PASS_PREFIX, second_output)
+        self.assertNotIn(printed_pass, second_output)
+
+    # -- matrix 2: reproducibility ----------------------------------------
+
+    def test_reproducibility_two_consecutive_builds_identical_sha256(self):
+        digests = []
+        for _ in range(2):
+            with tempfile.TemporaryDirectory(prefix="coreci-acc-repro-") as dist_dir:
+                artifact = build_cli.build(
+                    VERSION, repo=REPO, dist_dir=dist_dir
+                )
+                digests.append(_sha256(artifact))
+        self.assertNotEqual(digests[0], "0" * 64)
+        self.assertEqual(digests[0], digests[1])
+        # and the shared setUpClass build matches too (same environment)
+        self.assertEqual(self.artifact_sha256, digests[0])
+        print("v0.2 acceptance two-build sha256: {}".format(digests[0]))
+
+    # -- matrix 3: self-containment ---------------------------------------
+
+    def test_self_containment_archive_is_exactly_the_allowlist(self):
+        ui_root = os.path.join(REPO, "coreci_broker_ui")
+        expected = {
+            "__main__.py",
+            "_cli_version.py",
+            "coreci_broker.py",
+            "coreci_broker_client.py",
+            "coreci_broker_ui/",
+        }
+        expected.update(
+            "coreci_broker_ui/%s" % name
+            for name in os.listdir(ui_root)
+            if os.path.isfile(os.path.join(ui_root, name))
+        )
+        with zipfile.ZipFile(self.artifact) as archive:
+            names = set(archive.namelist())
+        # ONLY the allowlist members — nothing more, nothing less
+        self.assertEqual(expected, names)
+        # no .env, no dist/, no state, no tests, no bytecode can ride along
+        for name in names:
+            for forbidden in (".env", "dist", "broker.db", "broker.audit.log",
+                             "test", "__pycache__", ".pyc"):
+                self.assertNotIn(forbidden, name, name)
+        # per-release artifact name (D-020) for the milestone release
+        self.assertEqual(
+            build_cli.artifact_name(VERSION),
+            os.path.basename(self.artifact),
+        )
+
+    # -- matrix 4: stub publish roundtrip + stub release upload green ------
+
+    def test_stub_publish_roundtrip_module_green_in_suite(self):
+        module = self._load_test_module("test_publish_image.py")
+        self.assertTrue(hasattr(module, "PublishImageTests"))
+        count = self._runnable_case_count(module)
+        self.assertGreaterEqual(
+            count, STUB_MODULE_MIN_CASES["test_publish_image.py"]
+        )
+
+    def test_stub_release_upload_module_green_in_suite(self):
+        module = self._load_test_module("test_release_upload.py")
+        self.assertTrue(hasattr(module, "ReleaseUploadTests"))
+        self.assertTrue(hasattr(module, "EnvSecretsLoaderTests"))
+        count = self._runnable_case_count(module)
+        self.assertGreaterEqual(
+            count, STUB_MODULE_MIN_CASES["test_release_upload.py"]
+        )
+
+    # -- matrix 6: Dockerfile/.dockerignore consistency ---------------------
+
+    def test_dockerfile_consistency_module_green_in_suite(self):
+        module = self._load_test_module("test_dockerfile_consistency.py")
+        self.assertTrue(hasattr(module, "DockerfileConsistencyTests"))
+        count = self._runnable_case_count(module)
+        self.assertGreaterEqual(
+            count, STUB_MODULE_MIN_CASES["test_dockerfile_consistency.py"]
+        )
+
+    # -- matrix 5: broker zero-delta (AC14, frozen means untouched) --------
+
+    def test_broker_zero_delta_1390_lines_no_packaging_references(self):
+        with open(BROKER_SOURCE_PATH, "r", encoding="utf-8") as handle:
+            line_count = sum(1 for _ in handle)
+        # exact v0.1.5 count, under the D-016..D-018 cap of 1400 (AC14)
+        self.assertEqual(1390, line_count)
+        self.assertLess(line_count, 1400)
+        with open(BROKER_SOURCE_PATH, "r", encoding="utf-8") as handle:
+            source = handle.read()
+        # frozen means untouched: no v0.2 packaging reference leaked into it
+        for marker in FROZEN_MARKER_STRINGS:
+            self.assertNotIn(marker, source)
+
+    def test_broker_untouched_by_any_commit_since_v0_1_5(self):
+        """The AC14 no-touch clause, machine-checked when the tag exists."""
+        try:
+            verify = subprocess.run(
+                ["git", "rev-parse", "-q", "--verify", "v0.1.5^{commit}"],
+                cwd=REPO, capture_output=True, text=True,
+                timeout=GIT_TIMEOUT_SECONDS,
+            )
+        except (OSError, subprocess.TimeoutExpired):
+            self.skipTest("git unavailable in this environment")
+        if verify.returncode != 0:
+            self.skipTest("v0.1.5 tag not present in this working tree")
+        log = subprocess.run(
+            ["git", "log", "--oneline", "v0.1.5..HEAD", "--"] + list(FROZEN_SOURCES),
+            cwd=REPO, capture_output=True, text=True,
+            timeout=GIT_TIMEOUT_SECONDS,
+        )
+        self.assertEqual(0, log.returncode, log.stderr)
+        self.assertEqual(
+            "", log.stdout.strip(),
+            "v0.2 commits touched frozen broker sources:\n" + log.stdout
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_dockerfile_consistency.py

@@ -0,0 +1,264 @@
+"""Dockerfile consistency tests (G-016 — equivalence made testable).
+
+The Dockerfile documents the docker-user path; the daemonless client
+(``scripts/publish_image.py``) is the pipeline's source of truth for
+the published image. Byte-identity is NOT required — these tests pin the
+equivalence contract: same staged sources (member-set parity — the
+docker-staged layout must equal build_cli's staging EXACTLY, or the
+packaged shim materializes an empty UI dir and /ui/* 404s in
+docker-built images; P3-03/CWE-636), same zipapp invocation, same
+entrypoint/working-dir, same pinned base, and a hygienic build context.
+"""
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+import zipfile
+
+_SCRIPTS_DIR = os.path.join(os.path.dirname(os.path.dirname(
+    os.path.abspath(__file__))), "scripts")
+if _SCRIPTS_DIR not in sys.path:
+    sys.path.insert(0, _SCRIPTS_DIR)
+
+import build_cli  # noqa: E402  (scripts/ dir on sys.path)
+import publish_image  # noqa: E402  (scripts/ dir on sys.path)
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+APP_SRC = "/app/src/"
+DOCKERFILE_IGNORE_PATTERNS = (
+    ".git", ".env", ".env.*", "data/", "*.db", "*.log", "dist/",
+    "__pycache__/", "tests/", ".ciagent/",
+    "**/__pycache__/", "**/*.pyc",
+)
+
+
+def read(name):
+    with open(os.path.join(REPO, name), "r", encoding="utf-8") as handle:
+        return handle.read()
+
+
+def _tree_members(prefix, root):
+    """Zipapp member names for the tree at ``root`` staged under
+    ``prefix`` (``""`` for the archive root; the prefix itself is
+    included — zipapp writes directory entries)."""
+    members = {prefix} if prefix else set()
+    for current, dirs, files in os.walk(root):
+        relative = os.path.relpath(current, root)
+        base = prefix + ("" if relative == "." else relative + "/")
+        for name in dirs:
+            members.add(base + name + "/")
+        for name in files:
+            members.add(base + name)
+    return members
+
+
+def _build_cli_staged_members():
+    """The member set build_cli stages, derived via its constants:
+    ROOT_SOURCES at the root, UI under the coreci_broker_ui/ prefix,
+    cli/__main__.py staged as __main__.py, _cli_version.py generated."""
+    members = set(build_cli.ROOT_SOURCES)
+    members.add(build_cli.STAGED_SHIM)
+    members.add(build_cli.VERSION_MODULE)
+    members |= _tree_members(
+        build_cli.UI_SOURCE_DIR + "/",
+        os.path.join(REPO, build_cli.UI_SOURCE_DIR))
+    return members
+
+
+def _run_zipapp(staged_dir, artifact):
+    """The Dockerfile's packing invocation (zipapp writes dir entries;
+    the member set is interpreter-stable across 3.x)."""
+    completed = subprocess.run(
+        [sys.executable, "-m", "zipapp", staged_dir,
+         "-o", artifact,
+         "--python", "/usr/bin/env python3", "--compress"],
+        capture_output=True, text=True)
+    if completed.returncode != 0:
+        raise AssertionError(
+            "zipapp failed ({}): {}".format(
+                completed.returncode,
+                (completed.stderr or completed.stdout).strip()))
+
+
+class DockerfileConsistencyTests(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        cls.dockerfile = read("Dockerfile")
+        cls.dockerignore = read(".dockerignore")
+        # Join backslash continuations into logical stanzas; skip comments.
+        stanzas = []
+        for line in cls.dockerfile.splitlines():
+            stripped = line.strip()
+            if not stripped or stripped.startswith("#"):
+                continue
+            if stanzas and stanzas[-1].endswith("\\"):
+                stanzas[-1] = stanzas[-1].rstrip("\\").rstrip() + " " + stripped
+                continue
+            stanzas.append(stripped)
+        cls.stanzas = stanzas
+        cls.copy_lines = [s for s in stanzas if s.startswith("COPY ")]
+        cls.zipapp_lines = [s for s in stanzas
+                            if "zipapp" in s and s.startswith("RUN ")]
+        cls.version_lines = [s for s in stanzas
+                             if "_cli_version.py" in s
+                             and s.startswith("RUN ")]
+
+    def _copy_specs(self):
+        """[(sources, dest)] per COPY stanza, dest verified to stage
+        under /app/src/."""
+        specs = []
+        for line in self.copy_lines:
+            tokens = line.split()[1:]
+            sources, dest = tokens[:-1], tokens[-1]
+            self.assertTrue(dest.startswith(APP_SRC),
+                            "COPY destination {} must stage under /app/src/"
+                            .format(dest))
+            specs.append((sources, dest))
+        return specs
+
+    def _docker_staged_members(self):
+        """The member set the COPY stanzas stage under /app/src/ (Docker
+        semantics: directory CONTENTS land flattened INSIDE the
+        destination dir; a dest without trailing slash is a file rename),
+        plus the RUN-generated _cli_version.py."""
+        members = set()
+        for sources, dest in self._copy_specs():
+            if dest.endswith("/"):
+                prefix = dest[len(APP_SRC):]
+                for source in sources:
+                    path = os.path.join(REPO, source.rstrip("/"))
+                    self.assertTrue(os.path.exists(path),
+                                    "COPY source missing in the repo: {}"
+                                    .format(source))
+                    if os.path.isdir(path):
+                        members |= _tree_members(prefix, path)
+                    else:
+                        members.add(prefix + os.path.basename(source))
+            else:
+                # file-to-file rename (mirrors build_cli's STAGED_SHIM
+                # staging — zipapp requires __main__.py at the root)
+                self.assertEqual(len(sources), 1,
+                                 "file-rename COPY stages exactly one "
+                                 "source: {}".format(dest))
+                members.add(dest[len(APP_SRC):])
+        members.add(build_cli.VERSION_MODULE)  # the RUN printf stanza
+        return members
+
+    def test_from_pins_base(self):
+        from_lines = [line for line in self.dockerfile.splitlines()
+                      if line.startswith("FROM ")]
+        self.assertEqual(len(from_lines), 1)
+        self.assertEqual(from_lines[0].split()[1], "python:3.11-slim")
+
+    def test_copy_staged_member_set_matches_build_cli_staging(self):
+        """The COPY stanzas stage EXACTLY build_cli's member set — set
+        parity, no extras, no missing (a flattened UI copy or a missing
+        root __main__.py breaks the packaged boot; P3-03/CWE-636)."""
+        self.assertEqual(self._docker_staged_members(),
+                         _build_cli_staged_members(),
+                         "docker-staged member set == build_cli-staged "
+                         "member set")
+        for line in self.copy_lines:
+            self.assertNotIn("dist/", line,
+                             "no pre-built artifact in the build context")
+
+    def test_docker_layout_simulation_member_set_parity(self):
+        """Docker-layout simulation gate (P3-03/CWE-636): apply the
+        Dockerfile COPY semantics to the real repo sources, add the
+        generated _cli_version.py, zipapp-pack, and the archive's member
+        set must EQUAL the build_cli-built artifact's member set."""
+        with tempfile.TemporaryDirectory(prefix="docker-layout-") as tmp:
+            staged = os.path.join(tmp, "src")
+            os.makedirs(staged)
+            for sources, dest in self._copy_specs():
+                if dest.endswith("/"):
+                    target = os.path.join(staged, dest[len(APP_SRC):])
+                    os.makedirs(target, exist_ok=True)
+                    for source in sources:
+                        path = os.path.join(REPO, source.rstrip("/"))
+                        if os.path.isdir(path):
+                            # directory CONTENTS land inside the dest dir
+                            shutil.copytree(path, target,
+                                            dirs_exist_ok=True)
+                        else:
+                            shutil.copy2(
+                                path,
+                                os.path.join(target,
+                                             os.path.basename(source)))
+                else:
+                    self.assertEqual(len(sources), 1)
+                    shutil.copy2(os.path.join(REPO, sources[0]),
+                                 os.path.join(staged, dest[len(APP_SRC):]))
+            with open(os.path.join(staged, build_cli.VERSION_MODULE), "w",
+                      encoding="utf-8") as handle:
+                handle.write('VERSION = "{}"\n'.format("v0.2-layout-sim"))
+            docker_artifact = os.path.join(tmp, "docker-layout.pyz")
+            _run_zipapp(staged, docker_artifact)
+            with zipfile.ZipFile(docker_artifact) as archive:
+                docker_members = set(archive.namelist())
+        with tempfile.TemporaryDirectory(prefix="build-cli-ref-") as dist:
+            reference = build_cli.build("v0.2-layout-sim", repo=REPO,
+                                        dist_dir=dist)
+            with zipfile.ZipFile(reference) as archive:
+                build_members = set(archive.namelist())
+        self.assertEqual(docker_members, build_members,
+                         "docker-layout archive members == build_cli "
+                         "artifact members")
+
+    def test_zipapp_invocation_flags_match_build_script(self):
+        """Same --python interpreter and --compress flag as build_cli."""
+        self.assertEqual(len(self.zipapp_lines), 1)
+        stanza = self.zipapp_lines[0]
+        self.assertIn('--python "/usr/bin/env python3"', stanza)
+        self.assertIn("--compress", stanza)
+        self.assertIn("-m zipapp", stanza)
+        self.assertIn("/app/src", stanza)
+        self.assertIn("/app/coreci-broker.pyz", stanza)
+
+    def test_version_module_format_parity(self):
+        """ARG VERSION feeds the same VERSION = "<value>" format."""
+        self.assertEqual(len(self.version_lines), 1)
+        self.assertIn('VERSION = "%s"', self.version_lines[0])
+        self.assertEqual(build_cli.VERSION_MODULE, "_cli_version.py")
+        self.assertTrue(any(s.startswith("ARG VERSION=") for s in self.stanzas),
+                        "ARG VERSION provides the version")
+
+    def test_entrypoint_and_workdir_match_published_config(self):
+        """ENTRYPOINT/WORKDIR equal the daemonless client's pushed config."""
+        base_config = json.loads('{"config": {}, "rootfs": {}}')
+        blob = json.loads(publish_image.build_config_blob(
+            base_config, "sha256:" + "0" * 64, "v-parity"))
+        entrypoint = blob["config"]["Entrypoint"]
+        working_dir = blob["config"]["WorkingDir"]
+        docker_entrypoint = [line for line in self.dockerfile.splitlines()
+                             if line.startswith("ENTRYPOINT ")]
+        docker_workdir = [line for line in self.dockerfile.splitlines()
+                          if line.startswith("WORKDIR ")]
+        self.assertEqual(len(docker_entrypoint), 1)
+        self.assertEqual(json.loads(
+            docker_entrypoint[0].split(" ", 1)[1]), entrypoint)
+        self.assertEqual(len(docker_workdir), 1)
+        self.assertEqual(docker_workdir[0].split(" ", 1)[1], working_dir)
+        self.assertEqual(working_dir, "/state")
+
+    def test_dockerignore_hygiene(self):
+        entries = {line.strip() for line in self.dockerignore.splitlines()
+                   if line.strip() and not line.strip().startswith("#")}
+        for pattern in DOCKERFILE_IGNORE_PATTERNS:
+            self.assertIn(pattern, entries,
+                          ".dockerignore must exclude {}".format(pattern))
+
+    def test_no_absolute_source_copies(self):
+        for line in self.copy_lines:
+            for token in line.split()[1:-1]:
+                self.assertFalse(token.startswith("/"),
+                                "build context sources are repo-relative")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_publish_image.py

@@ -0,0 +1,799 @@
+"""Stub-registry tests for scripts/publish_image.py (REQ-23, G-019).
+
+Two local stub servers stand in for the public image registry (anonymous
+pull) and the gitea container registry (JWT auth + blob uploads +
+manifest PUTs). No network access; tiny in-memory fixtures; every
+spawned server is cleaned via addCleanup.
+"""
+
+import base64
+import hashlib
+import io
+import json
+import os
+import sys
+import tempfile
+import threading
+import time
+import unittest
+import urllib.parse
+from contextlib import redirect_stderr, redirect_stdout
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+_SCRIPTS_DIR = os.path.join(os.path.dirname(os.path.dirname(
+    os.path.abspath(__file__))), "scripts")
+if _SCRIPTS_DIR not in sys.path:
+    sys.path.insert(0, _SCRIPTS_DIR)
+
+import publish_image  # noqa: E402  (scripts/ dir on sys.path)
+
+OCI_MANIFEST = "application/vnd.oci.image.manifest.v1+json"
+OCI_INDEX = "application/vnd.oci.image.index.v1+json"
+OCI_CONFIG = "application/vnd.oci.image.config.v1+json"
+OCI_LAYER = "application/vnd.oci.image.layer.v1.tar+gzip"
+OWNER = "testowner"
+REPO = "testrepo"
+NAME = "{}/{}".format(OWNER, REPO)
+
+
+def _sha(data):
+    return "sha256:" + hashlib.sha256(data).hexdigest()
+
+
+def _dump(payload):
+    return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
+
+
+class HubFixtures:
+    """A tiny synthetic base image: index -> amd64/arm64 children."""
+
+    def __init__(self):
+        self.layer0 = b"base layer zero payload\x00" + b"\xa5" * 64
+        self.layer1 = b"base layer one payload" * 7
+        self.config = _dump({
+            "architecture": "amd64",
+            "os": "linux",
+            "config": {"Cmd": ["python3"], "Env": ["PATH=/usr/local/bin"]},
+            "rootfs": {"type": "layers",
+                       "diff_ids": [_sha(self.layer0), _sha(self.layer1)]},
+        })
+        self.child = {
+            "schemaVersion": 2,
+            "mediaType": OCI_MANIFEST,
+            "config": {"mediaType": OCI_CONFIG,
+                       "digest": _sha(self.config),
+                       "size": len(self.config)},
+            "layers": [
+                {"mediaType": OCI_LAYER, "digest": _sha(self.layer0),
+                 "size": len(self.layer0)},
+                {"mediaType": OCI_LAYER, "digest": _sha(self.layer1),
+                 "size": len(self.layer1)},
+            ],
+        }
+        self.child_digest = _sha(_dump(self.child))
+        arm64_child = dict(self.child)
+        arm64_child["annotations"] = {"test-fixture": "arm64"}
+        self.arm64_child_digest = _sha(_dump(arm64_child))
+        self.index = {
+            "schemaVersion": 2,
+            "mediaType": OCI_INDEX,
+            "manifests": [
+                {"mediaType": OCI_MANIFEST, "digest": self.child_digest,
+                 "size": len(_dump(self.child)),
+                 "platform": {"os": "linux", "architecture": "amd64"}},
+                {"mediaType": OCI_MANIFEST,
+                 "digest": self.arm64_child_digest,
+                 "size": len(_dump(arm64_child)),
+                 "platform": {"os": "linux", "architecture": "arm64"}},
+            ],
+        }
+        self.blobs = {
+            _sha(self.config): self.config,
+            _sha(self.layer0): self.layer0,
+            _sha(self.layer1): self.layer1,
+        }
+
+
+class _StubHubHandler(BaseHTTPRequestHandler):
+    fixtures = None  # set per test class run
+    requests = None
+    token_mode = "ok"  # ok | malformed (200 with an INVALID JSON body)
+
+    def _record(self):
+        length = int(self.headers.get("Content-Length") or 0)
+        body = self.rfile.read(length) if length else b""
+        self.requests.append({
+            "method": self.command,
+            "path": self.path,
+            "authorization": self.headers.get("Authorization", ""),
+            "accept": self.headers.get("Accept", ""),
+            "body": body,
+        })
+
+    def _send(self, status, payload, content_type):
+        data = payload if isinstance(payload, bytes) else _dump(payload)
+        self.send_response(status)
+        self.send_header("Content-Type", content_type)
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        if self.command != "HEAD":
+            self.wfile.write(data)
+
+    def do_GET(self):
+        self._record()
+        path = urllib.parse.urlsplit(self.path).path
+        query = dict(urllib.parse.parse_qsl(
+            urllib.parse.urlsplit(self.path).query))
+        if path == "/token":
+            if self.token_mode == "malformed":
+                # 200 + invalid JSON: the CWE-754 probe — must surface as
+                # a remote failure WITH the IMAGE_PENDING marker, never
+                # a bare traceback.
+                self._send(200, b"this { is not valid json", "application/json")
+            else:
+                self._send(200, {"token": "hub-anon-token"}, "application/json")
+        elif path == "/v2/library/python/manifests/3.11-slim":
+            self._send(200, self.fixtures.index, OCI_INDEX)
+        elif path == "/v2/library/python/manifests/" + self.fixtures.child_digest:
+            self._send(200, self.fixtures.child, OCI_MANIFEST)
+        else:
+            digest = path.rsplit("/", 1)[-1]
+            if digest in self.fixtures.blobs:
+                self._send(200, self.fixtures.blobs[digest],
+                           "application/octet-stream")
+            else:
+                self._send(404, {"errors": [{"digest": digest}]},
+                           "application/json")
+
+    def log_message(self, format, *args):
+        pass
+
+
+class _StubGiteaHandler(BaseHTTPRequestHandler):
+    """Scriptable gitea registry stub.
+
+    state dict keys: requests, uploaded, token_mode (ok|empty),
+    upload_mode (monolithic|redirect307|redirect307_cross|stall),
+    manifest_mode (ok|fail500), redirect_seen, manifest_puts,
+    cross_origin_base (absolute URL of the second-origin stub used by
+    redirect307_cross).
+    """
+
+    state = None
+
+    def _record(self):
+        length = int(self.headers.get("Content-Length") or 0)
+        body = self.rfile.read(length) if length else b""
+        self.state["requests"].append({
+            "method": self.command,
+            "path": self.path,
+            "authorization": self.headers.get("Authorization", ""),
+            "content_type": self.headers.get("Content-Type", ""),
+            "body": body,
+        })
+
+    def _send(self, status, payload=b"", content_type="application/json",
+              headers=None):
+        data = payload if isinstance(payload, bytes) else _dump(payload)
+        self.send_response(status)
+        self.send_header("Content-Type", content_type)
+        self.send_header("Content-Length", str(len(data)))
+        for key, value in (headers or {}).items():
+            self.send_header(key, value)
+        self.end_headers()
+        if self.command != "HEAD":
+            self.wfile.write(data)
+
+    def do_GET(self):
+        self._record()
+        parsed = urllib.parse.urlsplit(self.path)
+        if parsed.path == "/v2/":
+            # G-013 smoke check 1: the registry auth challenge (401 +
+            # Bearer; the realm defaults to this stub's own token
+            # endpoint so pre-existing state dicts work unchanged).
+            realm = self.state.get("challenge_realm") or \
+                "http://{}:{}/v2/token".format(
+                    self.server.server_address[0],
+                    self.server.server_address[1])
+            self._send(401, {"errors": [{"code": "UNAUTHORIZED"}]},
+                       headers={
+                           "WWW-Authenticate":
+                           'Bearer realm="{}",service="{}"'.format(
+                               realm, publish_image.GITEA_AUTH_SERVICE)})
+            return
+        if parsed.path == "/v2/token":
+            if self.state["token_mode"] == "empty":
+                self._send(200, {"token": ""})
+            else:
+                self._send(200, {"token": "stub-jwt"})
+            return
+        if parsed.path.startswith("/api/v1/repos/") and \
+                "/releases/tags/" in parsed.path:
+            # G-013 smoke check 3: the gitea API release-by-tag GET
+            # (state["releases"]: tag -> release payload; absent -> 404).
+            tag = parsed.path.rsplit("/", 1)[-1]
+            release = self.state.get("releases", {}).get(tag)
+            if release is None:
+                self._send(404, {"errors": [{"message": "release not found"}]})
+            else:
+                self._send(200, release)
+            return
+
+    def do_HEAD(self):
+        self._record()
+        parsed = urllib.parse.urlsplit(self.path)
+        digest = parsed.path.rsplit("/", 1)[-1]
+        if digest in self.state["uploaded"]:
+            self._send(200, b"")
+        else:
+            self._send(404, b"")
+
+    def do_POST(self):
+        self._record()
+        parsed = urllib.parse.urlsplit(self.path)
+        query = dict(urllib.parse.parse_qsl(parsed.query))
+        digest = query.get("digest", "")
+        if self.state["upload_mode"] == "stall":
+            time.sleep(15)
+            return
+        if self.state["upload_mode"] == "redirect307_cross" and \
+                digest not in self.state["redirect_seen"]:
+            # CWE-200 probe: ABSOLUTE Location pointing at a DIFFERENT
+            # origin (the second stub server) — the client must follow
+            # WITHOUT the Authorization header.
+            self.state["redirect_seen"].add(digest)
+            location = "{}/v2/{}/blobs/uploads/?digest={}".format(
+                self.state["cross_origin_base"], NAME, digest)
+            self._send(307, b"", headers={"Location": location})
+            return
+        if self.state["upload_mode"] == "redirect307" and \
+                digest not in self.state["redirect_seen"]:
+            self.state["redirect_seen"].add(digest)
+            location = "/v2/{}/blobs/uploads/?digest={}".format(NAME, digest)
+            self._send(307, b"", headers={"Location": location})
+            return
+        if _sha(self.state["requests"][-1]["body"]) == digest:
+            self.state["uploaded"][digest] = self.state["requests"][-1]["body"]
+            self._send(201, {"status": "created"})
+        else:
+            self._send(400, {"errors": [{"digest": digest}]})
+
+    def do_PUT(self):
+        self._record()
+        parsed = urllib.parse.urlsplit(self.path)
+        query = dict(urllib.parse.parse_qsl(parsed.query))
+        body = self.state["requests"][-1]["body"]
+        if "/manifests/" in parsed.path:
+            if self.state["manifest_mode"] == "fail500":
+                self.state.setdefault("manifest_fail_count", 0)
+                self.state["manifest_fail_count"] += 1
+                self._send(500, {"errors": [{"code": "UNKNOWN"}]})
+                return
+            manifest = json.loads(body)
+            referenced = [manifest["config"]["digest"]] + [
+                layer["digest"] for layer in manifest["layers"]]
+            if any(digest not in self.state["uploaded"]
+                   for digest in referenced):
+                self._send(400, {"errors": [{"code": "BLOB_UNKNOWN"}]})
+                return
+            if self.headers.get("Content-Type") != OCI_MANIFEST:
+                self._send(400, {"errors": [{"code": "MANIFEST_INVALID"}]})
+                return
+            self.state.setdefault("manifest_puts", []).append(
+                {"tag": parsed.path.rsplit("/", 1)[-1],
+                 "content_type": self.headers.get("Content-Type", ""),
+                 "manifest": manifest})
+            self._send(201, {})
+            return
+        digest = query.get("digest", "")
+        if _sha(body) == digest:
+            self.state["uploaded"][digest] = body
+            self._send(201, {})
+        else:
+            self._send(400, {"errors": [{"code": "DIGEST_INVALID"}]})
+
+
+class _StubCrossOriginHandler(BaseHTTPRequestHandler):
+    """Second-origin stub (another port) for the CWE-200 redirect probe.
+
+    state dict keys: requests, uploaded — ``uploaded`` is SHARED with the
+    gitea stub (the redirect target models the registry's presigned
+    external storage; a landed blob is visible to the registry), while
+    ``requests`` stays separate so the test can assert on exactly what
+    this origin received — above all that NO Authorization header
+    crossed the origin boundary.
+    """
+
+    state = None
+
+    def _record(self):
+        length = int(self.headers.get("Content-Length") or 0)
+        body = self.rfile.read(length) if length else b""
+        self.state["requests"].append({
+            "method": self.command,
+            "path": self.path,
+            "authorization": self.headers.get("Authorization", ""),
+            "content_type": self.headers.get("Content-Type", ""),
+            "body": body,
+        })
+        return body
+
+    def _send(self, status, payload=b"", content_type="application/json"):
+        data = payload if isinstance(payload, bytes) else _dump(payload)
+        self.send_response(status)
+        self.send_header("Content-Type", content_type)
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        if self.command != "HEAD":
+            self.wfile.write(data)
+
+    def do_POST(self):
+        body = self._record()
+        query = dict(urllib.parse.parse_qsl(
+            urllib.parse.urlsplit(self.path).query))
+        digest = query.get("digest", "")
+        if _sha(body) == digest:
+            self.state["uploaded"][digest] = body
+            self._send(201, {"status": "created"})
+        else:
+            self._send(400, {"errors": [{"code": "DIGEST_INVALID"}]})
+
+    def log_message(self, format, *args):
+        pass
+
+
+class PublishImageTests(unittest.TestCase):
+    """End-to-end publish flows against local stub registries."""
+
+    TAG = "v0.2.2-test"
+
+    @classmethod
+    def setUpClass(cls):
+        cls.fixtures = HubFixtures()
+
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.artifact = os.path.join(self.temp.name, "coreci-broker.pyz")
+        with open(self.artifact, "wb") as handle:
+            handle.write(b"fake pyz payload for tests" * 3)
+        self.env_root = os.path.join(self.temp.name, "envroot")
+        os.makedirs(self.env_root)
+        with open(os.path.join(self.env_root, ".env.secrets"), "w") as handle:
+            handle.write("GITEA_TOKEN=testtoken123\n")
+
+        self.hub_requests = []
+        _StubHubHandler.fixtures = self.fixtures
+        _StubHubHandler.requests = self.hub_requests
+        _StubHubHandler.token_mode = "ok"
+        self.hub = ThreadingHTTPServer(("127.0.0.1", 0), _StubHubHandler)
+        threading.Thread(target=self.hub.serve_forever, daemon=True).start()
+        self.addCleanup(self.hub.shutdown)
+        self.addCleanup(self.hub.server_close)
+
+        self.gitea_state = {
+            "requests": [], "uploaded": {}, "token_mode": "ok",
+            "upload_mode": "monolithic", "manifest_mode": "ok",
+            "redirect_seen": set(),
+        }
+        _StubGiteaHandler.state = self.gitea_state
+        self.gitea = ThreadingHTTPServer(("127.0.0.1", 0), _StubGiteaHandler)
+        threading.Thread(target=self.gitea.serve_forever, daemon=True).start()
+        self.addCleanup(self.gitea.shutdown)
+        self.addCleanup(self.gitea.server_close)
+
+    def run_publish(self, tag=None, timeout=60, artifact=None, env_root=None):
+        argv = [
+            tag or self.TAG,
+            "--artifact", artifact or self.artifact,
+            "--hub-registry", "http://127.0.0.1:{}".format(self.hub.server_address[1]),
+            "--hub-auth", "http://127.0.0.1:{}/token".format(self.hub.server_address[1]),
+            "--gitea-registry", "http://127.0.0.1:{}".format(self.gitea.server_address[1]),
+            "--owner", OWNER, "--repo", REPO,
+            "--env-root", env_root or self.env_root,
+            "--timeout", str(timeout),
+        ]
+        out, err = io.StringIO(), io.StringIO()
+        with redirect_stdout(out), redirect_stderr(err):
+            code = publish_image.main(argv)
+        return code, out.getvalue(), err.getvalue()
+
+    # 1 -----------------------------------------------------------------
+    def test_happy_path_end_to_end(self):
+        code, out, err = self.run_publish()
+        self.assertEqual(code, 0, err or out)
+        puts = self.gitea_state.get("manifest_puts", [])
+        self.assertEqual([entry["tag"] for entry in puts],
+                         [self.TAG, "latest"],
+                         "version tag FIRST, then latest (D-021)")
+        manifest = puts[0]["manifest"]
+        self.assertEqual(puts[0]["content_type"], OCI_MANIFEST)
+        self.assertEqual(manifest["config"]["mediaType"], OCI_CONFIG)
+        broker_layer = manifest["layers"][-1]
+        self.assertEqual(broker_layer["mediaType"], OCI_LAYER)
+        base_descriptors = self.fixtures.child["layers"]
+        self.assertEqual(manifest["layers"][:-1], base_descriptors,
+                         "base layer descriptors reused verbatim")
+        # deterministic broker layer digest matches an independent build
+        with open(self.artifact, "rb") as handle:
+            artifact_bytes = handle.read()
+        layer_blob, layer_digest, diff_id = publish_image.build_broker_layer(
+            artifact_bytes)
+        self.assertEqual(broker_layer["digest"], layer_digest)
+        self.assertEqual(broker_layer["size"], len(layer_blob))
+        # pushed config content
+        pushed_config = json.loads(
+            self.gitea_state["uploaded"][manifest["config"]["digest"]])
+        self.assertEqual(pushed_config["config"]["Entrypoint"],
+                         ["python3", "/app/coreci-broker.pyz"])
+        self.assertEqual(pushed_config["config"]["Cmd"], [])
+        self.assertEqual(pushed_config["config"]["WorkingDir"], "/state")
+        self.assertEqual(pushed_config["config"]["Labels"],
+                         {"org.opencontainers.image.version": self.TAG})
+        self.assertEqual(pushed_config["rootfs"]["diff_ids"][-1], diff_id)
+        self.assertEqual(len(pushed_config["rootfs"]["diff_ids"]),
+                         len(self.fixtures.child["layers"]) + 1)
+        # base config blob uploaded verbatim (digest invariant)
+        self.assertEqual(
+            self.gitea_state["uploaded"][self.fixtures.child["config"]["digest"]],
+            self.fixtures.config)
+        # all blobs present
+        for descriptor in manifest["layers"]:
+            self.assertIn(descriptor["digest"], self.gitea_state["uploaded"])
+        self.assertNotIn("IMAGE_PENDING", out)
+
+    # 2 -----------------------------------------------------------------
+    def test_head_skip_blob_not_reuploaded(self):
+        preseed = self.fixtures.child["layers"][0]["digest"]
+        self.gitea_state["uploaded"][preseed] = self.fixtures.layer0
+        code, out, err = self.run_publish()
+        self.assertEqual(code, 0, err or out)
+        posts = [r for r in self.gitea_state["requests"]
+                 if r["method"] == "POST" and preseed in r["path"]]
+        self.assertEqual(posts, [], "pre-seeded blob must not be re-uploaded")
+        hub_blob_gets = [r for r in self.hub_requests
+                         if r["method"] == "GET" and "/blobs/" in r["path"]
+                         and preseed in r["path"]]
+        self.assertEqual(hub_blob_gets, [],
+                         "skipped blob is never fetched from the hub")
+
+    # 3 -----------------------------------------------------------------
+    def test_redirect_307_relative_location_followed(self):
+        self.gitea_state["upload_mode"] = "redirect307"
+        code, out, err = self.run_publish()
+        self.assertEqual(code, 0, err or out)
+        blob_posts = [r for r in self.gitea_state["requests"]
+                      if r["method"] == "POST" and "/blobs/uploads/" in r["path"]
+                      and "digest=" in r["path"]]
+        self.assertGreater(len(blob_posts), 0)
+        digests = [dict(urllib.parse.parse_qsl(
+            urllib.parse.urlsplit(r["path"]).query))["digest"]
+            for r in blob_posts]
+        self.assertTrue(all(d in self.gitea_state["uploaded"]
+                            for d in digests),
+                        "every redirected upload must land")
+        # each redirected digest POSTed twice (307 + follow)
+        for digest in set(digests):
+            self.assertEqual(digests.count(digest), 2,
+                             "307 then explicit re-POST per digest")
+        # same-origin (relative Location) redirects KEEP the
+        # Authorization header — the CWE-200 strip is cross-origin ONLY
+        for request in blob_posts:
+            self.assertEqual(
+                request["authorization"], "Bearer stub-jwt",
+                "same-origin redirect must keep Authorization")
+
+    # 4 -----------------------------------------------------------------
+    def test_anonymous_token_trap_refused_before_push(self):
+        self.gitea_state["token_mode"] = "empty"
+        code, out, err = self.run_publish()
+        self.assertEqual(code, 1, "remote auth failure -> non-zero")
+        self.assertIn("anonymous-200 trap refused", err)
+        self.assertIn("IMAGE_PENDING", out)
+        post_put = [r for r in self.gitea_state["requests"]
+                    if r["method"] in ("POST", "PUT")]
+        self.assertEqual(post_put, [],
+                         "no blob upload or manifest PUT may happen")
+
+    # 5 -----------------------------------------------------------------
+    def test_layer_and_config_determinism(self):
+        with open(self.artifact, "rb") as handle:
+            artifact = handle.read()
+        first = publish_image.build_broker_layer(artifact)
+        second = publish_image.build_broker_layer(artifact)
+        self.assertEqual(first[1], second[1], "layer digest stable")
+        self.assertEqual(first[2], second[2], "diff_id stable")
+        base_config = json.loads(self.fixtures.config)
+        c1 = publish_image.build_config_blob(base_config, first[2], "t1")
+        c2 = publish_image.build_config_blob(base_config, first[2], "t1")
+        self.assertEqual(_sha(c1), _sha(c2), "config digest stable")
+
+    # 6 -----------------------------------------------------------------
+    def test_amd64_child_selected(self):
+        code, out, err = self.run_publish()
+        self.assertEqual(code, 0, err or out)
+        child_gets = [r for r in self.hub_requests
+                      if "/manifests/sha256:" in r["path"]]
+        self.assertTrue(child_gets, "child manifest must be fetched by digest")
+        self.assertTrue(all(self.fixtures.child_digest in r["path"]
+                            for r in child_gets),
+                        "amd64 child only")
+        self.assertFalse(any(self.fixtures.arm64_child_digest in r["path"]
+                              for r in self.hub_requests),
+                         "arm64 child never touched")
+
+    # 7 -----------------------------------------------------------------
+    def test_timeout_aborts_rather_than_hangs(self):
+        self.gitea_state["upload_mode"] = "stall"
+        started = time.monotonic()
+        code, out, err = self.run_publish(timeout=1)
+        elapsed = time.monotonic() - started
+        self.assertEqual(code, 1)
+        self.assertIn("IMAGE_PENDING", out)
+        self.assertLess(elapsed, 30, "bounded retries, no hang")
+        stalls = [r for r in self.gitea_state["requests"]
+                  if r["method"] == "POST" and "/blobs/uploads/" in r["path"]]
+        self.assertLessEqual(len(stalls), 3, "bounded retry attempts")
+
+    # 8 -----------------------------------------------------------------
+    def test_manifest_retry_exhaustion_marker(self):
+        self.gitea_state["manifest_mode"] = "fail500"
+        code, out, err = self.run_publish()
+        self.assertEqual(code, 1)
+        self.assertIn("IMAGE_PENDING", out)
+        self.assertEqual(self.gitea_state.get("manifest_fail_count"), 3,
+                         "exactly three bounded attempts on the tag PUT")
+        self.assertNotIn("latest",
+                         [r["path"].rsplit("/", 1)[-1]
+                          for r in self.gitea_state["requests"]
+                          if r["method"] == "PUT"],
+                         "latest must not be attempted after exhaustion")
+
+    # 9 -----------------------------------------------------------------
+    def test_token_discipline(self):
+        code, out, err = self.run_publish()
+        self.assertEqual(code, 0, err or out)
+        token_calls = [r for r in self.gitea_state["requests"]
+                       if r["path"].startswith("/v2/token")]
+        self.assertTrue(token_calls)
+        expected = base64.b64encode(
+            b"publisher:testtoken123").decode("ascii")
+        for call in token_calls:
+            self.assertEqual(call["authorization"],
+                             "Basic {}".format(expected))
+        # scope requested with pull,push (registry v2 scope form)
+        query = urllib.parse.unquote(
+            urllib.parse.urlsplit(token_calls[0]["path"]).query)
+        self.assertIn("repository:{}/{}:pull,push".format(OWNER, REPO),
+                      query)
+        # missing token: local precondition, zero remote requests
+        empty_root = os.path.join(self.temp.name, "emptyroot")
+        os.makedirs(empty_root)
+        before = len(self.gitea_state["requests"]) + len(self.hub_requests)
+        code, out, err = self.run_publish(env_root=empty_root)
+        self.assertEqual(code, 2, "local precondition -> exit 2")
+        self.assertNotIn("IMAGE_PENDING", out, "no marker for local errors")
+        after = len(self.gitea_state["requests"]) + len(self.hub_requests)
+        self.assertEqual(before, after, "no request of any kind was made")
+
+    # 10 ----------------------------------------------------------------
+    def test_malformed_hub_token_json_keeps_pending_marker(self):
+        """CWE-754 marker discipline: a 200 with an INVALID JSON body from
+        the hub token endpoint fails as a remote answer — exit 1 WITH the
+        IMAGE_PENDING marker — never an uncaught-ValueError traceback."""
+        _StubHubHandler.token_mode = "malformed"
+        code, out, err = self.run_publish()
+        self.assertEqual(code, 1)
+        self.assertIn("malformed JSON", err,
+                      "the failure class must be named in the error")
+        self.assertIn("IMAGE_PENDING", out,
+                      "marker discipline holds for malformed remote JSON")
+        self.assertNotIn("Traceback", out + err)
+        self.assertEqual(self.gitea_state["requests"], [],
+                         "nothing is attempted against gitea")
+
+    # 11 ----------------------------------------------------------------
+    def test_redirect_307_cross_origin_strips_authorization(self):
+        """CWE-200: a 307 whose Location resolves to a DIFFERENT origin
+        is followed WITHOUT the Authorization header — the registry JWT
+        never crosses origins — and the blob still lands."""
+        cross_state = {"requests": [], "uploaded": self.gitea_state["uploaded"]}
+        _StubCrossOriginHandler.state = cross_state
+        cross = ThreadingHTTPServer(("127.0.0.1", 0), _StubCrossOriginHandler)
+        threading.Thread(target=cross.serve_forever, daemon=True).start()
+        self.addCleanup(cross.shutdown)
+        self.addCleanup(cross.server_close)
+        self.gitea_state["upload_mode"] = "redirect307_cross"
+        self.gitea_state["cross_origin_base"] = "http://127.0.0.1:{}".format(
+            cross.server_address[1])
+        code, out, err = self.run_publish()
+        self.assertEqual(code, 0, err or out)
+        self.assertTrue(cross_state["requests"],
+                        "the redirect was followed to the other origin")
+        for request in cross_state["requests"]:
+            self.assertEqual(
+                request["authorization"], "",
+                "Authorization must be stripped on cross-origin redirects")
+        # the blob landed (visible through the shared registry blob store)
+        self.assertEqual(len(self.gitea_state["uploaded"]), 5,
+                         "base config + 2 base layers + broker layer + "
+                         "image config all landed")
+        # and the flow still completes: both manifest PUTs back on the
+        # origin, where Authorization remains in play
+        self.assertEqual(
+            [entry["tag"] for entry in self.gitea_state.get(
+                "manifest_puts", [])],
+            [self.TAG, "latest"])
+
+
+class SmokeModeTests(unittest.TestCase):
+    """G-013 --smoke logic, stub-tested.
+
+    The LIVE smoke never runs in this suite — it executes only at the
+    P3-05 ship, where the ship gate sets PUBLISH_SMOKE=1 in the
+    environment (the script itself never reads that variable). Here the
+    four GET-only checks run against fresh stub instances: zero network,
+    and the stub records double as the read-only proof (a POST/PUT
+    landing anywhere in a smoke run is a hard failure).
+    """
+
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.env_root = os.path.join(self.temp.name, "envroot")
+        os.makedirs(self.env_root)
+        with open(os.path.join(self.env_root, ".env.secrets"), "w") as handle:
+            handle.write("GITEA_TOKEN=testtoken123\n")
+
+        self.fixtures = HubFixtures()
+        self.hub_requests = []
+        _StubHubHandler.fixtures = self.fixtures
+        _StubHubHandler.requests = self.hub_requests
+        _StubHubHandler.token_mode = "ok"
+        self.hub = ThreadingHTTPServer(("127.0.0.1", 0), _StubHubHandler)
+        threading.Thread(target=self.hub.serve_forever, daemon=True).start()
+        self.addCleanup(self.hub.shutdown)
+        self.addCleanup(self.hub.server_close)
+
+        self.gitea_state = {
+            "requests": [], "uploaded": {}, "token_mode": "ok",
+            "upload_mode": "monolithic", "manifest_mode": "ok",
+            "redirect_seen": set(), "releases": {},
+        }
+        _StubGiteaHandler.state = self.gitea_state
+        self.gitea = ThreadingHTTPServer(("127.0.0.1", 0), _StubGiteaHandler)
+        threading.Thread(target=self.gitea.serve_forever, daemon=True).start()
+        self.addCleanup(self.gitea.shutdown)
+        self.addCleanup(self.gitea.server_close)
+        # The /v2/ challenge realm must point back at THIS stub's token
+        # endpoint so the recorded-realm exchange stays in-stub.
+        self.gitea_state["challenge_realm"] = \
+            "http://127.0.0.1:{}/v2/token".format(
+                self.gitea.server_address[1])
+
+    def run_smoke(self, tag, env_root=None):
+        argv = [
+            tag, "--smoke",
+            "--hub-registry",
+            "http://127.0.0.1:{}".format(self.hub.server_address[1]),
+            "--hub-auth",
+            "http://127.0.0.1:{}/token".format(
+                self.hub.server_address[1]),
+            "--gitea-registry",
+            "http://127.0.0.1:{}".format(self.gitea.server_address[1]),
+            "--owner", OWNER, "--repo", REPO,
+            "--env-root", env_root or self.env_root,
+        ]
+        out, err = io.StringIO(), io.StringIO()
+        with redirect_stdout(out), redirect_stderr(err):
+            code = publish_image.main(argv)
+        return code, out.getvalue(), err.getvalue()
+
+    def _post_put(self):
+        """(gitea POST/PUT, hub POST/PUT) across the stub records."""
+        return ([r for r in self.gitea_state["requests"]
+                 if r["method"] in ("POST", "PUT")],
+                [r for r in self.hub_requests
+                 if r["method"] in ("POST", "PUT")])
+
+    def _release_gets(self):
+        return [r for r in self.gitea_state["requests"]
+                if "/releases/tags/" in r["path"]]
+
+    # 12 ----------------------------------------------------------------
+    def test_smoke_all_checks_pass(self):
+        self.gitea_state["releases"]["v0.2.0"] = {
+            "tag_name": "v0.2.0",
+            "assets": [{"name": "phase0-notes.txt"}],
+        }
+        code, out, err = self.run_smoke("v0.2.0")
+        self.assertEqual(code, 0, err or out)
+        for marker in ("PASS [1/4]", "PASS [2/4]", "PASS [3/4]",
+                       "PASS [4/4]"):
+            self.assertIn(marker, out)
+        self.assertIn("G-021 confirmed: no .pyz asset", out,
+                      "the v0.2.0 run carries the G-021 forge-side "
+                      "confirmation")
+        self.assertIn("phase0-notes.txt", out, "asset names print as a fact")
+        self.assertIn(self.fixtures.child_digest, out,
+                      "the amd64 child digest prints (drift check)")
+        # the manifest GET carries the full Accept set
+        manifest_gets = [r for r in self.hub_requests
+                         if "/manifests/3.11-slim" in r["path"]]
+        self.assertEqual(len(manifest_gets), 1)
+        self.assertEqual(manifest_gets[0]["accept"],
+                         publish_image.ACCEPT_MANIFESTS)
+        # the release GET uses the gitea API token auth form
+        release_gets = self._release_gets()
+        self.assertEqual(len(release_gets), 1)
+        self.assertEqual(release_gets[0]["authorization"],
+                         "token testtoken123")
+        # READ-ONLY PROOF: zero POST/PUT recorded at either stub
+        gitea_writes, hub_writes = self._post_put()
+        self.assertEqual(gitea_writes, [], "no gitea write of any kind")
+        self.assertEqual(hub_writes, [], "no hub write of any kind")
+        self.assertNotIn("IMAGE_PENDING", out)
+
+    # 13 ----------------------------------------------------------------
+    def test_smoke_v0_2_0_no_pyz_assertion(self):
+        pyz_release = {"tag_name": "v0.2.0",
+                       "assets": [{"name": "coreci-broker-v0.2.0.pyz"}]}
+        self.gitea_state["releases"]["v0.2.0"] = pyz_release
+        self.gitea_state["releases"]["v0.2.2"] = pyz_release
+        # v0.2.0 + a .pyz asset: the G-021 hard assertion fails the smoke
+        code, out, err = self.run_smoke("v0.2.0")
+        self.assertEqual(code, 1)
+        self.assertIn("FAILED [3/4]", out)
+        self.assertIn("G-021", out, "the failure line names G-021")
+        self.assertIn(".pyz", out)
+        self.assertEqual(self.hub_requests, [],
+                         "a FAILED check stops the smoke: the hub "
+                         "resolve (step 4) never runs")
+        # the SAME release under a different tag: the assertion binds
+        # v0.2.0 only -> all four checks pass
+        code, out, err = self.run_smoke("v0.2.2")
+        self.assertEqual(code, 0, err or out)
+        for marker in ("PASS [3/4]", "PASS [4/4]"):
+            self.assertIn(marker, out)
+        self.assertNotIn("G-021 violation", out)
+        gitea_writes, hub_writes = self._post_put()
+        self.assertEqual(gitea_writes + hub_writes, [],
+                         "still read-only across both runs")
+
+    # 14 ----------------------------------------------------------------
+    def test_smoke_trap_refused(self):
+        self.gitea_state["token_mode"] = "empty"
+        self.gitea_state["releases"]["v0.2.3"] = {
+            "tag_name": "v0.2.3", "assets": [],
+        }
+        code, out, err = self.run_smoke("v0.2.3")
+        self.assertEqual(code, 1, "the trap is a loud remote failure")
+        self.assertIn("FAILED [2/4]", out)
+        self.assertIn("anonymous-200 trap refused", out)
+        self.assertIn("PASS [1/4]", out, "check 1 ran (order 1 -> 4)")
+        self.assertNotIn("PASS [3/4]", out, "checks after the trap never run")
+        self.assertNotIn("PASS [4/4]", out)
+        self.assertEqual(self._release_gets(), [],
+                         "the release GET (step 3) is skipped")
+        self.assertEqual(self.hub_requests, [],
+                         "zero hub requests: the index resolve (step 4) "
+                         "is skipped")
+
+    # 15 ----------------------------------------------------------------
+    def test_smoke_missing_token_local(self):
+        empty_root = os.path.join(self.temp.name, "emptyroot")
+        os.makedirs(empty_root)
+        code, out, err = self.run_smoke("v0.2.3", env_root=empty_root)
+        self.assertEqual(code, 2, "local precondition -> exit 2")
+        self.assertIn("ERROR", err)
+        self.assertIn("GITEA_TOKEN", err)
+        self.assertEqual(self.gitea_state["requests"], [],
+                         "zero gitea requests — the precondition fires "
+                         "before anything remote")
+        self.assertEqual(self.hub_requests, [], "zero hub requests")
+        self.assertNotIn("IMAGE_PENDING", out, "no marker for local errors")
+        self.assertNotIn("PASS", out, "no smoke check ever ran")
+
+
+if __name__ == "__main__":
+    unittest.main()