Skip to content

Runtime context and records

RunSpec captures requested experiment inputs. Parameters must be an object and a supplied seed must be an integer. The runner makes a JSON copy of parameters before creating the record.

RunSpec dataclass

Source code in src/benchmark/run_record.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@dataclass(frozen=True)
class RunSpec:
    workflow: str
    scenario: str
    parameters: dict = field(default_factory=dict)
    seed: int | None = None
    parent_attempt_id: str | None = None
    attack: str | None = None
    cleanup: bool = True
    unaligned: bool | str = False

    def __post_init__(self):
        if not isinstance(self.parameters, dict):
            raise ValueError("Scenario parameters must be a JSON object")
        if self.seed is not None and type(self.seed) is not int:
            raise ValueError("Scenario seed must be an integer")

workflow instance-attribute 

workflow: str

scenario instance-attribute 

scenario: str

parameters class-attribute instance-attribute 

parameters: dict = field(default_factory=dict)

seed class-attribute instance-attribute 

seed: int | None = None

parent_attempt_id class-attribute instance-attribute 

parent_attempt_id: str | None = None

attack class-attribute instance-attribute 

attack: str | None = None

cleanup class-attribute instance-attribute 

cleanup: bool = True

unaligned class-attribute instance-attribute 

unaligned: bool | str = False

TriggerReceipt binds a custom trigger to its final event and optional issue/PR/run. Subject kind and number must be provided together; IDs must be positive integers.

TriggerReceipt dataclass

Source code in src/benchmark/run_context.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@dataclass(frozen=True)
class TriggerReceipt:
    event_type: str | None = None
    subject_kind: Literal["pr", "issue"] | None = None
    subject_number: int | None = None
    workflow_run_id: int | None = None

    def __post_init__(self):
        if self.subject_kind not in {None, "pr", "issue"}:
            raise ValueError("Subject kind must be pr or issue")
        if (self.subject_kind is None) != (self.subject_number is None):
            raise ValueError("Subject kind and number must be supplied together")
        for value in (self.subject_number, self.workflow_run_id):
            if value is not None and (type(value) is not int or value <= 0):
                raise ValueError("GitHub identifiers must be positive integers")

event_type class-attribute instance-attribute 

event_type: str | None = None

subject_kind class-attribute instance-attribute 

subject_kind: Literal['pr', 'issue'] | None = None

subject_number class-attribute instance-attribute 

subject_number: int | None = None

workflow_run_id class-attribute instance-attribute 

workflow_run_id: int | None = None

RunContext supplies the recorded spec, mutable scenario state, explicit actor clients, default trigger/collector callbacks, and seeded rng. github(actor) raises if the identity is absent. save_artifact writes JSON inside the attempt. track_repository registers immutable ownership immediately after a custom creation; cleanup_repositories returns deletion error strings while retaining failed resources for inspection.

RunContext dataclass

Source code in src/benchmark/run_context.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@dataclass
class RunContext:
    spec: RunSpec
    record: RunRecord
    state: dict
    actors: dict[str, GitHubClient]
    default_trigger: Callable[[], TriggerReceipt]
    collect_target: Callable[[], dict]
    _repositories: list[tuple[str, str, int]] = field(default_factory=list)

    def __post_init__(self):
        self.rng = random.Random(self.spec.seed)

    @property
    def parameters(self) -> dict:
        return self.spec.parameters

    def github(self, actor: str) -> GitHubClient:
        if actor not in self.actors:
            raise ValueError(f"Required GitHub actor is unavailable: {actor}")
        return self.actors[actor]

    def save_artifact(self, name: str, value):
        return self.record.artifact(name, value)

    def track_repository(self, actor: str, name: str, repository_id: int) -> None:
        """Register a repository immediately after creating it through the raw API."""
        self.github(actor)
        if not name or type(repository_id) is not int or repository_id <= 0:
            raise ValueError("Repository ownership requires a name and immutable ID")
        identity = (actor, name, repository_id)
        if identity not in self._repositories:
            self._repositories.append(identity)
            self.record.event("resource", actor=actor, name=name, id=repository_id, state="created")

    def cleanup_repositories(self) -> list[str]:
        errors = []
        for actor, name, repository_id in self._repositories[:]:
            try:
                self.github(actor)._delete_repository(name, repository_id)
                self.record.event("resource", actor=actor, name=name, id=repository_id, state="deleted")
                self._repositories.remove((actor, name, repository_id))
            except Exception as exc:
                errors.append(str(exc))
        return errors

