fix(soak): require explicit existing acceptance input before operational IO

backend/tests/tooling/test_stage6_soak.py

@@ -17,6 +17,68 @@ from collector import manifest_at, switch  # noqa: E402
 from state import append, digest, read_journal  # noqa: E402
 
 
+# #region Test.Stage6Soak.AcceptanceAdmission [C:3] [TYPE Function]
+# @TEST_INVARIANT Missing/malformed actual input refuses before any secret/client/lock/SQL call or output creation.
+@pytest.mark.parametrize("body,code", [
+    (None, "SOAK_ACCEPTANCE_INPUT_UNAVAILABLE"),
+    ("{private-sentinel", "SOAK_ACCEPTANCE_INPUT_INVALID"),
+    ("[]", "SOAK_ACCEPTANCE_INPUT_INVALID"),
+    ('{"scenario_id":"s"}', "SOAK_ACCEPTANCE_INPUT_INVALID"),
+    ('{"scenario_id":"s","revision_id":"r","publication_commit_hash":"h","runs":{"manual":{"run_id":null}}}',
+     "SOAK_ACCEPTANCE_INPUT_INVALID"),
+])
+def test_prepare_input_refuses_before_operational_io(tmp_path, monkeypatch, capsys, body, code):
+    import collector
+
+    source = tmp_path / "acceptance.json"
+    if body is not None:
+        source.write_text(body)
+    root = tmp_path / "never-created"
+
+    def forbidden(*args, **kwargs):
+        pytest.fail("Input admission must precede secrets/client/lock/database")
+
+    for name in ("secrets", "Client", "lock", "database"):
+        monkeypatch.setattr(collector, name, forbidden)
+    monkeypatch.setattr(sys, "argv", ["collector", "prepare", "--acceptance", str(source), "--root", str(root)])
+    assert collector.main() == 1
+    assert json.loads(capsys.readouterr().out) == {"status": "BLOCKED", "error_code": code}
+    assert not root.exists()
+# #endregion Test.Stage6Soak.AcceptanceAdmission
+
+
+# #region Test.Stage6Soak.AcceptanceRead [C:2] [TYPE Function]
+# @TEST_INVARIANT Reading explicit source is pure input validation, not source-run or pin authority.
+def test_acceptance_input_retains_one_authored_payload(tmp_path):
+    from collector import acceptance_input
+
+    source = tmp_path / "independent-result-audit.json"
+    source.write_text('{"scenario_id":"scenario","revision_id":"revision","publication_commit_hash":"commit",'
+                      '"runs":{"manual":{"run_id":"manual"}}}')
+    actual = acceptance_input(source)
+    source.unlink()
+    assert actual == {"scenario_id": "scenario", "revision_id": "revision", "publication_commit_hash": "commit",
+                      "runs": {"manual": {"run_id": "manual"}}}
+# #endregion Test.Stage6Soak.AcceptanceRead
+
+
+# #region Test.Stage6Soak.ComposeInput [C:2] [TYPE Function]
+# @TEST_INVARIANT The existing host input must be explicit and cannot be autocreated; this is config inspection, not a soak run.
+def test_compose_acceptance_bind_is_required_existing_readonly():
+    import yaml
+
+    compose = Path(__file__).resolve().parents[3] / "docker-compose.full-flow-soak.yml"
+    value = yaml.safe_load(compose.read_text())
+    binds = [v for v in value["services"]["soak"]["volumes"] if isinstance(v, dict) and v.get("target") == "/acceptance"]
+    assert len(binds) == 1
+    assert binds[0]["type"] == "bind"
+    assert binds[0]["source"].startswith("${SOAK_ACCEPTANCE_DIR:?SOAK_ACCEPTANCE_INPUT_UNAVAILABLE")
+    assert binds[0]["read_only"] is True
+    assert binds[0]["bind"] == {"create_host_path": False}
+    assert "specs/050-mcp-interface/evidence" not in compose.read_text()
+# #endregion Test.Stage6Soak.ComposeInput
+
+
 # #region Test.Stage6Soak.Fixture [C:2] [TYPE Function] [SEMANTICS fixture,diagnostic]
 def fixture():
     start = datetime(2026, 10, 1, tzinfo=UTC)

docker-compose.full-flow-soak.yml

@@ -14,7 +14,12 @@ services:
       - ./scripts/stage6_soak:/opt/soak:ro
       - ${SOAK_EVIDENCE_DIR:-./artifacts/docker-stage6/soak}:/evidence
       - ${FULL_FLOW_SOAK_ENV_FILE:-/tmp/ss-tools-full-flow.env}:/run/secrets/full-flow.env:ro
-      - ./specs/050-mcp-interface/evidence/docker-full-flow-v2:/acceptance:ro
+      - type: bind
+        source: ${SOAK_ACCEPTANCE_DIR:?SOAK_ACCEPTANCE_INPUT_UNAVAILABLE - set an existing independently audited artifact or external directory}
+        target: /acceptance
+        read_only: true
+        bind:
+          create_host_path: false
       - storage:/retained-storage:ro
     depends_on:
       backend:

scripts/stage6_soak/README.md

@@ -19,9 +19,16 @@ identities and typed error names; Docker inspection retains an allowlist only.
 Before execution freeze the source/image/dirty-diff/migration identity separately
 in the evidence packet, and declare the queue/coverage SLO. The commands below
 use the existing actual independently audited M01 source run; preparation creates
