-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathrun_task_internal_command.py
More file actions
210 lines (183 loc) · 8.82 KB
/
Copy pathrun_task_internal_command.py
File metadata and controls
210 lines (183 loc) · 8.82 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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
"""Internal CLI subcommand executed inside the Docker container.
Not part of the public CLI surface -- the host's :class:`DockerRunner`
invokes it via ``docker run``. It loads the staged task + context from
``/work/input``, runs one full evaluation cycle in-process (driver=tempdir),
and writes ``task.json`` + ``task.html`` to ``/work/output``.
The container always exits 0 once ``task.json`` is written, even if the
task itself failed -- criterion failures are signaled via the final_status
field, not the container exit code. Setup failures (missing input,
malformed YAML) exit non-zero before producing task.json so the host can
distinguish them from task-level failures.
"""
from __future__ import annotations
import asyncio
import contextlib
import json
import logging
from pathlib import Path
import typer
from coder_eval.config import settings
from coder_eval.isolation.docker_runner import (
HEARTBEAT_FILENAME,
HEARTBEAT_STALE_SECONDS,
)
from coder_eval.logging_config import setup_logging
from coder_eval.models import (
CONTAINER_INPUT_DIR,
CONTAINER_OUTPUT_DIR,
CONTAINER_TASK_DIR,
ConfigLineageEntry,
PreservationMode,
)
from coder_eval.orchestration.task_loader import load_task
logger = logging.getLogger(__name__)
def heartbeat_is_alive(current: str, last_counter: str, current_mtime: float, last_mtime: float) -> bool:
"""True when the heartbeat shows a fresh signal of life.
Counter advance OR mtime advance counts as alive. The mtime arm covers the empty-counter
startup race (host touch + delayed first write → both reads ""; mtime advanced); the counter
arm covers bind-mount mtime latency (macOS gRPC-FUSE/VirtioFS) where mtime lags the write.
"""
return bool(current and current != last_counter) or current_mtime > last_mtime
def run_task_internal_command(
input_dir: Path = typer.Option( # noqa: B008
Path(CONTAINER_INPUT_DIR),
"--input",
help="Directory containing task.yaml and context.json (bind-mounted by host).",
),
output_dir: Path = typer.Option( # noqa: B008
Path(CONTAINER_OUTPUT_DIR),
"--output",
help="Directory to write task.json/task.html into (bind-mounted by host).",
),
task_dir: Path = typer.Option( # noqa: B008
Path(CONTAINER_TASK_DIR),
"--task-dir",
help="Original task directory mount (used to resolve relative template paths).",
),
verbose: bool = typer.Option(
False,
"--verbose",
"-v",
help="Enable verbose (DEBUG level) logging",
),
) -> None:
"""Run a single staged task inside the container."""
# Use the same logging path as the host CLI so LOG_LEVEL from the
# forwarded env is honoured. Without this, root stays at INFO and the
# DEBUG-level task_log_handler attached by Orchestrator never sees the
# agent's per-tool-call DEBUG records.
log_level = "DEBUG" if verbose else settings.log_level
setup_logging(level=log_level)
# Start the host-heartbeat watchdog: if the host process dies
# ungracefully (SIGKILL, Claude-Code Escape, crash) before it can
# `docker kill` us, the heartbeat file in output_dir goes stale and
# we self-exit -- otherwise the container would keep burning LLM
# budget orphaned. Daemon thread so it doesn't block normal shutdown.
import os as _os
import threading
import time
def _watch_host_heartbeat() -> None:
heartbeat = output_dir / HEARTBEAT_FILENAME
# Grace period for the host to write the first counter value.
time.sleep(HEARTBEAT_STALE_SECONDS)
last_counter = ""
last_mtime = 0.0
last_change = time.monotonic()
while True:
try:
current = heartbeat.read_text(encoding="utf-8")
except (FileNotFoundError, OSError):
current = ""
try:
current_mtime = heartbeat.stat().st_mtime
except (FileNotFoundError, OSError):
current_mtime = 0.0
now = time.monotonic()
if heartbeat_is_alive(current, last_counter, current_mtime, last_mtime):
last_counter = current
last_mtime = current_mtime
last_change = now
if now - last_change > HEARTBEAT_STALE_SECONDS:
logger.error(
"Host heartbeat stale (>%ss); exiting to reap orphan container.",
HEARTBEAT_STALE_SECONDS,
)
# os._exit skips atexit and IO flushing, so the error line
# above would routinely be lost -- making a genuine
# stale-heartbeat suicide indistinguishable from an external
# SIGKILL in the archived logs. Flush best-effort first;
# never let a flush failure stop the exit.
import sys as _sys
for _handler in logging.getLogger().handlers:
with contextlib.suppress(Exception):
_handler.flush()
with contextlib.suppress(Exception):
_sys.stdout.flush()
with contextlib.suppress(Exception):
_sys.stderr.flush()
_os._exit(137)
time.sleep(HEARTBEAT_STALE_SECONDS / 4)
threading.Thread(target=_watch_host_heartbeat, daemon=True).start()
task_yaml = input_dir / "task.yaml"
context_json = input_dir / "context.json"
if not task_yaml.exists():
typer.echo(f"FATAL: missing {task_yaml}", err=True)
raise typer.Exit(2)
if not context_json.exists():
typer.echo(f"FATAL: missing {context_json}", err=True)
raise typer.Exit(2)
context = json.loads(context_json.read_text(encoding="utf-8"))
variant_id: str = context["variant_id"]
replicate_index: int = context.get("replicate_index", 0)
# The host resolves the driver-derived default before dispatch; the container
# obeys it verbatim. This command only ever runs inside the docker driver, so
# a missing key falls back to the docker default (DIRECT_WRITE) — a deliberate
# default, not version back-compat.
preservation_mode = PreservationMode(context.get("preservation_mode", PreservationMode.DIRECT_WRITE.value))
# Docker WORKDIR alignment: the host resolves the concrete WORKDIR
# (config value / "auto" -> `docker inspect` / fallback) and forwards it here.
# Absent -> None -> standard run_dir/artifacts workspace.
workspace_dir_raw = context.get("workspace_dir")
workspace_dir = Path(workspace_dir_raw) if workspace_dir_raw else None
config_lineage = {k: ConfigLineageEntry.model_validate(v) for k, v in (context.get("config_lineage") or {}).items()}
# Prefer the host's raw source_yaml so task.json's audit trail matches
# the in-process driver. Fall back to the staged (post-override) YAML
# for older host versions that didn't forward it.
host_source_yaml: str | None = context.get("source_yaml")
# Load the post-override spec from the staged YAML. We then point
# `task_file` at a path *under the symmetric task_dir mount* so the
# Orchestrator's `task_file.parent` reasoning -- specifically the
# `TASK_DIR` env exposed to `run_command` criteria -- resolves to the
# original host task directory rather than `/work/input/`.
task, source_yaml = load_task(task_yaml)
if host_source_yaml is not None:
source_yaml = host_source_yaml
# The path below is never re-read; it only seeds Orchestrator's TASK_DIR.
runtime_task_file = task_dir / "task.yaml" if task_dir.is_dir() else task_yaml
# Force driver back to tempdir for the actual in-container run.
# We're already inside the container; another nested docker would be
# both wrong and impossible (no docker CLI in image).
if task.sandbox.driver == "docker":
task = task.model_copy(update={"sandbox": task.sandbox.model_copy(update={"driver": "tempdir"})})
output_dir.mkdir(parents=True, exist_ok=True)
# Late import: orchestrator pulls in heavy deps (anthropic SDK etc.)
# that we don't want to load just to print --help.
from coder_eval.orchestrator import Orchestrator
orchestrator = Orchestrator(
task=task,
run_dir=output_dir,
preservation_mode=preservation_mode,
task_file=runtime_task_file,
variant_id=variant_id,
source_yaml=source_yaml,
config_lineage=config_lineage,
replicate_index=replicate_index,
workspace_dir=workspace_dir,
)
# Install the stdout-NDJSON stream callback so per-tool-call events
# reach the host. Late import keeps the streaming module out of the
# default --help path.
from coder_eval.streaming.wire import StdoutNDJsonCallback
orchestrator.stream_callback = StdoutNDJsonCallback()
asyncio.run(orchestrator.run())
# Orchestrator.run() writes task.json to run_dir (== output_dir). Done.