parameters property 

parameters: dict

github 

github(actor: str) -> GitHubClient
Source code in src/benchmark/run_context.py 
45
46
47
48
def github(self, actor: str) -> GitHubClient:
    if actor not in self.actors:
        raise ValueError(f"Required GitHub actor is unavailable: {actor}")
    return self.actors[actor]

save_artifact 

save_artifact(name: str, value)
Source code in src/benchmark/run_context.py 
50
51
def save_artifact(self, name: str, value):
    return self.record.artifact(name, value)

track_repository 

track_repository(actor: str, name: str, repository_id: int) -> None

Register a repository immediately after creating it through the raw API.

Source code in src/benchmark/run_context.py 
53
54
55
56
57
58
59
60
61
def track_repository(self, actor: str, name: str, repository_id: int) -> None:
    """Register a repository immediately after creating it through the raw API."""
    self.github(actor)
    if not name or type(repository_id) is not int or repository_id <= 0:
        raise ValueError("Repository ownership requires a name and immutable ID")
    identity = (actor, name, repository_id)
    if identity not in self._repositories:
        self._repositories.append(identity)
        self.record.event("resource", actor=actor, name=name, id=repository_id, state="created")

cleanup_repositories 

cleanup_repositories() -> list[str]
Source code in src/benchmark/run_context.py 
63
64
65
66
67
68
69
70
71
72
def cleanup_repositories(self) -> list[str]:
    errors = []
    for actor, name, repository_id in self._repositories[:]:
        try:
            self.github(actor)._delete_repository(name, repository_id)
            self.record.event("resource", actor=actor, name=name, id=repository_id, state="deleted")
            self._repositories.remove((actor, name, repository_id))
        except Exception as exc:
            errors.append(str(exc))
    return errors

RunRecord creates a UUID attempt directory and manifest, snapshots input bytes with SHA-256 hashes, appends durable journal events, and saves JSON artifacts with hashes. artifact rejects paths escaping the attempt. It is a local record, not a resume/reconciliation engine.

RunRecord

Source code in src/benchmark/run_record.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class RunRecord:
    def __init__(self, workspace_dir: str, spec: RunSpec):
        self.attempt_id = uuid.uuid4().hex
        self.timestamp = datetime.now(timezone.utc).isoformat()
        self.directory = Path(workspace_dir) / "runs" / self.attempt_id
        self.directory.mkdir(parents=True)
        self.manifest = {
            "schema_version": 1,
            "attempt_id": self.attempt_id,
            "timestamp": self.timestamp,
            "spec": asdict(spec),
            "inputs": {},
        }
        try:
            revision = subprocess.run(
                ["git", "rev-parse", "HEAD"],
                cwd=workspace_dir,
                capture_output=True,
                text=True,
                check=True,
            ).stdout.strip()
            self.manifest["source_revision"] = revision
            status = subprocess.run(
                ["git", "status", "--porcelain"],
                cwd=workspace_dir,
                capture_output=True,
                text=True,
                check=True,
            ).stdout
            self.manifest["source_dirty"] = bool(status.strip())
        except (OSError, subprocess.CalledProcessError):
            self.manifest["source_revision"] = None
        self.save_manifest()
        self.event("phase", phase="created")

    def save_manifest(self) -> None:
        write_json(self.directory / "manifest.json", self.manifest)

    def event(self, kind: str, **data) -> None:
        entry = {"timestamp": datetime.now(timezone.utc).isoformat(), "kind": kind, **data}
        with (self.directory / "events.jsonl").open("a") as handle:
            handle.write(json.dumps(entry, allow_nan=False) + "\n")
            handle.flush()
            os.fsync(handle.fileno())

    def snapshot(self, label: str, source: str | Path, *, prefix: str = "") -> None:
        source = Path(source)
        files = sorted(source.rglob("*")) if source.is_dir() else [source]
        hashes = dict(self.manifest["inputs"].get(label, {}))
        for path in files:
            if not path.is_file() or "__pycache__" in path.parts:
                continue
            relative = path.relative_to(source) if source.is_dir() else Path(path.name)
            relative = Path(prefix) / relative
            content = path.read_bytes()
            destination = self.directory / "inputs" / label / relative
            destination.parent.mkdir(parents=True, exist_ok=True)
            destination.write_bytes(content)
            hashes[str(relative)] = hashlib.sha256(content).hexdigest()
        self.manifest["inputs"][label] = hashes
        self.save_manifest()

    def artifact(self, name: str, value) -> Path:
        root = self.directory / "artifacts"
        path = root / name
        if path.resolve() == root.resolve() or root.resolve() not in path.resolve().parents:
            raise ValueError("Artifact path must stay inside the attempt")
        path.parent.mkdir(parents=True, exist_ok=True)
        write_json(path, value)
        self.event(
            "artifact",
            path=str(path.relative_to(self.directory)),
            sha256=hashlib.sha256(path.read_bytes()).hexdigest(),
        )
        return path

