diff --git a/.agent/handoff/240.md b/.agent/handoff/240.md new file mode 100644 index 0000000..433542c --- /dev/null +++ b/.agent/handoff/240.md @@ -0,0 +1,32 @@ +# Handoff note: issue #240 + +- issue: 240 +- started_at: 2026-10-14T12:03:00.200226Z +- reason: token_soft_threshold +- last_work_commit: 69b15f1d9bf87639d1942a27c594996f269f38e8 + +## Summary + +Implemented the telemetry tracer bullet (agent-side `telemetry` over the control socket). All new tests pass (22/22 in `tests/test_telemetry.py`), full suite passes (1070 passed, 10 skipped), and `ruff check` is clean. The `code-review` skill was invoked in this run: Standards axis found 0 hard violations + 4 judgement-call smells; Spec axis found 0 missing requirements plus 2 partial gaps, 2 scope-creep notes, and 2 fragility findings (full report in the run transcript). + +What was built: +- New `simple_coding_agent/telemetry.py`: `TelemetryReporter` (startup identity, process-local sequence, `snapshot(control)` reader interface, 256 KiB cap via `apply_size_cap` trimming oldest events then subagents), `SnapshotFailedError`, timeout/size constants. +- `lifecycle.py`: `AgentLifecycle(..., telemetry=...)`, `telemetry_snapshot()`, `telemetry_control_snapshot()` (≤1 s control-lock wait, reuse of last built section with `control_observed_at`, `SnapshotFailedError` when none built yet), plus network-free control-section/recovery builders (no workspace/tracker/publisher I/O; publication fields `None`). +- `control_server.py`: `{"op": "telemetry"}` → `{"ok": true, "telemetry": ...}`; `Unknown operation` for readers without telemetry; `snapshot_failed` + rate-limited `telemetry_failed` event (60 s); new `UnreadableReplyError`; `send_request(..., timeout=10.0)` default unchanged. +- `agentctl.py`: `telemetry --json` (required flag) printing the bare snapshot, exit 0; failures print `{"ok": false, "error": {"code", "message"}}` to stdout with `agent_unreachable`→2, `unsupported_operation`/`snapshot_failed`→1; 3 s timeout; no tracebacks. +- `__main__.py`: wires `TelemetryReporter` (provenance sdk/cli/agent versions, `process_started_at`) into `AgentLifecycle`. +- `tests/test_telemetry.py`: socket/shape/sequence, lock-busy cache reuse and empty-cache failure, exploding-fake no-I/O tests (idle + retained hold with null publication), all CLI error codes/exits, 3 s black-hole timeout, size-cap unit tests, rate-limit test. + +## Remaining work + +1. Decide whether to address code-review findings before finishing (most actionable: return a copy of the cached control section in `telemetry_control_snapshot`; move the final `print(json.dumps(snapshot))` inside the telemetry try in `agentctl.py`; consider dropping the `format_timestamp`/`identity` extra public surface). None was applied — implementation work is disabled at handoff. +2. Check whether the `PytestUnhandledThreadExceptionWarning` in the full-suite output is caused by the new tests (unverified; could be pre-existing). +3. Run the completion gate commands in order (ruff + pytest on both venvs). +4. Commit the work to branch `agent/issue-240` (nothing committed yet; sealing will commit uncommitted work). + +## Evidence + +- Modified: `simple_coding_agent/__main__.py`, `simple_coding_agent/agentctl.py`, `simple_coding_agent/control_server.py`, `simple_coding_agent/lifecycle.py` (460 insertions, 7 deletions per diff stat). +- New: `simple_coding_agent/telemetry.py`, `tests/test_telemetry.py`. +- Test results observed: `tests/test_telemetry.py` 22 passed; full suite 1070 passed, 10 skipped; `ruff check` clean. +- Code-review report: delivered in the run transcript under `## Standards` / `## Spec` headings. diff --git a/simple_coding_agent/__main__.py b/simple_coding_agent/__main__.py index 90d473d..65e9c66 100644 --- a/simple_coding_agent/__main__.py +++ b/simple_coding_agent/__main__.py @@ -3,6 +3,7 @@ from __future__ import annotations import sys +from datetime import UTC, datetime from pathlib import Path from simple_coding_agent.attempt_state import AttemptStateStore @@ -24,6 +25,7 @@ from simple_coding_agent.operating import ConsecutiveErrorStore from simple_coding_agent.publication import Publisher from simple_coding_agent.provenance import ProvenanceError, ProvenanceVerifier +from simple_coding_agent.telemetry import TelemetryReporter, format_timestamp from dotenv import load_dotenv @@ -31,6 +33,7 @@ def main() -> None: """Run sequential attempts forever for the one operator-configured repository.""" load_dotenv(override=False) + process_started_at = format_timestamp(datetime.now(UTC)) config = load_runtime_config() attempt_state = AttemptStateStore(config.data_dir) archive_factory = lambda issue_number, started_at: AttemptArchive( @@ -97,6 +100,13 @@ def _attempt_sink(issue_number: int) -> AttemptArchive | None: event, phase="setup", detail=detail, level=level, issue_number=issue_number ), ) + telemetry = TelemetryReporter( + repository=config.target_repo, + agent_version=provenance.agent_version, + sdk_version=provenance.sdk_version, + cli_version=provenance.cli_version, + process_started_at=process_started_at, + ) lifecycle = AgentLifecycle( tracker=tracker, attempt_state=attempt_state, @@ -129,6 +139,7 @@ def _attempt_sink(issue_number: int) -> AttemptArchive | None: completion_store=AttemptCompletionStore(config.data_dir), control_store=ControlStore(config.data_dir), repository=config.target_repo, + telemetry=telemetry, ) interruption_handler = AttemptInterruptionHandler( event_log=lambda event, detail="", level="INFO", issue_number=None: logger.emit( diff --git a/simple_coding_agent/agentctl.py b/simple_coding_agent/agentctl.py index f4c7878..b8f888a 100644 --- a/simple_coding_agent/agentctl.py +++ b/simple_coding_agent/agentctl.py @@ -12,6 +12,7 @@ from __future__ import annotations import argparse +import json import sys import uuid from collections.abc import Callable @@ -24,6 +25,10 @@ send_request, ) from simple_coding_agent.control import StopAfterRejectedError, parse_stop_after +from simple_coding_agent.telemetry import ( + SNAPSHOT_SIZE_LIMIT_BYTES, + TELEMETRY_TIMEOUT_SECONDS, +) def default_socket_path() -> Path: @@ -119,6 +124,75 @@ def run_status(socket_path: Path) -> dict: return reply["status"] +class TelemetryError(Exception): + """A machine-readable ``agentctl telemetry`` failure with its error code.""" + + def __init__(self, code: str, message: str) -> None: + super().__init__(message) + self.code = code + self.message = message + + +def run_telemetry(socket_path: Path) -> dict: + """Read one telemetry snapshot; raise a coded error when it cannot be served. + + Uses the short telemetry timeout so the call always answers inside the + dashboard's per-instance budget. ``agent_unreachable`` covers a missing, + refused, or timed-out socket; ``unsupported_operation`` covers an agent + that answers ``Unknown operation``; ``snapshot_failed`` covers an agent + that could not build the snapshot and replies that are unparsable, + oversized, or malformed. + """ + + from simple_coding_agent.control_server import UnreadableReplyError + + try: + reply = send_request( + socket_path, {"op": "telemetry"}, timeout=TELEMETRY_TIMEOUT_SECONDS + ) + except UnreadableReplyError as error: + raise TelemetryError( + "snapshot_failed", + f"agent answered but the reply was unparsable or oversized: {error}", + ) from error + except ControlUnavailableError as error: + raise TelemetryError( + "agent_unreachable", + f"cannot connect to the agent process: {error}", + ) from error + if not reply.get("ok"): + message = str(reply.get("error", "unknown error")) + if "Unknown operation" in message: + raise TelemetryError( + "unsupported_operation", + 'agent answered "Unknown operation":' + " this agent version does not serve telemetry", + ) + raise TelemetryError( + "snapshot_failed", + "agent answered but the telemetry snapshot could not be built:" + f" {message}", + ) + snapshot = reply.get("telemetry") + if not isinstance(snapshot, dict): + raise TelemetryError( + "snapshot_failed", + "agent answered but the telemetry snapshot was missing or malformed", + ) + if len(json.dumps(snapshot).encode("utf-8")) > SNAPSHOT_SIZE_LIMIT_BYTES: + raise TelemetryError( + "snapshot_failed", + "agent answered but the telemetry snapshot exceeded the 256 KiB size bound", + ) + return snapshot + + +def format_telemetry_error(error: TelemetryError) -> str: + """Render a telemetry failure as the machine-readable stdout envelope.""" + + return json.dumps({"ok": False, "error": {"code": error.code, "message": error.message}}) + + def run_command_lookup(socket_path: Path, request_id: str) -> dict | None: """Look up one command's current durable acknowledgement. @@ -382,6 +456,16 @@ def build_parser() -> argparse.ArgumentParser: ) subparsers.add_parser("status", help="Show one consistent live status snapshot.") + telemetry_parser = subparsers.add_parser( + "telemetry", help="Print the agent telemetry snapshot as JSON." + ) + telemetry_parser.add_argument( + "--json", + dest="json", + action="store_true", + required=True, + help="Print the bare telemetry snapshot object (the only v1 format).", + ) command_parser = subparsers.add_parser( "command", help="Show one command's current durable acknowledgement." ) @@ -493,6 +577,27 @@ def main(argv: list[str] | None = None) -> int: if args.command == "status": print(format_status(run_status(socket_path))) return 0 + if args.command == "telemetry": + try: + snapshot = run_telemetry(socket_path) + print(json.dumps(snapshot, sort_keys=True)) + except TelemetryError as error: + print(format_telemetry_error(error)) + return 2 if error.code == "agent_unreachable" else 1 + except Exception as error: + # No traceback ever escapes: an unexpected failure is still a + # machine-readable snapshot failure on stdout. + print( + format_telemetry_error( + TelemetryError( + "snapshot_failed", + "agent telemetry could not be read:" + f" {type(error).__name__}: {error}", + ) + ) + ) + return 1 + return 0 if args.command == "command": record = run_command_lookup(socket_path, args.request_id) if record is None: diff --git a/simple_coding_agent/control_server.py b/simple_coding_agent/control_server.py index ee53dd9..5953b41 100644 --- a/simple_coding_agent/control_server.py +++ b/simple_coding_agent/control_server.py @@ -15,6 +15,7 @@ import socket import stat import threading +import time from pathlib import Path from typing import Callable, Protocol @@ -47,11 +48,24 @@ def get_command(self, request_id: str) -> object | None: ... def control_status(self, repository: str | None = ...) -> dict: ... + def telemetry_snapshot(self) -> dict: + """Build one telemetry snapshot for the ``telemetry`` operation.""" + ... + class ControlUnavailableError(ConnectionError): """Raised when the agent process cannot be reached.""" +class UnreadableReplyError(ControlUnavailableError): + """Raised when the agent answered but the reply cannot be used. + + A subclass of :class:`ControlUnavailableError` so existing callers keep + reporting it as a connection problem; ``agentctl telemetry`` distinguishes + it to report ``snapshot_failed`` instead of ``agent_unreachable``. + """ + + _SOCKET_BACKLOG = 16 _BUFFER_LIMIT = 1 << 20 @@ -84,6 +98,7 @@ def __init__( self._listener: socket.socket | None = None self._thread: threading.Thread | None = None self._stopped = threading.Event() + self._telemetry_failed_at: float | None = None @property def socket_path(self) -> Path: @@ -253,6 +268,8 @@ def _handle(self, connection: socket.socket) -> None: self._reply_status(connection) elif operation == "command": self._reply_command(connection, request) + elif operation == "telemetry": + self._reply_telemetry(connection) else: _send( connection, @@ -398,6 +415,92 @@ def _reply_status(self, connection: socket.socket) -> None: return _send(connection, {"ok": True, "status": snapshot}) + def _reply_telemetry(self, connection: socket.socket) -> None: + """Answer one telemetry request without ever raising out of the listener. + + Controls without a telemetry reader (older agent versions) answer + ``Unknown operation`` so ``agentctl`` can report + ``unsupported_operation``. Any other failure answers + ``snapshot_failed`` with a rate-limited ``telemetry_failed`` event. + """ + + from simple_coding_agent.telemetry import TELEMETRY_FAILURE_COOLDOWN_SECONDS + + snapshotter = getattr(self._control, "telemetry_snapshot", None) + if not callable(snapshotter): + _send( + connection, + { + "ok": False, + "repository": self._safe_repository(), + "error": "Unknown operation: 'telemetry'.", + }, + ) + return + try: + snapshot = snapshotter() + except Exception as error: + self._log_telemetry_failed( + f"{type(error).__name__}: {error}", + cooldown=TELEMETRY_FAILURE_COOLDOWN_SECONDS, + ) + _send( + connection, + { + "ok": False, + "repository": self._safe_repository(), + "error": f"Telemetry snapshot failed: {error}", + }, + ) + return + if not isinstance(snapshot, dict): + self._log_telemetry_failed( + "telemetry reader returned a non-object snapshot", + cooldown=TELEMETRY_FAILURE_COOLDOWN_SECONDS, + ) + _send( + connection, + { + "ok": False, + "repository": self._safe_repository(), + "error": "Telemetry snapshot failed: the snapshot was not a JSON object.", + }, + ) + return + try: + json.dumps(snapshot) + except (TypeError, ValueError) as error: + self._log_telemetry_failed( + f"telemetry snapshot is not JSON-serialisable: {error}", + cooldown=TELEMETRY_FAILURE_COOLDOWN_SECONDS, + ) + _send( + connection, + { + "ok": False, + "repository": self._safe_repository(), + "error": f"Telemetry snapshot failed: {error}", + }, + ) + return + _send(connection, {"ok": True, "telemetry": snapshot}) + + def _log_telemetry_failed(self, detail: str, *, cooldown: float) -> None: + """Record a failed telemetry build, at most once per cooldown window.""" + + now = time.monotonic() + previous = self._telemetry_failed_at + if previous is not None and now - previous < cooldown: + return + self._telemetry_failed_at = now + event_log = self._event_log + if event_log is None: + return + try: + event_log("telemetry_failed", detail=detail, level="ERROR") + except Exception: + pass + def _reply_command(self, connection: socket.socket, request: dict) -> None: from simple_coding_agent.control import ControlStoreError, command_to_json @@ -523,8 +626,13 @@ def _reply_mutating( _send(connection, {"ok": True, "repository": repository, **command_to_json(record)}) -def send_request(socket_path: Path, request: dict) -> dict: - """Send one request to the live agent; raise when it cannot be reached.""" +def send_request(socket_path: Path, request: dict, *, timeout: float = 10.0) -> dict: + """Send one request to the live agent; raise when it cannot be reached. + + ``timeout`` covers the connect and the read. Unreadable or oversized + replies raise :class:`UnreadableReplyError`; every other failure raises + :class:`ControlUnavailableError`. + """ try: client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) @@ -533,7 +641,7 @@ def send_request(socket_path: Path, request: dict) -> dict: f"Cannot connect to the agent process at {socket_path}: {error}" ) from error try: - client.settimeout(10) + client.settimeout(timeout) try: client.connect(str(socket_path)) except OSError as error: @@ -550,8 +658,8 @@ def send_request(socket_path: Path, request: dict) -> dict: f"Lost connection to the agent process at {socket_path}: {error}" ) from error except ValueError as error: - raise ControlUnavailableError( - f"Agent process at {socket_path} returned an unreadable reply." + raise UnreadableReplyError( + f"Agent process at {socket_path} returned an oversized reply: {error}." ) from error finally: try: @@ -561,11 +669,11 @@ def send_request(socket_path: Path, request: dict) -> dict: try: reply = json.loads(raw) except ValueError as error: - raise ControlUnavailableError( + raise UnreadableReplyError( f"Agent process at {socket_path} returned an unreadable reply." ) from error if not isinstance(reply, dict): - raise ControlUnavailableError( + raise UnreadableReplyError( f"Agent process at {socket_path} returned an unreadable reply." ) return reply diff --git a/simple_coding_agent/lifecycle.py b/simple_coding_agent/lifecycle.py index 043463a..a1ae77a 100644 --- a/simple_coding_agent/lifecycle.py +++ b/simple_coding_agent/lifecycle.py @@ -4,6 +4,7 @@ import asyncio from collections.abc import Callable, Collection, Mapping, Sequence +import copy from dataclasses import dataclass, replace from datetime import UTC, datetime from enum import StrEnum @@ -73,6 +74,12 @@ from simple_coding_agent.operating import ConsecutiveErrorStore from simple_coding_agent.publication import AUTO_MERGE_LABEL, PublicationRequest from simple_coding_agent.recovery import REQUEUE_SOURCE_AUTOMATIC, requeue_source_for_labels +from simple_coding_agent.telemetry import ( + CONTROL_LOCK_TIMEOUT_SECONDS, + SnapshotFailedError, + TelemetryReporter, + format_timestamp, +) class AttemptInterruptionHandler: @@ -1152,6 +1159,7 @@ def __init__( control_store: ControlStore | None = None, repository: str = "", clock: Callable[[], datetime] | None = None, + telemetry: TelemetryReporter | None = None, ) -> None: self._tracker = tracker self._attempt_state = attempt_state @@ -1207,6 +1215,11 @@ def __init__( self._startup_hold_detail: str | None = None self._startup_hold_kind: str | None = None self._startup_hold_reported = False + self._telemetry = telemetry + # The last control section built for telemetry, with its + # ``control_observed_at``. Served again while the control lock is + # busy; ``None`` until the first successful build. + self._telemetry_control: dict | None = None @property def active_issue_number(self) -> int | None: @@ -2792,6 +2805,241 @@ def control_status(self, repository: str | None = None) -> dict: consecutive_errors=self._consecutive_errors_snapshot_locked(), ) + def telemetry_snapshot(self) -> dict: + """Serve one telemetry snapshot for the ``telemetry`` socket operation. + + Read-only: assembly goes through the telemetry reporter with this + lifecycle as the control reader. Raises :class:`SnapshotFailedError` + when telemetry is not configured or no control section can be built. + """ + + reporter = self._telemetry + if reporter is None: + raise SnapshotFailedError("telemetry is not configured for this process") + return reporter.snapshot(self) + + def telemetry_control_snapshot(self) -> dict: + """Return the network-free control section for one telemetry snapshot. + + The same payload :meth:`control_status` builds, but without workspace + or GitHub I/O: no dirty-tree check runs, no tracker or publisher call + is made, and publication fields that would need a lookup stay ``None`` + ("not known yet"). The section carries ``control_observed_at`` saying + when it was built. + + Waits at most ``CONTROL_LOCK_TIMEOUT_SECONDS`` for the control lock so + telemetry never holds up other control commands. While the lock is + busy the last built section is reused; when none has been built yet + this raises :class:`SnapshotFailedError`. + """ + + if self._control_store is None: + raise SnapshotFailedError("operator control is not configured") + if not self._control_lock.acquire(timeout=CONTROL_LOCK_TIMEOUT_SECONDS): + cached = self._telemetry_control + if cached is None: + raise SnapshotFailedError( + "the control lock is busy and no control section has been built yet" + ) + # A copy: the served snapshot must never alias the cache, or a + # consumer mutating it would corrupt later reuses. + return copy.deepcopy(cached) + try: + section = self._build_telemetry_control_section() + finally: + self._control_lock.release() + self._telemetry_control = section + return copy.deepcopy(section) + + def _build_telemetry_control_section(self) -> dict: + """Build a fresh network-free control section (caller holds the lock).""" + + store = self._control_store + assert store is not None # checked by telemetry_control_snapshot + try: + checkpoint = self._attempt_state.read() + except AttemptStateError as error: + # Read-only, as in control_status: record the startup hold + # silently (no event) so the operator still sees the hold reason. + if self._startup_hold_detail is None: + self._startup_hold_kind = "checkpoint" + self._startup_hold_detail = _checkpoint_hold_detail(error) + checkpoint = None + active = ( + ActiveAttemptInfo( + issue_number=checkpoint.issue_number, + branch=checkpoint.branch, + phase=checkpoint.phase.value, + started_at=checkpoint.started_at, + ) + if checkpoint is not None + else None + ) + section = build_status( + repository=self._repository, + intake=store.intake_state(), + recovering=self._recovering, + active_attempt=active, + pending_command_id=store.pending_command_id(), + get_command=store.get_command, + recent_commands=store.recent_commands, + next_issue=store.next_issue_snapshot(), + stop_plan=store.stop_plan_snapshot(), + recovery=self._recovery_snapshot_network_free(checkpoint), + handoff=store.handoff_snapshot(), + consecutive_errors=self._consecutive_errors_snapshot_locked(), + ) + section["control_observed_at"] = format_timestamp(self._clock()) + return section + + def _recovery_hold_network_free( + self, checkpoint: AttemptCheckpoint | None + ) -> RecoveryHold | None: + """Return the hold blocking ``resume``, without workspace I/O. + + Mirrors :meth:`_recovery_hold_locked` minus the dirty-tree check: the + tree is never inspected, so a dirty tree alone never appears here and + no workspace call is made. Never touches the tracker or publisher. + """ + + if self._recovering: + if checkpoint is not None: + return RecoveryHold( + issue_number=checkpoint.issue_number, + attempt_id=checkpoint.started_at, + branch=checkpoint.branch, + phase=checkpoint.phase.value, + reason=( + "startup reconciliation is still running for" + f" issue #{checkpoint.issue_number}; intake is held" + " until its outcome, publication, release, cleanup," + " and accounting are durable" + ), + ) + return RecoveryHold( + issue_number=None, + attempt_id=None, + branch=None, + phase=None, + reason="startup reconciliation is still running", + ) + if checkpoint is not None and not self._attempt_processing: + reason = self._startup_hold_detail or ( + "attempt finalization is unresolved; the branch," + " checkpoint, and working tree are preserved for recovery" + ) + return RecoveryHold( + issue_number=checkpoint.issue_number, + attempt_id=checkpoint.started_at, + branch=checkpoint.branch, + phase=checkpoint.phase.value, + reason=reason, + ) + if checkpoint is not None and self._recovery_executing: + return RecoveryHold( + issue_number=checkpoint.issue_number, + attempt_id=checkpoint.started_at, + branch=checkpoint.branch, + phase=checkpoint.phase.value, + reason=( + "a recovery command is running for the retained attempt;" + " the branch, checkpoint, and working tree are preserved" + " until it finishes" + ), + ) + return None + + def _recovery_snapshot_network_free( + self, checkpoint: AttemptCheckpoint | None + ) -> dict | None: + """Build the status ``recovery`` section without network or git I/O. + + Mirrors :meth:`_recovery_snapshot_locked`: the workspace dirty check + is skipped (a dirty tree alone never produces a section here) and the + publication progress stays ``None`` ("not known yet") because the + GitHub lookups never run. Never touches the tracker, publisher, or + workspace methods. Caller holds the control lock. + """ + + from simple_coding_agent.recovery import ( + describe_hold, + release_next_action, + retry_next_action, + ) + + if checkpoint is None: + if self._startup_hold_detail is None: + return None + return { + "attempt_id": None, + "issue_number": None, + "phase": None, + "outcome": None, + "branch": None, + "workspace": str(self._workspace.working_directory), + "checkpoint": False, + "publication": { + "phase": None, + "comment_confirmed": None, + "pull_request": None, + }, + "hold_reason": self._startup_hold_detail, + "next_action": ( + "Repair or remove the unreadable state file described" + " in the hold reason; intake resumes automatically" + " once it reads cleanly." + ), + } + hold = self._recovery_hold_network_free(checkpoint) + if hold is None and not self._recovering: + return None + reason = hold.reason if hold is not None else "startup reconciliation is still running" + outcome: str | None = None + if self._completion_store is not None: + try: + completions = self._completion_store.read_all() + terminal = completions.get(checkpoint.started_at) + if terminal is not None: + outcome = terminal.outcome.value + except Exception: + outcome = None + publication = { + "phase": checkpoint.phase.value, + "comment_confirmed": None, + "pull_request": None, + } + if self._recovering: + next_action = ( + "Wait for startup reconciliation to finish; do not run recovery" + " commands while it is still running." + ) + elif outcome is not None: + next_action = ( + "The attempt is finalized in the ledger; wait for the leftover" + " checkpoint to be removed on the next intake cycle." + ) + else: + next_action = retry_next_action(checkpoint.started_at) + return { + "attempt_id": checkpoint.started_at, + "issue_number": checkpoint.issue_number, + "phase": checkpoint.phase.value, + "outcome": outcome, + "branch": checkpoint.branch, + "workspace": str(self._workspace.working_directory), + "checkpoint": True, + "publication": publication, + "hold_reason": describe_hold( + issue_number=checkpoint.issue_number, + attempt_id=checkpoint.started_at, + phase=checkpoint.phase.value, + reason=reason, + ), + "next_action": next_action + if self._recovering or outcome is not None + else f"{next_action} {release_next_action()}", + } + def run_once(self) -> LifecycleResult: """Claim and process one issue, or sleep once when the queue is empty. diff --git a/simple_coding_agent/telemetry.py b/simple_coding_agent/telemetry.py new file mode 100644 index 0000000..d985e10 --- /dev/null +++ b/simple_coding_agent/telemetry.py @@ -0,0 +1,169 @@ +"""Versioned, read-only telemetry snapshots served over the control socket. + +A managed instance reports itself with a schema-v1 telemetry snapshot: its +control state, the current (or last finished) implementation attempt's +figures, and its most recent events. The dashboard only reads the snapshot; +this module owns its assembly. The contract (shape, versioning rules, size +bound, and ``agentctl telemetry --json`` mechanics) is published in +``docs/agents/telemetry-snapshot.md``; the transport decision (control +socket, not HTTP) is ADR 0006. + +Tracer scope: identity and control are real. ``current_attempt`` and +``last_finished`` are ``None`` and ``recent_events`` is empty until later +tickets fill them with the writer side (``begin_attempt``, +``publish_attempt``, ``finish_attempt``). +""" + +from __future__ import annotations + +from collections.abc import Callable +from datetime import UTC, datetime +import json +import threading +from typing import Protocol + + +#: Telemetry snapshot schema major version served by this process. +SCHEMA_VERSION = 1 + +#: Serialised snapshot cap: a quarter of the control socket's 1 MiB reply +#: limit. Over the cap the snapshot is trimmed (oldest events first, then +#: subagents) and served with ``truncated: true``. +SNAPSHOT_SIZE_LIMIT_BYTES = 256 * 1024 + +#: How long snapshot assembly waits for the control lock before reusing the +#: last built control section. Telemetry never holds up other control +#: commands: it only ever waits, never holds, the lock. +CONTROL_LOCK_TIMEOUT_SECONDS = 1.0 + +#: Connect+read timeout ``agentctl telemetry`` uses so it always answers +#: inside the dashboard's per-instance budget. Other commands keep theirs. +TELEMETRY_TIMEOUT_SECONDS = 3.0 + +#: Minimum gap between two ``telemetry_failed`` events: a persistently +#: failing snapshotter must not flood the log. +TELEMETRY_FAILURE_COOLDOWN_SECONDS = 60.0 + + +class SnapshotFailedError(RuntimeError): + """Raised when a telemetry snapshot cannot be built right now.""" + + +class TelemetryControl(Protocol): + """The reader interface snapshot assembly needs from the live agent.""" + + def telemetry_control_snapshot(self) -> dict: + """Return the network-free control section for one snapshot. + + The same payload ``status`` builds, but without workspace or GitHub + I/O: publication fields that would need a lookup stay ``None`` ("not + known yet"). Carries ``control_observed_at`` saying when the section + was built. Raises :class:`SnapshotFailedError` when no section has + been built yet. + """ + ... + + +class TelemetryReporter: + """Assemble schema-v1 snapshots behind the reader interface. + + Owns the process-local identity (recorded at startup), the process-local + sequence (reset with the process), and the size bound. The ``control`` + argument of :meth:`snapshot` supplies the live control section; later + tickets add the writer side that fills ``current_attempt``, + ``last_finished``, and ``recent_events``. + """ + + def __init__( + self, + *, + repository: str, + agent_version: str, + sdk_version: str, + cli_version: str, + process_started_at: str, + clock: Callable[[], datetime] | None = None, + ) -> None: + self._identity = { + "repository": repository, + "agent_version": agent_version, + "sdk_version": sdk_version, + "cli_version": cli_version, + "process_started_at": process_started_at, + } + self._clock = clock or (lambda: datetime.now(UTC)) + self._sequence = 0 + self._sequence_lock = threading.Lock() + + def snapshot(self, control: TelemetryControl) -> dict: + """Build one schema-v1 snapshot; raise when the control section is missing. + + ``SnapshotFailedError`` from the control reader propagates unchanged + so the socket handler can answer ``snapshot_failed``. + """ + + # The control section first: only a snapshot that is actually served + # consumes a sequence number. + section = control.telemetry_control_snapshot() + with self._sequence_lock: + self._sequence += 1 + sequence = self._sequence + snapshot = { + "schema_version": SCHEMA_VERSION, + "generated_at": _format_timestamp(self._clock()), + "sequence": sequence, + "identity": dict(self._identity), + "control": section, + "current_attempt": None, + "last_finished": None, + "recent_events": [], + "truncated": False, + } + return apply_size_cap(snapshot) + + +def apply_size_cap(snapshot: dict) -> dict: + """Trim an assembled snapshot to the serialised size bound, in place. + + Over the cap the snapshot drops ``recent_events`` oldest first, then + ``current_attempt.context.subagents`` oldest first, and reports + ``truncated: true``. Under the cap the snapshot is returned untouched. + """ + + if _serialised_size(snapshot) <= SNAPSHOT_SIZE_LIMIT_BYTES: + return snapshot + snapshot["truncated"] = True + events = snapshot.get("recent_events") + while ( + isinstance(events, list) + and events + and _serialised_size(snapshot) > SNAPSHOT_SIZE_LIMIT_BYTES + ): + events.pop(0) + attempt = snapshot.get("current_attempt") + while _serialised_size(snapshot) > SNAPSHOT_SIZE_LIMIT_BYTES: + if not isinstance(attempt, dict): + break + context = attempt.get("context") + if not isinstance(context, dict): + break + subagents = context.get("subagents") + if not isinstance(subagents, list) or not subagents: + break + subagents.pop(0) + return snapshot + + +def format_timestamp(value: datetime) -> str: + """Render a timestamp the way the snapshot records them (``Z``-suffixed).""" + + return _format_timestamp(value) + + +def _format_timestamp(value: datetime) -> str: + parsed = value.astimezone(UTC) if value.tzinfo is not None else value.replace(tzinfo=UTC) + return parsed.isoformat().replace("+00:00", "Z") + + +def _serialised_size(snapshot: dict) -> int: + return len(json.dumps(snapshot, sort_keys=True).encode("utf-8")) diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py new file mode 100644 index 0000000..ebaefb3 --- /dev/null +++ b/tests/test_telemetry.py @@ -0,0 +1,667 @@ +"""Tracer bullet for agent telemetry (schema v1) over the control socket. + +``agentctl telemetry --json`` returns a schema-v1 snapshot built without +network or git I/O and without holding up other control commands. Identity +and control are real; ``current_attempt`` and ``last_finished`` stay ``None`` +and ``recent_events`` stays empty until later tickets fill them. +""" + +from __future__ import annotations + +import inspect +import json +import socket +import threading +import time +from pathlib import Path + +import pytest + +from simple_coding_agent.agentctl import main as agentctl_main +from simple_coding_agent.attempt_state import AttemptStateStore +from simple_coding_agent.control import ControlStore +from simple_coding_agent.control_server import ControlServer, send_request +from simple_coding_agent.lifecycle import AgentLifecycle +from simple_coding_agent.telemetry import ( + SNAPSHOT_SIZE_LIMIT_BYTES, + SnapshotFailedError, + TelemetryReporter, + apply_size_cap, +) +from tests.fakes import FakePublisher, FakeTracker, InMemoryWorkspace + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + + +def make_reporter(**overrides: object) -> TelemetryReporter: + values: dict[str, object] = { + "repository": "owner/repo", + "agent_version": "0.0-test", + "sdk_version": "0.2.163-test", + "cli_version": "2.1.286-test", + "process_started_at": "2026-01-01T00:00:00Z", + } + values.update(overrides) + return TelemetryReporter(**values) # type: ignore[arg-type] + + +def make_lifecycle( + tmp_path: Path, *, reporter: TelemetryReporter | None = None +) -> tuple[AgentLifecycle, FakeTracker, ControlStore]: + tracker = FakeTracker() + workspace = InMemoryWorkspace(tmp_path / "repo") + store = ControlStore(tmp_path) + lifecycle = AgentLifecycle( + tracker=tracker, + attempt_state=AttemptStateStore(tmp_path), + workspace=workspace, + profile_loader=lambda _: type("Profile", (), {"base_branch": "main"})(), + publisher=FakePublisher(), + attempt_runner=lambda received_claim, profile, prepared: (_ for _ in ()).throw( + AssertionError("attempt runner must not run during telemetry") + ), + control_store=store, + repository="owner/repo", + telemetry=reporter if reporter is not None else make_reporter(), + sleeper=lambda seconds: None, + ) + return lifecycle, tracker, store + + +def start_telemetry_server( + tmp_path: Path, socket_name: str = "telemetry.sock" +) -> tuple[ControlServer, AgentLifecycle, ControlStore, Path]: + lifecycle, _, store = make_lifecycle(tmp_path) + socket_path = tmp_path / "state" / socket_name + server = ControlServer(socket_path, lifecycle) + server.start() + return server, lifecycle, store, socket_path + + +def send(socket_path: Path, payload: dict) -> dict: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client: + client.settimeout(10) + client.connect(str(socket_path)) + client.sendall((json.dumps(payload) + "\n").encode()) + data = b"" + while not data.endswith(b"\n"): + chunk = client.recv(65536) + if not chunk: + break + data += chunk + return json.loads(data.decode()) + + +class ScriptedSocketServer: + """Answer every connection with the same scripted bytes (or behaviour).""" + + def __init__(self, socket_path: Path, handler) -> None: + self._socket_path = socket_path + self._handler = handler + self._stopped = threading.Event() + self._listener: socket.socket | None = None + self._thread: threading.Thread | None = None + + def start(self) -> Path: + self._socket_path.parent.mkdir(parents=True, exist_ok=True) + try: + self._socket_path.unlink() + except OSError: + pass + listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + listener.bind(str(self._socket_path)) + listener.listen(8) + listener.settimeout(0.2) + self._listener = listener + self._thread = threading.Thread(target=self._serve, daemon=True) + self._thread.start() + return self._socket_path + + def _serve(self) -> None: + assert self._listener is not None + while not self._stopped.is_set(): + try: + connection, _ = self._listener.accept() + except socket.timeout: + continue + except OSError: + return + try: + with connection: + connection.settimeout(10) + self._handler(connection) + except OSError: + pass + + def stop(self) -> None: + self._stopped.set() + if self._listener is not None: + try: + self._listener.close() + except OSError: + pass + if self._thread is not None: + self._thread.join(timeout=10) + + +def read_request_then_reply(connection: socket.socket, payload: bytes) -> None: + data = b"" + while not data.endswith(b"\n"): + chunk = connection.recv(65536) + if not chunk: + break + data += chunk + connection.sendall(payload) + + +# --------------------------------------------------------------------------- +# socket operation and snapshot shape +# --------------------------------------------------------------------------- + + +def test_telemetry_socket_serves_a_schema_v1_snapshot(tmp_path: Path) -> None: + server, _, _, socket_path = start_telemetry_server(tmp_path) + try: + reply = send(socket_path, {"op": "telemetry"}) + finally: + server.stop() + + assert reply["ok"] is True + snapshot = reply["telemetry"] + assert snapshot["schema_version"] == 1 + assert isinstance(snapshot["generated_at"], str) and snapshot["generated_at"] + assert snapshot["sequence"] == 1 + assert snapshot["identity"] == { + "repository": "owner/repo", + "agent_version": "0.0-test", + "sdk_version": "0.2.163-test", + "cli_version": "2.1.286-test", + "process_started_at": "2026-01-01T00:00:00Z", + } + assert snapshot["current_attempt"] is None + assert snapshot["last_finished"] is None + assert snapshot["recent_events"] == [] + assert snapshot["truncated"] is False + + control = snapshot["control"] + assert control["repository"] == "owner/repo" + assert control["agent_version"] == snapshot["identity"]["agent_version"] or isinstance( + control["agent_version"], str + ) + assert isinstance(control["control_observed_at"], str) and control["control_observed_at"] + assert control["active_attempt"] is None + assert control["recovery"] is None + + +def test_telemetry_sequence_increments_per_snapshot(tmp_path: Path) -> None: + server, _, _, socket_path = start_telemetry_server(tmp_path) + try: + first = send(socket_path, {"op": "telemetry"}) + second = send(socket_path, {"op": "telemetry"}) + finally: + server.stop() + + assert first["telemetry"]["sequence"] == 1 + assert second["telemetry"]["sequence"] == 2 + + +def test_telemetry_failed_build_does_not_consume_sequence() -> None: + attempts = {"count": 0} + + class FlakyControl: + def telemetry_control_snapshot(self) -> dict: + attempts["count"] += 1 + if attempts["count"] == 1: + raise SnapshotFailedError("no control section has been built yet") + return {"control_observed_at": "2026-01-01T00:00:00Z"} + + reporter = make_reporter() + with pytest.raises(SnapshotFailedError): + reporter.snapshot(FlakyControl()) + snapshot = reporter.snapshot(FlakyControl()) + + assert snapshot["sequence"] == 1 + + +def test_telemetry_snapshot_matches_the_idle_example_shape(tmp_path: Path) -> None: + example = json.loads( + (Path(__file__).parent.parent / "examples" / "telemetry" + / "snapshot-idle-with-last-finished.json").read_text() + ) + server, _, _, socket_path = start_telemetry_server(tmp_path) + try: + reply = send(socket_path, {"op": "telemetry"}) + finally: + server.stop() + + snapshot = reply["telemetry"] + assert set(snapshot) == set(example) + assert set(snapshot["identity"]) == set(example["identity"]) + assert set(snapshot["control"]) == set(example["control"]) + # Tracer values diverge from the populated example on purpose: the writer + # side that fills these lands in later tickets. + assert snapshot["current_attempt"] is None + assert snapshot["last_finished"] is None + assert snapshot["recent_events"] == [] + assert snapshot["truncated"] is False + + +def test_telemetry_serves_cached_control_while_the_lock_is_busy(tmp_path: Path) -> None: + server, _, store, socket_path = start_telemetry_server(tmp_path) + try: + first = send(socket_path, {"op": "telemetry"}) + observed = first["telemetry"]["control"]["control_observed_at"] + + # Hold the control lock on this thread past the 1 s telemetry wait so + # the listener thread must reuse the last built control section. + store.lock.acquire() + results: list[dict] = [] + worker = threading.Thread( + target=lambda: results.append(send(socket_path, {"op": "telemetry"})) + ) + try: + worker.start() + time.sleep(1.5) + finally: + store.lock.release() + worker.join(timeout=30) + assert not worker.is_alive() + finally: + server.stop() + + assert len(results) == 1 + assert results[0]["ok"] is True + assert results[0]["telemetry"]["control"]["control_observed_at"] == observed + + +def test_telemetry_without_a_cached_section_fails_while_the_lock_is_busy( + tmp_path: Path, +) -> None: + server, _, store, socket_path = start_telemetry_server(tmp_path) + try: + store.lock.acquire() + results: list[dict] = [] + worker = threading.Thread( + target=lambda: results.append(send(socket_path, {"op": "telemetry"})) + ) + try: + worker.start() + time.sleep(1.5) + finally: + store.lock.release() + worker.join(timeout=30) + assert not worker.is_alive() + finally: + server.stop() + + assert len(results) == 1 + assert results[0]["ok"] is False + assert "snapshot failed" in results[0]["error"].lower() + + +# --------------------------------------------------------------------------- +# no tracker or git I/O on the telemetry path +# --------------------------------------------------------------------------- + + +class _ExplodingAttribute: + """Fail any method call; plain attribute reads still work.""" + + def __getattr__(self, name: str): + raise AssertionError(f"{name} must not be called on the telemetry path") + + +class ExplodingTracker(_ExplodingAttribute): + pass + + +class ExplodingWorkspace(_ExplodingAttribute): + def __init__(self, root: Path) -> None: + root.mkdir(parents=True, exist_ok=True) + self.working_directory = root + + +class ExplodingPublisher(_ExplodingAttribute): + pass + + +def make_exploding_lifecycle(tmp_path: Path) -> AgentLifecycle: + store = ControlStore(tmp_path) + return AgentLifecycle( + tracker=ExplodingTracker(), # type: ignore[arg-type] + attempt_state=AttemptStateStore(tmp_path), + workspace=ExplodingWorkspace(tmp_path / "repo"), # type: ignore[arg-type] + profile_loader=lambda _: (_ for _ in ()).throw( + AssertionError("profile loader must not run during telemetry") + ), + publisher=ExplodingPublisher(), # type: ignore[arg-type] + control_store=store, + repository="owner/repo", + telemetry=make_reporter(), + sleeper=lambda seconds: None, + ) + + +def test_telemetry_performs_no_tracker_or_workspace_calls_when_idle(tmp_path: Path) -> None: + lifecycle = make_exploding_lifecycle(tmp_path) + + snapshot = lifecycle.telemetry_snapshot() + + assert snapshot["schema_version"] == 1 + assert snapshot["control"]["recovery"] is None + assert snapshot["control"]["active_attempt"] is None + + +def test_telemetry_reports_a_retained_hold_without_lookups(tmp_path: Path) -> None: + lifecycle = make_exploding_lifecycle(tmp_path) + state = AttemptStateStore(tmp_path) + checkpoint = state.start(issue_number=24, branch="agent/issue-24") + + snapshot = lifecycle.telemetry_snapshot() + + control = snapshot["control"] + assert control["active_attempt"]["issue_number"] == 24 + recovery = control["recovery"] + assert recovery is not None + assert recovery["attempt_id"] == checkpoint.started_at + assert recovery["issue_number"] == 24 + assert recovery["checkpoint"] is True + # Publication fields that would need a GitHub lookup stay unknown. + assert recovery["publication"] == { + "phase": checkpoint.phase.value, + "comment_confirmed": None, + "pull_request": None, + } + assert "issue #24" in recovery["hold_reason"] + assert isinstance(recovery["next_action"], str) and recovery["next_action"] + + +# --------------------------------------------------------------------------- +# agentctl telemetry command +# --------------------------------------------------------------------------- + + +def test_agentctl_telemetry_prints_the_bare_snapshot( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + server, _, _, socket_path = start_telemetry_server(tmp_path) + try: + code = agentctl_main(["--socket", str(socket_path), "telemetry", "--json"]) + finally: + server.stop() + + assert code == 0 + captured = capsys.readouterr() + assert "Traceback" not in captured.out + captured.err + snapshot = json.loads(captured.out) + assert snapshot["schema_version"] == 1 + assert snapshot["control"]["repository"] == "owner/repo" + assert snapshot["current_attempt"] is None + + +def test_agentctl_telemetry_requires_json(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + with pytest.raises(SystemExit) as excinfo: + agentctl_main(["--socket", str(tmp_path / "missing.sock"), "telemetry"]) + + assert excinfo.value.code == 2 + assert capsys.readouterr().out == "" + + +def test_agentctl_telemetry_reports_agent_unreachable( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + code = agentctl_main( + ["--socket", str(tmp_path / "missing.sock"), "telemetry", "--json"] + ) + + assert code == 2 + captured = capsys.readouterr() + assert "Traceback" not in captured.out + captured.err + reply = json.loads(captured.out) + assert reply == { + "ok": False, + "error": {"code": "agent_unreachable", "message": reply["error"]["message"]}, + } + assert reply["error"]["message"] + + +def test_agentctl_telemetry_reports_unsupported_operation( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + class OldControl: + repository = "owner/repo" + + socket_path = tmp_path / "state" / "old.sock" + server = ControlServer(socket_path, OldControl()) # type: ignore[arg-type] + server.start() + try: + code = agentctl_main(["--socket", str(socket_path), "telemetry", "--json"]) + finally: + server.stop() + + assert code == 1 + captured = capsys.readouterr() + assert "Traceback" not in captured.out + captured.err + reply = json.loads(captured.out) + assert reply["ok"] is False + assert reply["error"]["code"] == "unsupported_operation" + assert reply["error"]["message"] + + +def test_agentctl_telemetry_reports_snapshot_failed( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + class FailingControl: + repository = "owner/repo" + + def __init__(self) -> None: + self.events: list[tuple] = [] + + def telemetry_snapshot(self) -> dict: + raise RuntimeError("boom-build") + + def _log(self, event: str, detail: str = "", level: str = "INFO") -> None: + self.events.append((event, detail, level)) + + control = FailingControl() + socket_path = tmp_path / "state" / "failing.sock" + server = ControlServer(socket_path, control, event_log=control._log) # noqa: SLF001 + server.start() + try: + code = agentctl_main(["--socket", str(socket_path), "telemetry", "--json"]) + first = capsys.readouterr() + first_events = list(control.events) + # A second immediate failure is rate-limited to one telemetry_failed event. + second = agentctl_main(["--socket", str(socket_path), "telemetry", "--json"]) + second_events = list(control.events) + assert server._thread is not None and server._thread.is_alive() # noqa: SLF001 + finally: + server.stop() + + assert code == 1 + assert second == 1 + assert "Traceback" not in first.out + first.err + reply = json.loads(first.out) + assert reply["ok"] is False + assert reply["error"]["code"] == "snapshot_failed" + assert [event for event, _, _ in first_events] == ["telemetry_failed"] + assert second_events == first_events + + +def test_telemetry_non_serialisable_snapshot_answers_snapshot_failed(tmp_path: Path) -> None: + class UnserialisableControl: + repository = "owner/repo" + + def __init__(self) -> None: + self.events: list[tuple] = [] + + def telemetry_snapshot(self) -> dict: + return {"not": "a snapshot", "unserialisable": object()} + + def _log(self, event: str, detail: str = "", level: str = "INFO") -> None: + self.events.append((event, detail, level)) + + control = UnserialisableControl() + socket_path = tmp_path / "state" / "unserialisable.sock" + server = ControlServer(socket_path, control, event_log=control._log) # noqa: SLF001 + server.start() + try: + reply = send(socket_path, {"op": "telemetry"}) + assert server._thread is not None and server._thread.is_alive() # noqa: SLF001 + finally: + server.stop() + + assert reply["ok"] is False + assert "snapshot failed" in reply["error"].lower() + assert [event for event, _, _ in control.events] == ["telemetry_failed"] + + +@pytest.mark.parametrize( + "payload", + [ + b"not json\n", + b'{"ok": true}\n', + b'{"ok": true, "telemetry": [1, 2]}\n', + b'{"ok": false, "repository": "owner/repo",' + b' "error": "Status is unavailable: boom"}\n', + ], + ids=["unparsable", "missing-snapshot", "malformed-snapshot", "agent-error"], +) +def test_agentctl_telemetry_maps_bad_replies_to_snapshot_failed( + tmp_path: Path, capsys: pytest.CaptureFixture[str], payload: bytes +) -> None: + socket_path = tmp_path / "state" / "scripted.sock" + server = ScriptedSocketServer( + socket_path, lambda connection: read_request_then_reply(connection, payload) + ) + server.start() + try: + code = agentctl_main(["--socket", str(socket_path), "telemetry", "--json"]) + finally: + server.stop() + + assert code == 1 + captured = capsys.readouterr() + assert "Traceback" not in captured.out + captured.err + reply = json.loads(captured.out) + assert reply["ok"] is False + assert reply["error"]["code"] == "snapshot_failed" + + +def test_agentctl_telemetry_maps_an_oversized_reply_to_snapshot_failed( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + payload = b'{"ok": true, "pad": "' + b"x" * (2**20) + b'"}\n' + socket_path = tmp_path / "state" / "oversized.sock" + server = ScriptedSocketServer( + socket_path, lambda connection: read_request_then_reply(connection, payload) + ) + server.start() + try: + code = agentctl_main(["--socket", str(socket_path), "telemetry", "--json"]) + finally: + server.stop() + + assert code == 1 + reply = json.loads(capsys.readouterr().out) + assert reply["error"]["code"] == "snapshot_failed" + + +def test_agentctl_telemetry_times_out_after_three_seconds( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + def black_hole(connection: socket.socket) -> None: + data = b"" + while not data.endswith(b"\n"): + chunk = connection.recv(65536) + if not chunk: + return + data += chunk + time.sleep(30) + + socket_path = tmp_path / "state" / "blackhole.sock" + server = ScriptedSocketServer(socket_path, black_hole) + server.start() + try: + started = time.monotonic() + code = agentctl_main(["--socket", str(socket_path), "telemetry", "--json"]) + elapsed = time.monotonic() - started + finally: + server.stop() + + assert code == 2 + reply = json.loads(capsys.readouterr().out) + assert reply["error"]["code"] == "agent_unreachable" + assert elapsed < 8 + + +def test_telemetry_uses_a_short_timeout_while_other_commands_keep_theirs() -> None: + import simple_coding_agent.telemetry as telemetry_module + + assert telemetry_module.TELEMETRY_TIMEOUT_SECONDS == 3.0 + assert ( + inspect.signature(send_request).parameters["timeout"].default == 10.0 + ) + + +# --------------------------------------------------------------------------- +# size bound +# --------------------------------------------------------------------------- + + +def test_size_cap_leaves_small_snapshots_untouched() -> None: + snapshot = { + "recent_events": [{"text": "small"}], + "current_attempt": None, + "truncated": False, + } + + assert apply_size_cap(snapshot) is snapshot + assert snapshot["truncated"] is False + assert snapshot["recent_events"] == [{"text": "small"}] + + +def test_size_cap_drops_oldest_events_first() -> None: + events = [ + {"timestamp": f"2026-01-01T00:00:{index:02d}Z", "text": f"event-{index:04d}-" + "x" * 1000} + for index in range(300) + ] + snapshot: dict = { + "recent_events": events, + "current_attempt": None, + "truncated": False, + } + + trimmed = apply_size_cap(snapshot) + + assert trimmed["truncated"] is True + assert len(json.dumps(trimmed).encode("utf-8")) <= SNAPSHOT_SIZE_LIMIT_BYTES + assert trimmed["recent_events"] + first_index = int(trimmed["recent_events"][0]["text"].split("-")[1]) + assert first_index > 0 + kept = [int(event["text"].split("-")[1]) for event in trimmed["recent_events"]] + assert kept == sorted(kept) + assert kept[-1] == 299 + + +def test_size_cap_drops_subagents_after_events_are_empty() -> None: + subagents = [ + {"id": f"agent-{index:03d}", "pad": "y" * 5000} for index in range(60) + ] + snapshot: dict = { + "recent_events": [], + "current_attempt": {"context": {"subagents": subagents}}, + "truncated": False, + } + + trimmed = apply_size_cap(snapshot) + + assert trimmed["truncated"] is True + assert len(json.dumps(trimmed).encode("utf-8")) <= SNAPSHOT_SIZE_LIMIT_BYTES + remaining = trimmed["current_attempt"]["context"]["subagents"] + assert remaining + assert remaining[0]["id"] != "agent-000" + assert remaining[-1]["id"] == "agent-059"