diff --git a/simple_coding_agent/__main__.py b/simple_coding_agent/__main__.py index 580e627..e69eab4 100644 --- a/simple_coding_agent/__main__.py +++ b/simple_coding_agent/__main__.py @@ -28,7 +28,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 simple_coding_agent.telemetry import RecentEventsBuffer, TelemetryReporter, format_timestamp from dotenv import load_dotenv @@ -52,10 +52,12 @@ def _attempt_sink(issue_number: int) -> AttemptArchive | None: return None return archive_factory(issue_number, checkpoint.started_at) + recent_events = RecentEventsBuffer() logger = JsonEventLogger( sys.stdout, redactions=config.credential_redactions, attempt_sink=_attempt_sink, + recent_events=recent_events, ) try: provenance = ProvenanceVerifier( @@ -117,6 +119,7 @@ def _attempt_sink(issue_number: int) -> AttemptArchive | None: # the served hit rate stays unavailable below this many measured # main-thread responses. cache_min_measured_responses=PROMPT_CACHE_MIN_MEASURED_RESPONSES, + recent_events=recent_events, ) lifecycle = AgentLifecycle( tracker=tracker, diff --git a/simple_coding_agent/observability.py b/simple_coding_agent/observability.py index 93d8c26..7f5f6a1 100644 --- a/simple_coding_agent/observability.py +++ b/simple_coding_agent/observability.py @@ -6,10 +6,13 @@ from datetime import UTC, datetime import json from pathlib import Path -from typing import TextIO +from typing import TYPE_CHECKING, TextIO from simple_coding_agent.agent_version import agent_version +if TYPE_CHECKING: + from simple_coding_agent.telemetry import RecentEventsBuffer + class Redactor: """Apply write-time credential redaction to all operational evidence.""" @@ -40,7 +43,9 @@ class JsonEventLogger: Each emitted record is also mirrored into the active attempt's ``agent_output.json`` (when ``attempt_sink`` resolves one), so the same output an operator sees on stdout is durably readable next to the - attempt's other evidence files. + attempt's other evidence files. Each redacted record is also fed to the + telemetry ``recent_events`` ring buffer (when one is attached), so every + buffered entry has passed through the redactor. """ def __init__( @@ -49,10 +54,12 @@ def __init__( *, redactions: Iterable[str] = (), attempt_sink: AttemptSink | None = None, + recent_events: RecentEventsBuffer | None = None, ) -> None: self._stream = stream self._redactor = Redactor(redactions) self._attempt_sink = attempt_sink + self._recent_events = recent_events def emit( self, @@ -76,6 +83,11 @@ def emit( json.dump(record, self._stream, sort_keys=True) self._stream.write("\n") self._stream.flush() + if self._recent_events is not None: + try: + self._recent_events.record(record) + except Exception: + pass if issue_number is not None and self._attempt_sink is not None: archive = self._attempt_sink(issue_number) if archive is not None: diff --git a/simple_coding_agent/telemetry.py b/simple_coding_agent/telemetry.py index deaa2c2..3495ccc 100644 --- a/simple_coding_agent/telemetry.py +++ b/simple_coding_agent/telemetry.py @@ -28,10 +28,14 @@ attempt archive record) so it survives restarts. Archives written before the pull-request and title fields existed seed ``pull_request: None`` and an ``"unknown"`` title. + +``recent_events`` is served from the process-local ring buffer fed by the +JSON event logger after redaction. """ from __future__ import annotations +from collections import deque from collections.abc import Callable from datetime import UTC, datetime import copy @@ -39,7 +43,7 @@ import logging from pathlib import Path import threading -from typing import Protocol +from typing import Any, Protocol from simple_coding_agent.observability import read_latest_completed_attempt @@ -100,6 +104,69 @@ logger = logging.getLogger(__name__) +#: How many redacted log events the process-local ``recent_events`` buffer +#: keeps. In-memory only, so it starts empty after a restart. +RECENT_EVENTS_LIMIT = 50 + +#: How many characters of an event's detail survive into ``recent_events``. +RECENT_EVENT_TEXT_LIMIT = 300 + +#: High-volume events excluded from ``recent_events``. +RECENT_EVENTS_EXCLUDED = frozenset({"limits_checked", "tool_call"}) + + +class RecentEventsBuffer: + """Process-local ring buffer of the last redacted log events. + + Fed by the JSON event logger after redaction, so every entry has passed + through the redactor. Keeps the last ``max_events`` entries, newest + last. Thread-safe: the logger emits from several threads while the + snapshot reader copies concurrently. + """ + + def __init__(self, max_events: int = RECENT_EVENTS_LIMIT) -> None: + self._max_events = max_events + self._entries: deque[dict[str, Any]] = deque(maxlen=max_events) + self._lock = threading.Lock() + + def record(self, redacted_record: dict[str, Any]) -> None: + """Append one redacted log record, newest last. + + ``redacted_record`` is the already-redacted mapping the JSON event + logger emits (``timestamp``, ``level``, ``event``, ``issue_number``, + ``phase``, ``detail``). High-volume excluded events are dropped. + Never raises: telemetry must not break logging. + """ + + try: + event = redacted_record.get("event") + if event in RECENT_EVENTS_EXCLUDED: + return + detail = redacted_record.get("detail", "") + text = detail if isinstance(detail, str) else str(detail) + entry = { + "timestamp": redacted_record.get("timestamp"), + "level": redacted_record.get("level"), + "event": event, + "issue_number": redacted_record.get("issue_number"), + "phase": redacted_record.get("phase"), + "text": text[:RECENT_EVENT_TEXT_LIMIT], + } + except Exception: + return + try: + with self._lock: + self._entries.append(entry) + except Exception: + return + + def snapshot(self) -> list[dict[str, Any]]: + """Return a copy of the buffered entries, newest last.""" + + with self._lock: + return [dict(entry) for entry in self._entries] + + class SnapshotFailedError(RuntimeError): """Raised when a telemetry snapshot cannot be built right now.""" @@ -635,12 +702,13 @@ 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), the size bound, the claim-time - ``current_attempt`` view (set by :meth:`begin_attempt`, cleared by - :meth:`finish_attempt`), and the ``last_finished`` view (set by - :meth:`finish_attempt` with a summary, seeded from disk at startup). - Configured limits passed at construction are defaults; seed values from - the claim override them per attempt. + sequence (reset with the process), the process-local ``recent_events`` + ring buffer, the size bound, the claim-time ``current_attempt`` view + (set by :meth:`begin_attempt`, cleared by :meth:`finish_attempt`), and + the ``last_finished`` view (set by :meth:`finish_attempt` with a + summary, seeded from disk at startup). Configured limits passed at + construction are defaults; seed values from the claim override them per + attempt. """ def __init__( @@ -661,6 +729,7 @@ def __init__( model_timeout: int | None = None, max_auto_continuations: int | None = None, cache_min_measured_responses: int = CACHE_MIN_MEASURED_RESPONSES, + recent_events: RecentEventsBuffer | None = None, ) -> None: self._identity = { "repository": repository, @@ -672,6 +741,9 @@ def __init__( self._clock = clock or (lambda: datetime.now(UTC)) self._sequence = 0 self._sequence_lock = threading.Lock() + self._recent_events = ( + recent_events if recent_events is not None else RecentEventsBuffer() + ) self._attempt_lock = threading.Lock() self._configured = { "hard_ceiling": hard_ceiling if hard_ceiling is not None else max_budget_tokens, @@ -827,7 +899,7 @@ def snapshot(self, control: TelemetryControl) -> dict: "control": section, "current_attempt": current, "last_finished": last_finished, - "recent_events": [], + "recent_events": self._recent_events.snapshot(), "truncated": False, } return apply_size_cap(snapshot) diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index da33ec0..bf394d0 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -1,14 +1,15 @@ """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. +network or git I/O and without holding up other control commands. Identity, +control, ``current_attempt``, ``last_finished`` and ``recent_events`` are +all served from process-local state. """ from __future__ import annotations import inspect +import io import json import socket import threading @@ -34,8 +35,10 @@ from simple_coding_agent.control_server import ControlServer, send_request from simple_coding_agent.github_tracker import TrackerIssue from simple_coding_agent.lifecycle import AgentLifecycle, AttemptEvidence +from simple_coding_agent.observability import JsonEventLogger from simple_coding_agent.telemetry import ( SNAPSHOT_SIZE_LIMIT_BYTES, + RecentEventsBuffer, SnapshotFailedError, TelemetryReporter, apply_size_cap, @@ -254,7 +257,8 @@ def test_telemetry_snapshot_matches_the_idle_example_shape(tmp_path: Path) -> No 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. + # side that fills the attempt sections lands in later tickets, and no + # events were emitted, so the process-local buffer is still empty. assert snapshot["current_attempt"] is None assert snapshot["last_finished"] is None assert snapshot["recent_events"] == [] @@ -1224,3 +1228,112 @@ def prepare_for_profile_read(self) -> None: assert current["continuation"]["index"] == {"value": 2, "basis": "configured"} assert current["continuation"]["max"] == {"value": 2, "basis": "configured"} assert snapshot["control"]["active_attempt"]["issue_title"] == "Fix the widget" + + +# --------------------------------------------------------------------------- +# recent_events ring buffer +# --------------------------------------------------------------------------- + + +def _stub_control(section: dict | None = None): + class StubControl: + def telemetry_control_snapshot(self) -> dict: + return dict( + section if section is not None else {"control_observed_at": "2026-01-01T00:00:00Z"} + ) + + return StubControl() + + +def _wired_buffer( + secret: str | None = None, +) -> tuple[io.StringIO, RecentEventsBuffer, JsonEventLogger, TelemetryReporter]: + """One shared buffer behind both a redacting logger and a reporter.""" + + stream = io.StringIO() + buffer = RecentEventsBuffer() + logger = JsonEventLogger( + stream, + redactions=(secret,) if secret is not None else (), + recent_events=buffer, + ) + return stream, buffer, logger, make_reporter(recent_events=buffer) + + +def test_recent_events_buffer_starts_empty() -> None: + _, _, _, reporter = _wired_buffer() + + snapshot = reporter.snapshot(_stub_control()) + + assert snapshot["recent_events"] == [] + + +def test_secret_in_event_detail_never_reaches_the_snapshot() -> None: + stream, _, logger, reporter = _wired_buffer(secret="super-secret-value") + + logger.emit( + "attempt_claimed", + issue_number=977, + phase="claim", + detail="using token super-secret-value to proceed", + ) + + snapshot = reporter.snapshot(_stub_control()) + assert len(snapshot["recent_events"]) == 1 + entry = snapshot["recent_events"][0] + assert set(entry) == { + "timestamp", + "level", + "event", + "issue_number", + "phase", + "text", + } + assert entry["event"] == "attempt_claimed" + assert entry["issue_number"] == 977 + assert entry["phase"] == "claim" + assert entry["level"] == "INFO" + assert isinstance(entry["timestamp"], str) and entry["timestamp"] + assert "super-secret-value" not in json.dumps(snapshot) + assert "super-secret-value" not in entry["text"] + assert "[REDACTED]" in entry["text"] + # The stdout line is redacted too: the buffer only ever saw redacted text. + assert "super-secret-value" not in stream.getvalue() + + +def test_excluded_events_are_absent_from_recent_events() -> None: + _, _, logger, reporter = _wired_buffer() + + logger.emit("limits_checked", detail="usage=1") + logger.emit("tool_call", detail="Bash ls") + logger.emit("attempt_claimed", issue_number=1, phase="claim", detail="kept") + + snapshot = reporter.snapshot(_stub_control()) + assert [entry["event"] for entry in snapshot["recent_events"]] == [ + "attempt_claimed" + ] + + +def test_recent_events_buffer_holds_at_most_fifty_newest_last() -> None: + _, _, logger, reporter = _wired_buffer() + + for index in range(60): + logger.emit("attempt_claimed", detail=f"event-{index:02d}") + + snapshot = reporter.snapshot(_stub_control()) + events = snapshot["recent_events"] + assert len(events) == 50 + assert events[0]["text"] == "event-10" + assert events[-1]["text"] == "event-59" + assert [entry["text"] for entry in events] == sorted( + entry["text"] for entry in events + ) + + +def test_recent_event_text_is_truncated_to_300_characters() -> None: + _, _, logger, reporter = _wired_buffer() + + logger.emit("attempt_claimed", detail="x" * 500) + + snapshot = reporter.snapshot(_stub_control()) + assert len(snapshot["recent_events"][0]["text"]) == 300