docs(P02): complete docker-release-integration phase

---ci---
phase: 2
milestone: v0.2
status: complete
requirements:
  covered: [REQ-23]
  partial: []
---/ci---

.ciagent/CHECKPOINT.json

@@ -0,0 +1,8 @@
+{
+  "phase": 2,
+  "stage": "execute",
+  "milestone": "v0.2",
+  "phase_role": "execution",
+  "attempts": 0,
+  "updated_at": "2026-10-10T21:20: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 | complete |
-| REQ-22 | 1 | complete |
-| REQ-23 | 2 | complete |
+| REQ-21 | 1 | pending |
+| REQ-22 | 1 | pending |
+| REQ-23 | 2 | pending |

.ciagent/ROADMAP.md

@@ -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**: complete (shipped as `v0.2.2` — 341 cumulative tests green, verify four-layer PASS, .pyz attached by script, asset id 397)
+- **Status**: not_started
 
 ## 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**: complete (shipped as `v0.2.3` — milestone release; 356 cumulative tests green, security review PASS-WITH-FINDINGS remediated, `.pyz` attached + notes, image published once)
+- **Status**: not_started
 
 ## Post-Milestone Notes (carry-forward from v0.1, GRILL G-008)
 

.dockerignore

@@ -9,9 +9,4 @@ data/
 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
+.ciagent/

Dockerfile

@@ -18,17 +18,7 @@
 #   - 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
+COPY coreci_broker.py coreci_broker_client.py coreci_broker_ui/ cli/ /app/src/
 
 ARG VERSION=dev
 RUN printf 'VERSION = "%s"\n' "$VERSION" > /app/src/_cli_version.py \

README.md

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

scripts/publish_image.py

@@ -2,7 +2,7 @@
 """Daemonless OCI image publish client (D-019; stdlib only).
 
 Usage:
-    python3 scripts/publish_image.py <tag> [--smoke]
+    python3 scripts/publish_image.py <tag>
         [--artifact PATH] [--hub-registry URL] [--hub-auth URL]
         [--gitea-registry URL] [--env-root DIR] [--owner O] [--repo R]
         [--base-manifest DIGEST] [--timeout SECS]
@@ -38,47 +38,6 @@ OR a permanent registry answer — printing the machine-readable
 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
@@ -89,7 +48,6 @@ import http.client
 import io
 import json
 import os
-import re
 import sys
 import tarfile
 import time
@@ -112,9 +70,6 @@ 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"
@@ -171,31 +126,6 @@ def _error_detail(body):
     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)
@@ -210,33 +140,16 @@ def _attempt_once(method, url, headers, data, timeout):
         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."""
+    NOT method-preserve on PUT/POST — the Location may be relative)."""
     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)
+            method, current_url, headers, data, timeout)
         if status in (307, 308) and resp_headers.get("Location"):
-            next_url = urllib.parse.urljoin(
+            current_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(
@@ -281,10 +194,8 @@ def _hub_token(hub_auth, scope):
         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 ""
+    token = json.loads(body).get("token") or json.loads(body).get(
+        "access_token") or ""
     if not token:
         raise RegistryApiError("hub token endpoint returned no token",
                                status, body)
@@ -299,28 +210,18 @@ def _fetch_manifest(registry, repository, ref, bearer, accept=ACCEPT_MANIFESTS):
     if status != 200:
         raise RegistryApiError(
             "manifest fetch {} {}".format(repository, ref), status, body)
-    return _parse_json(
-        body, "manifest fetch {} {}".format(repository, ref), status)
+    return json.loads(body)
 
 
 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
+    for descriptor in index.get("manifests", []):
+        platform = descriptor.get("platform") or {}
         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"))
+        json.dumps(index.get("manifests", [])).encode())
 
 
 def _fetch_blob(registry, repository, digest, bearer, timeout):
@@ -337,37 +238,7 @@ def _fetch_blob(registry, repository, digest, bearer, timeout):
     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).
-    """
+def _gitea_jwt(gitea_registry, owner, repo, token, timeout):
     query = urllib.parse.urlencode({
         "service": GITEA_AUTH_SERVICE,
         "scope": "repository:{}/{}:pull,push".format(owner, repo),
@@ -375,13 +246,16 @@ def _gitea_jwt(token_url, owner, repo, token, timeout):
     basic = base64.b64encode(
         "publisher:{}".format(token).encode("utf-8")).decode("ascii")
     status, body, _ = _request(
-        "GET", "{}?{}".format(token_url, query),
+        "GET", "{}/v2/token?{}".format(gitea_registry.rstrip("/"), 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 ""
+    jwt = ""
+    try:
+        jwt = json.loads(body).get("token") or ""
+    except ValueError:
+        pass
     if not jwt:
         # The endpoint answers 200 even anonymously with an unusable
         # (empty-scope) JWT — a 200 proves nothing about push rights.
@@ -392,38 +266,6 @@ def _gitea_jwt(token_url, owner, repo, token, timeout):
     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)
@@ -542,127 +384,6 @@ def _load_artifact(path):
         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."""
@@ -686,21 +407,18 @@ def publish(tag, artifact_path, hub_registry, hub_auth, gitea_registry,
                         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,
+        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)
+    base_config = json.loads(base_config_bytes)
+    base_layer_descriptors = list(child.get("layers") or [])
 
-    jwt = _gitea_jwt("{}/v2/token".format(gitea_registry.rstrip("/")),
-                     owner, repo, gitea_token, metadata_timeout)
+    jwt = _gitea_jwt(gitea_registry, owner, repo, gitea_token,
+                     metadata_timeout)
 
     pushed = []
     pushed.append(_push_blob_if_missing(
-        gitea_registry, name, child_config["digest"], base_config_bytes,
+        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,
@@ -733,7 +451,7 @@ def publish(tag, artifact_path, hub_registry, hub_auth, gitea_registry,
         "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_config_digest": child["config"]["digest"],
         "base_layers": [descriptor["digest"]
                         for descriptor in base_layer_descriptors],
         "broker_layer_digest": layer_digest,
@@ -764,10 +482,6 @@ def main(argv=None):
     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 "
@@ -785,22 +499,9 @@ def main(argv=None):
               " 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))
+    blob_timeout = max(1.0, float(args.timeout))
     # Reuse the smaller metadata timeout for the registry auth exchange so
     # the timeout-abort path stays bounded for every request class.
     try:

tests/test_acceptance_v02.py

@@ -1,548 +0,0 @@
-"""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

@@ -3,21 +3,14 @@
 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
+equivalence contract: same staged sources, 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")
@@ -28,11 +21,9 @@ 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",
 )
 
 
@@ -41,49 +32,6 @@ def read(name):
         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
@@ -108,107 +56,24 @@ class DockerfileConsistencyTests(unittest.TestCase):
                              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_copy_set_matches_staging_allowlist(self):
+        """The build context stage EXACTLY the build script's allowlist."""
+        self.assertEqual(len(self.copy_lines), 1,
+                         "one COPY line stages the sources")
+        staged = set(self.copy_lines[0].split()[1:-1])
+        expected = set(build_cli.ROOT_SOURCES) | {
+            build_cli.UI_SOURCE_DIR + "/", build_cli.SHIM_SOURCE.split(os.sep)[0] + "/"}
+        self.assertEqual(staged, expected,
+                         "COPY set == build_cli staging allowlist")
+        self.assertTrue(self.copy_lines[0].split()[-1].startswith("/app/src/"))
+        self.assertNotIn("dist/", self.copy_lines[0],
+                         "no pre-built artifact in the build context")
 
     def test_zipapp_invocation_flags_match_build_script(self):
         """Same --python interpreter and --compress flag as build_cli."""
@@ -238,7 +103,7 @@ class DockerfileConsistencyTests(unittest.TestCase):
         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 ")]
+                           if line.startswith("WORKDIR ")]
         self.assertEqual(len(docker_entrypoint), 1)
         self.assertEqual(json.loads(
             docker_entrypoint[0].split(" ", 1)[1]), entrypoint)

tests/test_publish_image.py

@@ -97,7 +97,6 @@ class HubFixtures:
 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)
@@ -125,13 +124,7 @@ class _StubHubHandler(BaseHTTPRequestHandler):
         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")
+            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:
@@ -153,10 +146,8 @@ 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).
+    upload_mode (monolithic|redirect307|stall), manifest_mode
+    (ok|fail500), redirect_seen, manifest_puts.
     """
 
     state = None
@@ -187,37 +178,11 @@ class _StubGiteaHandler(BaseHTTPRequestHandler):
     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()
@@ -236,16 +201,6 @@ class _StubGiteaHandler(BaseHTTPRequestHandler):
         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)
@@ -293,55 +248,6 @@ class _StubGiteaHandler(BaseHTTPRequestHandler):
             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."""
 
@@ -365,7 +271,6 @@ class PublishImageTests(unittest.TestCase):
         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)
@@ -476,12 +381,6 @@ class PublishImageTests(unittest.TestCase):
         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):
@@ -576,224 +475,6 @@ class PublishImageTests(unittest.TestCase):
         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()