agent: Keep recent events in telemetry

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,

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:

simple_coding_agent/telemetry.py

@@ -28,18 +28,22 @@
 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
 import json
 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)

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