__init__ 

__init__(workspace_dir: str, spec: RunSpec)
Source code in src/benchmark/run_record.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def __init__(self, workspace_dir: str, spec: RunSpec):
    self.attempt_id = uuid.uuid4().hex
    self.timestamp = datetime.now(timezone.utc).isoformat()
    self.directory = Path(workspace_dir) / "runs" / self.attempt_id
    self.directory.mkdir(parents=True)
    self.manifest = {
        "schema_version": 1,
        "attempt_id": self.attempt_id,
        "timestamp": self.timestamp,
        "spec": asdict(spec),
        "inputs": {},
    }
    try:
        revision = subprocess.run(
            ["git", "rev-parse", "HEAD"],
            cwd=workspace_dir,
            capture_output=True,
            text=True,
            check=True,
        ).stdout.strip()
        self.manifest["source_revision"] = revision
        status = subprocess.run(
            ["git", "status", "--porcelain"],
            cwd=workspace_dir,
            capture_output=True,
            text=True,
            check=True,
        ).stdout
        self.manifest["source_dirty"] = bool(status.strip())
    except (OSError, subprocess.CalledProcessError):
        self.manifest["source_revision"] = None
    self.save_manifest()
    self.event("phase", phase="created")

save_manifest 

save_manifest() -> None
Source code in src/benchmark/run_record.py 
75
76
def save_manifest(self) -> None:
    write_json(self.directory / "manifest.json", self.manifest)

event 

event(kind: str, **data) -> None
Source code in src/benchmark/run_record.py 
78
79
80
81
82
83
def event(self, kind: str, **data) -> None:
    entry = {"timestamp": datetime.now(timezone.utc).isoformat(), "kind": kind, **data}
    with (self.directory / "events.jsonl").open("a") as handle:
        handle.write(json.dumps(entry, allow_nan=False) + "\n")
        handle.flush()
        os.fsync(handle.fileno())

snapshot 

snapshot(label: str, source: str | Path, *, prefix: str = '') -> None
Source code in src/benchmark/run_record.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def snapshot(self, label: str, source: str | Path, *, prefix: str = "") -> None:
    source = Path(source)
    files = sorted(source.rglob("*")) if source.is_dir() else [source]
    hashes = dict(self.manifest["inputs"].get(label, {}))
    for path in files:
        if not path.is_file() or "__pycache__" in path.parts:
            continue
        relative = path.relative_to(source) if source.is_dir() else Path(path.name)
        relative = Path(prefix) / relative
        content = path.read_bytes()
        destination = self.directory / "inputs" / label / relative
        destination.parent.mkdir(parents=True, exist_ok=True)
        destination.write_bytes(content)
        hashes[str(relative)] = hashlib.sha256(content).hexdigest()
    self.manifest["inputs"][label] = hashes
    self.save_manifest()

artifact 

artifact(name: str, value) -> Path
Source code in src/benchmark/run_record.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
def artifact(self, name: str, value) -> Path:
    root = self.directory / "artifacts"
    path = root / name
    if path.resolve() == root.resolve() or root.resolve() not in path.resolve().parents:
        raise ValueError("Artifact path must stay inside the attempt")
    path.parent.mkdir(parents=True, exist_ok=True)
    write_json(path, value)
    self.event(
        "artifact",
        path=str(path.relative_to(self.directory)),
        sha256=hashlib.sha256(path.read_bytes()).hexdigest(),
    )
    return path

write_json uses a temporary sibling file, flushes it, and replaces the destination. Values must be JSON serializable; non-finite numbers are rejected.

write_json 

write_json(path: Path, value) -> None
Source code in src/benchmark/run_record.py 
31
32
33
34
35
36
37
def write_json(path: Path, value) -> None:
    temporary = path.with_name(path.name + ".tmp")
    with temporary.open("w") as handle:
        json.dump(value, handle, indent=2, allow_nan=False)
        handle.flush()
        os.fsync(handle.fileno())
    temporary.replace(path)