Validate complete SOXL P1 inputs on the VPS

.github/workflows/research_input_readback.yml

@@ -14,7 +14,7 @@ jobs:
   readback:
     if: github.repository == 'QuantStrategyLab/AIAuditBridge' && github.ref == 'refs/heads/main'
     runs-on: [self-hosted, codex-vps]
-    timeout-minutes: 5
+    timeout-minutes: 10
     permissions:
       contents: read
       id-token: write
@@ -24,6 +24,26 @@ jobs:
         with:
           persist-credentials: false
 
+      - name: Checkout the frozen original P1 validator
+        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        with:
+          repository: QuantStrategyLab/UsEquitySnapshotPipelines
+          ref: ca61b82c2a508a1cc81fb5831294ba9835ac41c2
+          path: validator-source
+          persist-credentials: false
+
+      - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v6.2.0
+        with:
+          python-version: '3.11'
+
+      # Match the source workflow's locked environment before minting the
+      # short-lived cloud credential, so installation cannot consume its life.
+      - name: Install the frozen validator runtime
+        working-directory: validator-source
+        run: |
+          python -m pip install --quiet 'uv==0.11.19'
+          env -u UV_PYTHON -u VIRTUAL_ENV uv sync --locked --no-dev --no-editable --python 3.11
+
       # Direct federation grants only object reads on the fixed P1 root below.
       # The provider additionally binds numeric repo/owner IDs, main and this workflow.
       # No service-account key, user ADC, daemon credential or AI invocation is used.
@@ -36,16 +56,33 @@ jobs:
           export_environment_variables: true
           cleanup_credentials: true
 
-      - name: Read completion marker metadata without downloading research data
+      - name: Read and verify the completed immutable P1 root
         env:
           CLOUDSDK_CORE_DISABLE_PROMPTS: '1'
           CLOUDSDK_CORE_DISABLE_USAGE_REPORTING: 'true'
+          INPUT_ROOT: gs://qsl-runtime-logs-shared/soxl-p1-p3/06ce97ad581fdc465896ef324ce16ac65f2695af95ec18ce27238bdd71032f74/
+          EXPECTED_MANIFEST_SHA256: 06ce97ad581fdc465896ef324ce16ac65f2695af95ec18ce27238bdd71032f74
+          READBACK_ROOT: ${{ runner.temp }}/aab-research-input-${{ github.run_id }}-${{ github.run_attempt }}
+          VALIDATOR_SOURCE: ${{ github.workspace }}/validator-source
         run: |
+          set -euo pipefail
+          cleanup() {
+            rm -rf -- "$READBACK_ROOT"
+          }
+          trap cleanup EXIT
+          if [ -e "$READBACK_ROOT" ] || [ -L "$READBACK_ROOT" ]; then
+            echo '{"status":"unavailable","reason":"temporary_root_unavailable"}'
+            exit 1
+          fi
+          umask 077
+          mkdir -m 700 "$READBACK_ROOT"
+          mkdir -m 700 "$READBACK_ROOT/root"
+
           python3 - <<'PY'
           import json
           import os
+          import stat
           import subprocess
-          from datetime import datetime
           from pathlib import Path
 
           def fail(reason):
@@ -66,27 +103,87 @@ jobs:
           except (OSError, ValueError, KeyError):
               fail('temporary_identity_invalid')
 
