-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathinvocation_log.py
More file actions
151 lines (124 loc) · 5.3 KB
/
Copy pathinvocation_log.py
File metadata and controls
151 lines (124 loc) · 5.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
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
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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
"""The structured invocation log: the recording shim that writes it, and the reader.
Named for the artifact rather than the writer because both sides live here -- the
shim template `SandboxConfig.record_cli` renders, and `parse_log`, which the
`cli_called` criterion reads it back with.
The rendered script runs INSIDE the sandbox, where ``coder_eval`` is not
installed, so it imports nothing from this package: its configuration arrives as
embedded literals and everything else comes from the standard library.
Keeping the template here rather than inline in :mod:`coder_eval.sandbox` lets
:func:`render_recorder` be exercised directly (render, execute, read the log)
without standing up a sandbox.
"""
import json
import sys
from coder_eval.models import RecordedCli
# Written beside the shims, inside the generated recorder directory, so the log
# travels with them if the sandbox root moves.
LOG_FILENAME = "calls.jsonl"
_TEMPLATE = '''\
#!{interpreter}
"""Recording shim for `{tool}` - generated by coder_eval SandboxConfig.record_cli.
Appends one JSON record per invocation to {log_filename} beside this script, in
the format the `cli_called` success criterion reads. Do not edit: regenerated on
every sandbox setup.
"""
import json
import os
import sys
import time
TOOL = {tool!r}
EXIT_CODE = {exit_code!r}
STDOUT_TEXT = {stdout!r}
STDERR_TEXT = {stderr!r}
SHIM_DIR = os.path.dirname(os.path.abspath(__file__))
LOG_PATH = os.path.join(SHIM_DIR, {log_filename!r})
LOG_ERROR_PATH = LOG_PATH + ".error"
def record(argv, exit_code):
"""Append this invocation to the log.
Best-effort: a logging failure must never break the command the agent ran,
which would turn an evidence problem into a behaviour problem.
argv is stored as a LIST. A space-joined string cannot distinguish
`--flag "two words"` from two arguments, which is the whole reason this log
exists instead of a flattened command line. stdin is deliberately never
read: it would block whenever the sandbox leaves it on an open pipe, and in
passthrough mode it would consume the payload the real tool needs.
"""
entry = {{
"ts": round(time.time(), 3),
"tool": TOOL,
"argv": list(argv),
"exit": exit_code,
}}
try:
# ensure_ascii escapes non-ASCII and any stray surrogate from
# undecodable argv bytes, so an exotic argument cannot make this write
# raise and silently drop the record.
with open(LOG_PATH, "a", encoding="utf-8", newline="\\n") as handle:
handle.write(json.dumps(entry) + "\\n")
except OSError as exc:
# Keep the agent's command working, but never lose a record silently: a
# dropped record reads exactly like "the agent never ran it".
sys.stderr.write("coder_eval recorder: log write failed: %r\\n" % (exc,))
try:
with open(LOG_ERROR_PATH, "a", encoding="utf-8") as sentinel:
sentinel.write("%r %r\\n" % (exc, argv))
except OSError:
pass
def main(argv):
"""Record the invocation, then fail like the tool would with nothing behind it.
Nothing is executed: no network, no auth, no side effects. A test that needs
the real tool's behavior recorded instead should supply its own wrapper under
mock_path_dirs -- proxying a live executable is a different job from stubbing
one, and this shim deliberately does only the second.
"""
record(argv[1:], EXIT_CODE)
if STDOUT_TEXT:
sys.stdout.write(STDOUT_TEXT)
if STDERR_TEXT:
sys.stderr.write(STDERR_TEXT)
return EXIT_CODE
if __name__ == "__main__":
sys.exit(main(sys.argv))
'''
def render_recorder(spec: RecordedCli, interpreter: str | None = None) -> str:
"""Render the shim source for one ``record_cli`` entry.
``interpreter`` is baked into the shebang as an ABSOLUTE path (defaulting to
the running interpreter). A ``#!/usr/bin/env python3`` shebang resolves
through the same PATH the recorder dir is prepended to, so `tool: python3`
made the shim re-exec itself forever.
"""
return _TEMPLATE.format(
interpreter=interpreter or sys.executable,
tool=spec.tool,
exit_code=spec.exit_code,
stdout=spec.stdout,
stderr=spec.stderr,
log_filename=LOG_FILENAME,
)
def parse_log(text: str) -> tuple[list[tuple[list[str], dict[str, object]]], int]:
"""Parse recorder-log text into ``(usable, unusable_count)``.
A usable entry pairs the record's ``argv`` with the whole record. Unusable
means unparseable, not an object, or an ``argv`` that is not a list of
strings — counted rather than dropped, because a record that cannot be read
might be the very call a negative guard forbids.
"""
usable: list[tuple[list[str], dict[str, object]]] = []
unusable = 0
for line in text.splitlines():
stripped = line.strip()
if not stripped:
continue
try:
parsed = json.loads(stripped)
except ValueError:
unusable += 1
continue
if not isinstance(parsed, dict):
unusable += 1
continue
argv = parsed.get("argv")
if isinstance(argv, list) and all(isinstance(item, str) for item in argv):
usable.append((argv, parsed))
else:
unusable += 1
return usable, unusable