"""Run one worker CLI and record what happened. Shared by every provider: the entrypoint scripts parse arguments and hand over here. What differs per provider is the command (an ``ExecutionStrategy``); what differs per surface is the presentation. Idle is measured from stream arrival, never from the log file's mtime. The screen deliberately drops thinking events, so an mtime-based watchdog would SIGTERM a healthy worker in the middle of a long reasoning stretch. """ from __future__ import annotations import dataclasses import json import os import selectors import signal import subprocess import time from pathlib import Path from typing import Any, Callable, Mapping from .domain.provider import ServedModelAttestation, ServedModelNormalizer from .domain.worker_exec import ( NO_RESULT_EXIT_CODE, SERVED_MODEL_MISMATCH_EXIT_CODE, ExecCommand, ExecutionStrategy, WorkerExecRequest, ) from .domain.worker_presentation import JsonEvents, Presentation from .session_transcript import SessionTranscript from .json_boundary import JsonBoundaryError, write_owned_object_atomic LIVE = "live" QUIET = "quiet" _SELECT_TIMEOUT_SECONDS = 0.25 _TERM_GRACE_SECONDS = 5 # How long the streams stay open after the worker process itself is gone. A pipe # holds at most its capacity of already-written output when its writer exits, so # this only has to cover a drain, never a producer. _DRAIN_AFTER_EXIT_SECONDS = 2 _TIMEOUT_EXIT_CODE = 124 _READ_SIZE = 8192 _WRITE_SIZE = 8192 _NO_STATUS_EXTRA: Mapping[str, Any] = {} # The signals that end this process without raising anything Python can catch on # the way out. SIGINT is absent on purpose: it arrives as KeyboardInterrupt and # the exception path already closes the sidecar. _ABNORMAL_SIGNALS = (signal.SIGTERM, signal.SIGHUP) def run_worker( strategy: ExecutionStrategy, request: WorkerExecRequest, *, presentation: str, log_path: Path, status_path: Path, status_extra: Mapping[str, Any] = _NO_STATUS_EXTRA, served_model_normalizer: ServedModelNormalizer | None = None, served_model_verifier: Callable[[ServedModelAttestation], str] | None = None, ) -> int: _validate_status_extra(status_extra) _validate_request_write_contract(request, status_extra) command = strategy.build_command(request) started_monotonic = time.monotonic() status = _started_status(status_extra, log_path) _write_status(status_path, status) guard = _AbnormalExit(status_path, status, started_monotonic) try: with guard: exit_code, timed_out, idle_seconds, raw_model, usage = _launch( command, log_path, presentation=presentation, idle_timeout_seconds=request.idle_timeout_seconds, on_spawn=guard.watch, ) at_exit = getattr(command.presentation, "served_model_at_exit", None) if raw_model is None and at_exit is not None and request.session_id: raw_model = at_exit(request.session_id, command.cwd) except BaseException as exc: # Whatever ended this run — an OS error, a Ctrl-C, a bug in this file — # the sidecar has to stop saying `started`. Nothing downstream rewrites # it, so `worker_liveness` would read the worker as still working # forever. No exit code is invented — its absence is already how every # reader spells a run that failed. The exits that never raise at all are # the guard's to close. guard.close(f"{type(exc).__name__}: {exc}") raise status = _closed(status, started_monotonic) if usage is not None: # 스트림이 마지막으로 보고한 토큰 스냅샷, 공급자 어휘 그대로. 토큰 # 수집기(`okstra_token_usage`)가 홈 트랜스크립트가 없는 공급자의 # 사용량을 여기서 읽는다. status["usage"] = dict(usage) if served_model_normalizer is not None: attestation = served_model_normalizer(raw_model) status["servedModelAttestation"] = _attestation_payload(attestation) if served_model_verifier is not None: failure = served_model_verifier(attestation) if failure: status["failure"] = failure exit_code = SERVED_MODEL_MISMATCH_EXIT_CODE status["exit_code"] = exit_code if timed_out: status.update( timeout=True, idle_at_ts=status["ended_ts"], idle_seconds=idle_seconds, terminated_by="idle-watchdog", ) _write_status(status_path, status) return exit_code def _validate_status_extra(status_extra: Mapping[str, Any]) -> None: schema = status_extra.get("schemaVersion") identity_version = status_extra.get("executionIdentityVersion") identity_keys = { "participantRef", "roleExecutionRef", "executionLabel", "dutyId", "invocationRef", } if schema is None and identity_version is None: if set(status_extra) & ( identity_keys | {"attempt", "servedModelAttestation"} ): raise ValueError("worker status mixes v1 and v2 execution identity") return if schema != "2.0" or identity_version != 2: raise ValueError("worker status mixes v1 and v2 execution identity") if any( not isinstance(status_extra.get(key), str) or not status_extra[key] for key in identity_keys ): raise ValueError("worker status v2 execution identity is incomplete") attempt = status_extra.get("attempt") if not isinstance(attempt, int) or isinstance(attempt, bool) or attempt < 1: raise ValueError("worker status v2 attempt must be positive") attestation = status_extra.get("servedModelAttestation") if not isinstance(attestation, Mapping) or set(attestation) != { "observedModel", "normalizedModelRef", "level", "source", }: raise ValueError("worker status served model attestation is invalid") def _validate_request_write_contract( request: WorkerExecRequest, status_extra: Mapping[str, Any] ) -> None: if status_extra.get("executionIdentityVersion") != 2: return policy = request.policy.write_policy enforcement = request.policy.write_enforcement if policy is None or enforcement is None: raise ValueError("v2 worker request has no write contract") if status_extra.get("writePolicyDigest") != policy.digest: raise ValueError("v2 worker request write policy digest changed") if status_extra.get("writeEnforcement") != enforcement.to_payload(): raise ValueError("v2 worker request write enforcement changed") def _launch( command: ExecCommand, log_path: Path, *, presentation: str, idle_timeout_seconds: int, on_spawn: Callable[[subprocess.Popen[bytes]], None], ) -> tuple[int, bool, int, str | None, Mapping[str, Any] | None]: live = presentation == LIVE transcript = SessionTranscript(log_path, live=live) observation = _ServedModelObservation() usage_observation = _UsageObservation() strategy = _with_stream_observations( command.presentation, observation, usage_observation ) try: # The strategy decided where this provider runs — some CLIs work in the # stage tree, others in the project root and reach the tree by flag. process = subprocess.Popen( list(command.argv), cwd=str(command.cwd), # 보낼 것이 없으면 DEVNULL 이다. `None` 은 부모의 stdin 을 물려주는 # 것이라, stdin 을 읽는 CLI 가 터미널에서 실행됐을 때 EOF 를 못 받고 # 영원히 기다린다 — 워커는 대화형이 아니므로 물려줄 이유가 없다. stdin=( subprocess.PIPE if command.stdin_text is not None else subprocess.DEVNULL ), stdout=subprocess.PIPE, stderr=_stderr_target(strategy), start_new_session=True, env=_child_env(command.environment), ) on_spawn(process) exit_code, timed_out, idle_seconds = _pump( process, transcript, strategy=strategy, presentation=presentation, idle_timeout_seconds=idle_timeout_seconds, stdin_text=command.stdin_text, ) return ( exit_code, timed_out, idle_seconds, observation.raw_model, usage_observation.usage, ) finally: transcript.close() def _with_stream_observations( presentation: Presentation, observation: _ServedModelObservation, usage_observation: _UsageObservation, ) -> Presentation: """JSON 경로의 서빙 모델·토큰 사용량 관측을 러너가 모아 둔다. 해석 전략은 이벤트를 보고 모델 문자열과 사용량 스냅샷만 돌려준다. 그 값을 사이드카에 적는 일은 러너의 것이라, 여기서 한 번 감싼다. 사용량은 어댑터가 읽는 법을 넘긴 어휘에서만 관측된다. """ if not isinstance(presentation, JsonEvents): return presentation original = presentation.observe original_usage = presentation.observe_usage def observe(event: Mapping[str, Any]) -> str | None: observed = original(event) observation.record(observed) if original_usage is not None: usage_observation.record(original_usage(event)) return observed # 필드를 손으로 옮기지 않는다 — 그렇게 하던 동안 어댑터가 선언한 # `is_conclusion` 이 여기서 떨어져 나가, grok 의 종결 판정이 무효였다. return dataclasses.replace(presentation, observe=observe) class _ServedModelObservation: def __init__(self) -> None: self.raw_model: str | None = None def record(self, raw_model: str | None) -> None: if self.raw_model is None and isinstance(raw_model, str) and raw_model.strip(): self.raw_model = raw_model class _UsageObservation: """스트림이 보고한 마지막 사용량 스냅샷. 첫 값이 아니라 마지막 값이다 — 단계별 스냅샷을 내는 CLI 도 종결 이벤트에 합계를 싣고, 그 이벤트가 스트림의 끝에 온다. """ def __init__(self) -> None: self.usage: Mapping[str, Any] | None = None def record(self, usage: Mapping[str, Any] | None) -> None: if isinstance(usage, Mapping) and usage: self.usage = usage class _AbnormalExit: """Close the status sidecar for the exits that raise nothing at all. ``except BaseException`` around the run covers an exception, including the ``KeyboardInterrupt`` a SIGINT raises. It does not cover SIGTERM or SIGHUP: their default disposition ends the process outright, and nothing in this file runs (measured — a bash ``trap … EXIT`` does fire on SIGTERM, which is why the shell wrappers needed no equivalent of this class). Those two are the common abnormal exits: a pane close, ``okstra team reclaim`` / ``okstra team teardown``, session teardown. Without this the sidecar stays at ``started`` and ``worker_liveness`` reads a dead worker as a working one. SIGKILL and a host crash remain uncovered because nothing can cover them. A sidecar still reading ``started`` is the residue they leave, and no reader should take it as proof the worker is alive. """ def __init__( self, status_path: Path, status: Mapping[str, Any], started_monotonic: float ) -> None: self._status_path = status_path self._status = status self._started_monotonic = started_monotonic self._process: subprocess.Popen[bytes] | None = None self._restore: dict[int, Any] = {} def watch(self, process: subprocess.Popen[bytes]) -> None: """Adopt the child, so a signal tears down its group rather than orphan it.""" self._process = process def close(self, failure: str) -> None: _write_status( self._status_path, {**_closed(self._status, self._started_monotonic), "failure": failure}, ) def __enter__(self) -> _AbnormalExit: for number in _ABNORMAL_SIGNALS: try: self._restore[number] = signal.signal(number, self._handle) except ValueError: # Handlers install from the main thread only. A caller running # the runner off-thread keeps the exception path and nothing # more, which is what it had before this class existed. break return self def __exit__(self, *_exception: Any) -> bool: for number, previous in self._restore.items(): signal.signal(number, previous) self._restore.clear() return False def _handle(self, number: int, _frame: Any) -> None: if self._process is not None: _terminate(self._process) self.close(f"signal {signal.Signals(number).name}") # Then die the way the sender asked, so the exit code still names the # signal instead of reporting a clean stop this run did not make. signal.signal(number, signal.SIG_DFL) os.kill(os.getpid(), number) def _closed(status: Mapping[str, Any], started_monotonic: float) -> dict[str, Any]: """The sidecar's terminal shape, whatever it was that ended the run. ``stage`` is the field every reader keys on — ``wrapper_status.is_terminal``, the pane reclaim and the dispatch record all ask whether it reads ``exited``. """ return { **status, "stage": "exited", "ended_ts": int(time.time()), "duration_ms": int((time.monotonic() - started_monotonic) * 1000), } def _started_status(status_extra: Mapping[str, Any], log_path: Path) -> dict[str, Any]: """The sidecar as it looks before the CLI has produced anything. ``status_extra`` carries what only the entrypoint knows — which wrapper the caller invoked and which role it dispatched — and is placed where the wrapper-written sidecars already carry those keys. """ return { "schemaVersion": 1, **dict(status_extra), "pid": os.getpid(), "started_ts": int(time.time()), "log_path": str(log_path), "stage": "started", } def _stderr_target(presentation: Presentation) -> int: """stderr 를 stdout 에 합칠지는 전략이 정한다. 한 스트림에 진행과 결과가 함께 오는 CLI 는 합쳐 읽어야 순서가 보존되고, 둘을 갈라 내는 CLI 는 갈라 읽어야 답과 진행이 섞이지 않는다. """ return subprocess.STDOUT if presentation.merges_stderr() else subprocess.PIPE def _pump( process: subprocess.Popen[bytes], transcript: SessionTranscript, *, strategy: Presentation, presentation: str, idle_timeout_seconds: int, stdin_text: str | None = None, ) -> tuple[int, bool, int]: selector = selectors.DefaultSelector() readers = _register_streams(selector, process, strategy, transcript) outgoing = _register_prompt(selector, process, stdin_text) closing_text: str | None = None last_output = time.monotonic() timed_out = False idle_seconds = 0 drain_deadline: float | None = None while selector.get_map(): for key, events in selector.select(timeout=_SELECT_TIMEOUT_SECONDS): if events & selectors.EVENT_WRITE: outgoing = _push_prompt(selector, key, outgoing) continue chunk = os.read(key.fd, _READ_SIZE) if not chunk: selector.unregister(key.fileobj) continue # Either stream is proof of life: a text CLI reports progress for # minutes before a result exists to send. last_output = time.monotonic() closing_text = readers[key.fd].feed(chunk) or closing_text idle_seconds = int(time.monotonic() - last_output) if ( idle_timeout_seconds and idle_seconds >= idle_timeout_seconds and process.poll() is None ): timed_out = True _terminate(process) drain_deadline = _drain_deadline(process, drain_deadline) if drain_deadline is not None and time.monotonic() >= drain_deadline: _stop_reading(selector) for reader in readers.values(): closing_text = reader.flush() or closing_text strategy.flush(transcript) exit_code = process.wait() if isinstance(strategy, JsonEvents) and not strategy.concluded(): # JSON 스트림은 그 어휘의 종결 이벤트로 끝나야 한다. 텍스트를 흘리는 # CLI 에는 놓칠 종결 이벤트가 없다. 무엇이 종결인지는 전략이 안다. transcript.note("no result event in the CLI's output") if exit_code == 0 and not timed_out: # 실측(dev-10341, 한 CLI 4회): 종결 이벤트도 산출물도 없이 exit 0. # 그대로 돌려주면 사이드카와 셸 소비자에게는 성공으로 읽힌다. # CLI 자신의 비영 코드는 보존한다 — 그쪽이 더 구체적인 사실이다. exit_code = NO_RESULT_EXIT_CODE if closing_text is not None: # Result 는 format_* 가 줄을 만들지 않는다. 아카이브 결론은 러너가 적는다. # 상한 밖에 둔다 — 잘린 결론은 사후 분석 전체를 잃게 한다. if isinstance(strategy, JsonEvents): transcript.write("worker", closing_text, capped=False) if presentation == QUIET: print(closing_text, flush=True) return (_TIMEOUT_EXIT_CODE if timed_out else exit_code), timed_out, idle_seconds def _drain_deadline( process: subprocess.Popen[bytes], deadline: float | None ) -> float | None: """When to stop reading, once the worker process itself is gone. Without this the loop ends only at EOF on every stream, and a worker that leaves a background process behind — a dev server, a file watcher, the watcher the container-build skill starts on purpose — leaves that process holding the write end of the pipe open for as long as it lives. The shell wrappers returned as soon as `wait ` did, so blocking on a grandchild is a regression rather than a policy, and the exit code the caller gets stays the worker's own. The deadline is set once and never pushed back: output arriving after the worker exited is the grandchild's, and letting it extend the wait would restore exactly the hang this bounds. """ if deadline is not None or process.poll() is None: return deadline return time.monotonic() + _DRAIN_AFTER_EXIT_SECONDS def _stop_reading(selector: selectors.BaseSelector) -> None: """Drop every remaining stream without signalling whoever still holds it. A process the worker deliberately left running is not this runner's to kill. The idle watchdog is what ends a run that went wrong; this only ends the reading of a run that already finished. """ for key in list(selector.get_map().values()): selector.unregister(key.fileobj) def _register_streams( selector: selectors.BaseSelector, process: subprocess.Popen[bytes], presentation: Presentation, transcript: SessionTranscript, ) -> dict[int, _LineReader]: """이 자식이 말하는 스트림마다 전략이 지정한 싱크를 붙인다.""" assert process.stdout is not None streams = {"stdout": process.stdout, "stderr": process.stderr} readers: dict[int, _LineReader] = {} for channel, sink in presentation.sinks(transcript): stream = streams[channel] assert stream is not None, channel selector.register(stream, selectors.EVENT_READ) readers[stream.fileno()] = _LineReader(sink) return readers class _LineReader: """Whole lines out of one byte stream, handed to that stream's sink. One per stream rather than one shared buffer: the two streams of a text CLI arrive interleaved, and a shared buffer would splice half a progress line onto the front of the result. """ def __init__(self, emit: Callable[[str], str | None]) -> None: self._emit = emit self._pending = b"" def feed(self, chunk: bytes) -> str | None: self._pending += chunk *complete, self._pending = self._pending.split(b"\n") return self._drain(complete) def flush(self) -> str | None: """Whatever the stream ended on without a closing newline.""" trailing, self._pending = self._pending, b"" return self._drain([trailing] if trailing else []) def _drain(self, raw_lines: list[bytes]) -> str | None: closing: str | None = None for raw in raw_lines: closing = self._emit(raw.decode("utf-8", "replace")) or closing return closing def _register_prompt( selector: selectors.BaseSelector, process: subprocess.Popen[bytes], stdin_text: str | None, ) -> bytes: """Queue the prompt for a CLI that reads it from stdin. Fed inside the pump rather than written whole before it. A prompt larger than the pipe buffer — worker prompts are tens of kilobytes — would otherwise block this process while the CLI blocks writing the stdout that nobody is draining yet, and the idle watchdog could not even reach that stall because it only runs once the pump is looping. """ if stdin_text is None or process.stdin is None: return b"" os.set_blocking(process.stdin.fileno(), False) selector.register(process.stdin, selectors.EVENT_WRITE) return stdin_text.encode("utf-8") def _push_prompt( selector: selectors.BaseSelector, key: selectors.SelectorKey, payload: bytes ) -> bytes: """Hand over as much of the prompt as the pipe will take right now.""" try: written = os.write(key.fd, payload[:_WRITE_SIZE]) if payload else 0 except BlockingIOError: return payload except OSError: # The CLI exited before reading its prompt. Its exit code is the story; # this half-delivered write is not. _close_prompt(selector, key) return b"" payload = payload[written:] if not payload: # EOF is what tells the CLI its prompt is complete. _close_prompt(selector, key) return payload def _close_prompt(selector: selectors.BaseSelector, key: selectors.SelectorKey) -> None: selector.unregister(key.fileobj) key.fileobj.close() def _attestation_payload(attestation: ServedModelAttestation) -> dict[str, Any]: return { "observedModel": attestation.observed_model, "normalizedModelRef": attestation.normalized_model_ref, "level": attestation.level, "source": attestation.source, } def _terminate(process: subprocess.Popen[bytes]) -> None: """SIGTERM the worker's process group, then SIGKILL whatever survives. The group, not the process: a CLI spawns shells and build tools, and signalling only the direct child leaves those running with the pipe open. ``start_new_session=True`` at spawn is what makes the group addressable by the child's own pid. """ try: os.killpg(process.pid, signal.SIGTERM) except ProcessLookupError: return try: process.wait(timeout=_TERM_GRACE_SECONDS) except subprocess.TimeoutExpired: try: os.killpg(process.pid, signal.SIGKILL) except ProcessLookupError: # expected-miss: SIGTERM 유예 사이에 그룹이 스스로 끝난 경우다. # 죽이려던 대상이 이미 없는 것은 성공이지 실패가 아니다. pass process.wait() def _child_env(overrides: Mapping[str, str | None]) -> dict[str, str]: """The worker's environment with git's fsmonitor disabled. The main worktree's fsmonitor daemon leaks its IPC socket into task worktrees, so git status/commit there intermittently fail with `fsmonitor_ipc__send_query`. GIT_CONFIG_* scopes the override to this process tree without touching the user's repo config, appending after any GIT_CONFIG_* the caller already set. """ env = dict(os.environ) for key, value in overrides.items(): if value is None: env.pop(key, None) else: env[key] = value index = int(env.get("GIT_CONFIG_COUNT", "0") or "0") env[f"GIT_CONFIG_KEY_{index}"] = "core.fsmonitor" env[f"GIT_CONFIG_VALUE_{index}"] = "false" env["GIT_CONFIG_COUNT"] = str(index + 1) return env def write_host_event( path: Path, *, invocation_ref: str, attempt: int, event_type: str, payload: Mapping[str, Any], ) -> Mapping[str, Any]: """Append one orchestrator-owned host event from the runner boundary.""" from .attempt_evidence import HostEventStreamWriter return HostEventStreamWriter( path, invocation_ref=invocation_ref, attempt=attempt ).append(event_type, payload) def _write_status(path: Path, status: Mapping[str, Any]) -> None: """Best-effort status write; a sidecar failure must not break the run. Replaced whole rather than rewritten in place: readers poll this file while the worker is still running and must never see a half-written document. """ try: write_owned_object_atomic( path, status, artifact="worker wrapper status", ) except (OSError, JsonBoundaryError): return