-          root = 'gs://qsl-runtime-logs-shared/soxl-p1-p3/06ce97ad581fdc465896ef324ce16ac65f2695af95ec18ce27238bdd71032f74/'
+          source = os.environ['INPUT_ROOT']
+          workspace = Path(os.environ['READBACK_ROOT'])
+          local_root = workspace / 'root'
+          members = {
+              'binding.json': (local_root / 'binding.json', 1024 * 1024),
+              'bars.json': (local_root / 'bars.json', 32 * 1024 * 1024),
+              'manifest.json': (local_root / 'manifest.json', 1024 * 1024),
+              'p1-complete.json': (workspace / 'p1-complete.json', 1024 * 1024),
+          }
+          observed_sizes = {}
+          for name, (_, maximum) in members.items():
+              try:
+                  result = subprocess.run(
+                      ['gcloud', 'storage', 'objects', 'describe', source + name, '--raw', '--format=json'],
+                      capture_output=True, text=True, timeout=30, check=False,
+                  )
+              except (OSError, subprocess.TimeoutExpired):
+                  fail('metadata_read_unavailable')
+              if result.returncode:
+                  fail('metadata_read_unavailable')
+              try:
+                  size = int(json.loads(result.stdout)['size'])
+              except (KeyError, ValueError, TypeError, json.JSONDecodeError):
+                  fail('metadata_invalid')
+              if size <= 0 or size > maximum:
+                  fail('object_size_invalid')
+              observed_sizes[name] = size
+          if sum(observed_sizes.values()) > 35 * 1024 * 1024:
+              fail('read_budget_exceeded')
+
+          for name, (destination, _) in members.items():
+              try:
+                  result = subprocess.run(
+                      ['gcloud', 'storage', 'cp', '--quiet', source + name, str(destination)],
+                      capture_output=True, text=True, timeout=60, check=False,
+                  )
+              except (OSError, subprocess.TimeoutExpired):
+                  fail('object_read_unavailable')
+              if result.returncode:
+                  fail('object_read_unavailable')
+              try:
+                  item_stat = destination.lstat()
+              except OSError:
+                  fail('object_read_unavailable')
+              if not stat.S_ISREG(item_stat.st_mode) or item_stat.st_size != observed_sizes[name]:
+                  fail('object_read_invalid')
+          PY
+
+          cd "$VALIDATOR_SOURCE"
+          ROOT="$READBACK_ROOT/root" COMPLETION="$READBACK_ROOT/p1-complete.json" \
+            env -u UV_PYTHON -u VIRTUAL_ENV uv run --no-sync python - <<'PY'
+          import json
+          import os
+          from pathlib import Path
+
           try:
-              result = subprocess.run(
-                  ['gcloud', 'storage', 'objects', 'describe', root + 'p1-complete.json', '--raw', '--format=json'],
-                  capture_output=True, text=True, timeout=30, check=False,
+              from us_equity_snapshot_pipelines.lifecycle.soxl_core_only_p1_publisher import (
+                  verify_soxl_core_only_input_root,
+                  verify_soxl_core_only_p1_remote_completion,
               )
-          except (OSError, subprocess.TimeoutExpired):
-              fail('metadata_read_unavailable')
-          if result.returncode:
-              fail('metadata_read_unavailable')
-          try:
-              metadata = json.loads(result.stdout)
-              size = int(metadata['size'])
-              created_at = datetime.fromisoformat(metadata['timeCreated'].replace('Z', '+00:00'))
-              if size <= 0 or created_at.utcoffset() is None:
-                  fail('metadata_invalid')
-          except (KeyError, ValueError, TypeError, AttributeError):
-              fail('metadata_invalid')
+              manifest_sha256 = verify_soxl_core_only_input_root(Path(os.environ['ROOT']))
+              if manifest_sha256 != os.environ['EXPECTED_MANIFEST_SHA256']:
+                  raise ValueError
+              if verify_soxl_core_only_p1_remote_completion(
+                  Path(os.environ['ROOT']), Path(os.environ['COMPLETION'])
+              ) != manifest_sha256:
+                  raise ValueError
+          except Exception:
+              print(json.dumps({'status': 'unavailable', 'reason': 'p1_validation_failed'}))
+              raise SystemExit(1) from None
           print(json.dumps({
-              'status': 'available', 'identity': 'github_oidc',
-              'size': size, 'created_at': created_at.isoformat(),
-              'metadata_only': True, 'research_executed': False,
+              'status': 'accepted',
+              'identity': 'github_oidc',
+              'manifest_sha256': manifest_sha256,
+              'member_count': 4,
+              'research_executed': False,
           }))
           PY
+
+      - name: Remove the bounded readback workspace
+        if: always()
+        env:
+          READBACK_ROOT: ${{ runner.temp }}/aab-research-input-${{ github.run_id }}-${{ github.run_attempt }}
+        run: rm -rf -- "$READBACK_ROOT"

tests/test_research_input_readback_workflow.py

@@ -0,0 +1,214 @@
+"""Behavior checks for the bounded VPS P1 input readback workflow."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+import textwrap
+import types
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+WORKFLOW = Path(__file__).resolve().parents[1] / ".github/workflows/research_input_readback.yml"
+OBJECTS = {
+    "binding.json": b"binding",
+    "bars.json": b"bars-data",
+    "manifest.json": b"manifest",
+    "p1-complete.json": b"completion",
+}
+
+
+def workflow_text() -> str:
+    return WORKFLOW.read_text()
+
+
+def python_blocks() -> tuple[str, str]:
+    parts = workflow_text().split("<<'PY'\n")[1:]
+    assert len(parts) == 2
+    blocks = [textwrap.dedent(part.split("\n          PY", maxsplit=1)[0]) for part in parts]
+    return blocks[0], blocks[1]
+
+
+def configure_readback_environment(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
+    credential = tmp_path / "gha-creds.json"
+    credential.write_text('{"type":"external_account"}')
+    workspace = tmp_path / "readback"
+    (workspace / "root").mkdir(parents=True, mode=0o700)
+    monkeypatch.setenv("GOOGLE_APPLICATION_CREDENTIALS", str(credential))
+    monkeypatch.setenv("CLOUDSDK_AUTH_CREDENTIAL_FILE_OVERRIDE", str(credential))
+    monkeypatch.setenv("INPUT_ROOT", "gs://fixed-root/")
+    monkeypatch.setenv("READBACK_ROOT", str(workspace))
+    return workspace
+
+
+def install_fake_validator(
+    monkeypatch: pytest.MonkeyPatch, *, manifest_sha256: str, reject: bool = False
+) -> list[tuple[str, Path, Path | None]]:
+    calls: list[tuple[str, Path, Path | None]] = []
+    package = types.ModuleType("us_equity_snapshot_pipelines")
+    lifecycle = types.ModuleType("us_equity_snapshot_pipelines.lifecycle")
+    publisher = types.ModuleType(
+        "us_equity_snapshot_pipelines.lifecycle.soxl_core_only_p1_publisher"
+    )
+
+    def verify_root(root: Path) -> str:
+        calls.append(("root", root, None))
+        if reject:
+            raise ValueError("hidden input detail")
+        return manifest_sha256
+
+    def verify_completion(root: Path, marker: Path) -> str:
+        calls.append(("completion", root, marker))
+        return manifest_sha256
+
+    publisher.verify_soxl_core_only_input_root = verify_root  # type: ignore[attr-defined]
+    publisher.verify_soxl_core_only_p1_remote_completion = verify_completion  # type: ignore[attr-defined]
+    monkeypatch.setitem(sys.modules, package.__name__, package)
+    monkeypatch.setitem(sys.modules, lifecycle.__name__, lifecycle)
+    monkeypatch.setitem(sys.modules, publisher.__name__, publisher)
+    return calls
+
+
+def test_success_reads_exactly_four_objects_and_calls_both_original_gates(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    download_block, validator_block = python_blocks()
+    workspace = configure_readback_environment(monkeypatch, tmp_path)
+    subprocess_calls: list[list[str]] = []
+
+    def fake_run(args: list[str], **_kwargs: object) -> SimpleNamespace:
+        subprocess_calls.append(args)
+        name = args[4].removeprefix("gs://fixed-root/")
+        if args[2:4] == ["objects", "describe"]:
+            return SimpleNamespace(returncode=0, stdout=json.dumps({"size": len(OBJECTS[name])}))
+        destination = Path(args[-1])
+        destination.write_bytes(OBJECTS[name])
+        return SimpleNamespace(returncode=0, stdout="")
+
+    monkeypatch.setattr(subprocess, "run", fake_run)
+    exec(compile(download_block, "<download-block>", "exec"), {})
+
+    manifest_sha256 = "a" * 64
+    monkeypatch.setenv("ROOT", str(workspace / "root"))
+    monkeypatch.setenv("COMPLETION", str(workspace / "p1-complete.json"))
+    monkeypatch.setenv("EXPECTED_MANIFEST_SHA256", manifest_sha256)
+    validator_calls = install_fake_validator(monkeypatch, manifest_sha256=manifest_sha256)
+    exec(compile(validator_block, "<validator-block>", "exec"), {})
+
+    described = [call[4] for call in subprocess_calls if call[2:4] == ["objects", "describe"]]
+    copied = [call[4] for call in subprocess_calls if call[2:4] == ["cp", "--quiet"]]
+    assert described == ["gs://fixed-root/" + name for name in OBJECTS]
+    assert copied == ["gs://fixed-root/" + name for name in OBJECTS]
+    assert [call[0] for call in validator_calls] == ["root", "completion"]
+    assert json.loads(capsys.readouterr().out) == {
+        "status": "accepted",
+        "identity": "github_oidc",
+        "manifest_sha256": manifest_sha256,
+        "member_count": 4,
+        "research_executed": False,
+    }
+
+
+def test_oversize_object_is_rejected_before_any_copy(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    download_block, _ = python_blocks()
+    configure_readback_environment(monkeypatch, tmp_path)
+    calls: list[list[str]] = []
+
+    def fake_run(args: list[str], **_kwargs: object) -> SimpleNamespace:
+        calls.append(args)
+        return SimpleNamespace(returncode=0, stdout=json.dumps({"size": 1024 * 1024 + 1}))
+
+    monkeypatch.setattr(subprocess, "run", fake_run)
+    with pytest.raises(SystemExit):
+        exec(compile(download_block, "<download-block>", "exec"), {})
+
+    assert not any(call[2:4] == ["cp", "--quiet"] for call in calls)
+    assert json.loads(capsys.readouterr().out) == {
+        "status": "unavailable",
+        "reason": "object_size_invalid",
+    }
+
+
+def test_missing_temporary_identity_never_calls_gcloud(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    download_block, _ = python_blocks()
+    workspace = tmp_path / "readback"
+    (workspace / "root").mkdir(parents=True, mode=0o700)
+    monkeypatch.setenv("READBACK_ROOT", str(workspace))
+    monkeypatch.delenv("GOOGLE_APPLICATION_CREDENTIALS", raising=False)
+    monkeypatch.delenv("CLOUDSDK_AUTH_CREDENTIAL_FILE_OVERRIDE", raising=False)
+
+    def forbidden_run(*_args: object, **_kwargs: object) -> None:
+        raise AssertionError("gcloud must not run")
+
+    monkeypatch.setattr(subprocess, "run", forbidden_run)
+    with pytest.raises(SystemExit):
+        exec(compile(download_block, "<download-block>", "exec"), {})
+
+    assert json.loads(capsys.readouterr().out) == {
+        "status": "unavailable",
+        "reason": "temporary_identity_missing",
+    }
+
+
+def test_copy_failure_outputs_only_the_fixed_safe_category(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    download_block, _ = python_blocks()
+    configure_readback_environment(monkeypatch, tmp_path)
+
+    def fake_run(args: list[str], **_kwargs: object) -> SimpleNamespace:
+        if args[2:4] == ["objects", "describe"]:
+            name = args[4].removeprefix("gs://fixed-root/")
+            return SimpleNamespace(returncode=0, stdout=json.dumps({"size": len(OBJECTS[name])}))
+        return SimpleNamespace(returncode=1, stdout="", stderr="sensitive provider detail")
+
+    monkeypatch.setattr(subprocess, "run", fake_run)
+    with pytest.raises(SystemExit):
+        exec(compile(download_block, "<download-block>", "exec"), {})
+
+    output = capsys.readouterr().out
+    assert json.loads(output) == {"status": "unavailable", "reason": "object_read_unavailable"}
+    assert "sensitive" not in output
+    assert "accepted" not in output
+
+
+def test_validator_rejection_outputs_only_the_fixed_safe_category(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    _, validator_block = python_blocks()
+    manifest_sha256 = "a" * 64
+    monkeypatch.setenv("ROOT", str(tmp_path / "root"))
+    monkeypatch.setenv("COMPLETION", str(tmp_path / "p1-complete.json"))
+    monkeypatch.setenv("EXPECTED_MANIFEST_SHA256", manifest_sha256)
+    install_fake_validator(monkeypatch, manifest_sha256=manifest_sha256, reject=True)
+
+    with pytest.raises(SystemExit):
+        exec(compile(validator_block, "<validator-block>", "exec"), {})
+
+    output = capsys.readouterr().out
+    assert json.loads(output) == {"status": "unavailable", "reason": "p1_validation_failed"}
+    assert "hidden" not in output
+    assert "accepted" not in output
+
+
+def test_workflow_keeps_frozen_source_main_identity_and_no_research_execution() -> None:
+    text = workflow_text()
+
+    assert "github.ref == 'refs/heads/main'" in text
+    assert "repository: QuantStrategyLab/UsEquitySnapshotPipelines" in text
+    assert "ref: ca61b82c2a508a1cc81fb5831294ba9835ac41c2" in text
+    assert "uv sync --locked --no-dev --no-editable --python 3.11" in text
+    assert text.index("name: Install the frozen validator runtime") < text.index(
+        "name: Authenticate for this research input only"
+    )
+    assert "trap cleanup EXIT" in text
+    assert "if: always()" in text
+    assert "upload-artifact" not in text
+    assert "run_soxl_core_only_p3_evidence" not in text