-its own **disabled** pinned schedule. It cannot enable an old schedule.
+its own **disabled** pinned schedule. It cannot enable an old schedule. Set
+`SOAK_ACCEPTANCE_DIR` explicitly to an existing ignored artifact directory or an
+external directory containing `independent-result-audit.json`; Compose refuses
+an unset variable or absent bind source and never creates that input directory.
+No historical input is restored from Git. For example, after independently
+auditing an actual source run into that existing directory:
 
 ```sh
+export SOAK_ACCEPTANCE_DIR="$PWD/artifacts/docker-full-flow-v2"
+test -f "$SOAK_ACCEPTANCE_DIR/independent-result-audit.json"
 docker compose --env-file /tmp/ss-tools-full-flow.env -f docker-compose.full-flow.yml -f docker-compose.full-flow-soak.yml --profile stage6-soak run --rm soak prepare --source-commit "$(git rev-parse HEAD)" --max-gap 180 --max-queue-age 300
 docker compose --env-file /tmp/ss-tools-full-flow.env -f docker-compose.full-flow.yml -f docker-compose.full-flow-soak.yml --profile stage6-soak run --rm soak status
 docker compose --env-file /tmp/ss-tools-full-flow.env -f docker-compose.full-flow.yml -f docker-compose.full-flow-soak.yml --profile stage6-soak run --rm soak tabs --tabs 5
@@ -30,6 +37,11 @@ docker compose --env-file /tmp/ss-tools-full-flow.env -f docker-compose.full-flo
 ```
 
 The image needs its existing Playwright Chromium installation and `psycopg2`.
+Preparation validates the input before credentials or operational I/O and reports
+`SOAK_ACCEPTANCE_INPUT_UNAVAILABLE` or `SOAK_ACCEPTANCE_INPUT_INVALID` on refusal.
+This historical M01 recipe still requires its authored metric admission/pin
+format; compatibility with a current source run and operational acceptance have
+not been proved by these input changes. Unsupported pins remain unavailable.
 Read-only database access targets the isolated `db`/`ss_tools`; storage is mounted
 read-only at `/retained-storage`. The host evidence directory is
 `artifacts/docker-stage6/soak`, override via

scripts/stage6_soak/collector.py

@@ -19,6 +19,41 @@ from state import append, atomic, digest, lock, now, read_journal
 from tabs import run as tab_canary
 
 
+# #region Tooling.Stage6Soak.Collector.InputRefusal [C:2] [TYPE Class]
+class AcceptanceInputRefusal(ValueError):
+    def __init__(self, code="SOAK_ACCEPTANCE_INPUT_INVALID"):
+        if code not in {"SOAK_ACCEPTANCE_INPUT_INVALID", "SOAK_ACCEPTANCE_INPUT_UNAVAILABLE"}:
+            raise ValueError("SOAK_ACCEPTANCE_INPUT_INVALID")
+        self.code = code
+        super().__init__(code)
+# #endregion Tooling.Stage6Soak.Collector.InputRefusal
+
+
+# #region Tooling.Stage6Soak.Collector.AcceptanceInput [C:3] [TYPE Function] [SEMANTICS input,explicit,refusal]
+# @PRE An operator supplies an existing independently audited result; no historical copy or default discovery is performed.
+# @POST Missing or malformed input refuses before credentials, clients, locks, database or operational writes.
+def acceptance_input(path):
+    try:
+        source = path.resolve(strict=True)
+        if not source.is_file() or source.stat().st_size > 1_048_576:
+            raise AcceptanceInputRefusal()
+        value = json.loads(source.read_text(encoding="utf-8"))
+    except (OSError, RuntimeError):
+        raise AcceptanceInputRefusal("SOAK_ACCEPTANCE_INPUT_UNAVAILABLE") from None
+    except (UnicodeError, ValueError):
+        raise AcceptanceInputRefusal() from None
+    if not isinstance(value, dict):
+        raise AcceptanceInputRefusal()
+    fields = [value.get(name) for name in ("scenario_id", "revision_id", "publication_commit_hash")]
+    runs = value.get("runs")
+    manual = runs.get("manual") if isinstance(runs, dict) else None
+    fields.append(manual.get("run_id") if isinstance(manual, dict) else None)
+    if any(not isinstance(field, str) or not field.strip() for field in fields):
+        raise AcceptanceInputRefusal()
+    return value
+# #endregion Tooling.Stage6Soak.Collector.AcceptanceInput
+
+
 # #region Tooling.Stage6Soak.Collector.Prepare [C:4] [TYPE Function] [SEMANTICS manifest,pin,disabled,schedule]
 # @RELATION CALLS -> [Tooling.Stage6Soak.State.Append]
 # @PRE Existing independent acceptance identifies a genuine completed M01 source run; private fixture input is isolated.
@@ -31,7 +66,7 @@ def prepare(args, env, client):
         raise ValueError("SOAK_SLO_OUT_OF_BOUNDS")
     if (root / "manifest.json").exists():
         raise ValueError("SOAK_MANIFEST_ALREADY_EXISTS")
-    acceptance = json.loads(args.acceptance.read_text())
+    acceptance = args.acceptance_data
     with database(env) as connection, connection.cursor(cursor_factory=RealDictCursor) as cursor:
         cursor.execute("SELECT id,status,runner_plan FROM scenario_runs WHERE id=%s", (acceptance["runs"]["manual"]["run_id"],))
         source = cursor.fetchone()
@@ -190,6 +225,12 @@ def main():
     parser.add_argument("--max-queue-age", type=int, default=300)
     parser.add_argument("--tabs", type=int, choices=[5, 15, 50])
     args = parser.parse_args()
+    if args.command == "prepare":
+        try:
+            args.acceptance_data = acceptance_input(args.acceptance)
+        except AcceptanceInputRefusal as failure:
+            print(json.dumps({"status": "BLOCKED", "error_code": failure.code}))
+            return 1
     env = secrets(args.env_file)
     client = Client(env, args.base)
     if args.command == "run":