From 3811d80528c99b2d3baacf0976db79e03636c838 Mon Sep 17 00:00:00 2001 From: TensorTemplar Date: Tue, 18 Aug 2026 20:05:21 +0300 Subject: [PATCH] Open the source dimension + stable envelope ingest for third-party agents MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds on the #54 protocol: keeps its AbstractEventType taxonomy and adapter/dispatch architecture, and adds what #46 still asked for. - AbstractHookEvent.source is now an open string; AbstractEventSource becomes documented vocabulary ("has an adapter"), not a gate. The DB reader no longer casts sources through the enum, so third-party source names round-trip storage intact. - EventEnvelope (protocol/schema.py): the stable external contract — source (open), schema_version (fail-fast), required event_id, canonical kind, seq, occurred_at, tool fields, raw payload. - `slopometry ingest --source [--file] [--working-directory]`: JSONL envelopes from stdin/file, atomic validate-then-insert, loud exit-2 failures. Dedupe on (source, event_id) makes trace backfills idempotent; sequence numbers continue after the highest stored/batched seq when envelopes omit one. - Migration020: event_id column + unique partial index (source, event_id) WHERE event_id IS NOT NULL; also repairs event_type values written by the abandoned hook-protocol branch (1:1 remap back onto the canonical taxonomy). Numbered 020 because 017-019 were consumed by diverged trees. --- src/slopometry/cli.py | 56 ++++- src/slopometry/core/database.py | 37 +++- src/slopometry/core/migrations.py | 51 +++++ src/slopometry/core/models/protocol/events.py | 16 +- src/slopometry/core/protocol/ingest.py | 97 +++++++++ src/slopometry/core/protocol/schema.py | 48 +++++ .../solo/services/transcript_finder.py | 2 +- tests/test_migrations.py | 46 ++++ tests/test_protocol_events.py | 19 +- tests/test_protocol_ingest.py | 199 ++++++++++++++++++ tests/test_protocol_schema.py | 68 ++++++ 11 files changed, 621 insertions(+), 18 deletions(-) create mode 100644 src/slopometry/core/protocol/ingest.py create mode 100644 src/slopometry/core/protocol/schema.py create mode 100644 tests/test_protocol_ingest.py create mode 100644 tests/test_protocol_schema.py diff --git a/src/slopometry/cli.py b/src/slopometry/cli.py index 0c44361..5ac1365 100644 --- a/src/slopometry/cli.py +++ b/src/slopometry/cli.py @@ -4,6 +4,7 @@ import sys import warnings from importlib.metadata import version +from pathlib import Path # REASON: analyzed repos may contain invalid escape sequences that emit SyntaxWarnings during AST parsing warnings.filterwarnings("ignore", category=SyntaxWarning) @@ -118,7 +119,7 @@ def hook_opencode(event_type: str) -> None: "--source", required=True, type=click.Choice(["claude_code", "opencode"]), - help="Which harness produced the event on stdin.", + help="Which harness produced the event on stdin. Third-party collectors without an adapter should use 'slopometry ingest' instead.", ) @click.option( "--type", @@ -153,6 +154,59 @@ def emit_event(source: str, event_type: str | None) -> None: sys.exit(emit_event_from_stdin(abstract_source, event_type_override=abstract_type)) +@cli.command("ingest") +@click.option("--source", required=True, help="Agent tool that produced the events, e.g. 'mmkr'.") +@click.option( + "--file", + "input_file", + type=click.Path(exists=True, dir_okay=False, path_type=Path), + default=None, + help="JSONL file of event envelopes; reads from stdin when omitted.", +) +@click.option( + "--working-directory", + type=click.Path(exists=True, file_okay=False, path_type=Path), + default=None, + help="Working directory recorded on ingested events; defaults to the current directory.", +) +def ingest(source: str, input_file: Path | None, working_directory: Path | None) -> None: + """Ingest hook-protocol event envelopes (JSONL) from any agent tool. + + Each line must be an EventEnvelope JSON object; see + slopometry.core.protocol.schema for the stable schema. Re-ingesting the + same (source, event_id) pairs is an idempotent no-op. + """ + import json + + from pydantic import ValidationError + + from slopometry.core.protocol.ingest import ingest_envelopes + from slopometry.core.protocol.schema import EventEnvelope + + content = input_file.read_text() if input_file else sys.stdin.read() + envelopes: list[EventEnvelope] = [] + for line_number, line in enumerate(content.splitlines(), start=1): + if not line.strip(): + continue + try: + envelope = EventEnvelope.model_validate(json.loads(line)) + except (json.JSONDecodeError, ValidationError) as e: + console.print(f"[red]Invalid envelope at line {line_number}: {e}[/red]") + sys.exit(2) + if envelope.source != source: + console.print( + f"[red]Envelope source '{envelope.source}' at line {line_number} does not match --source '{source}'[/red]" + ) + sys.exit(2) + envelopes.append(envelope) + + report = ingest_envelopes(envelopes, working_directory=str(working_directory) if working_directory else None) + console.print( + f"[green]Ingested {report.inserted} events from source '{source}'" + f" (skipped {report.skipped_duplicates} duplicates)[/green]" + ) + + @cli.command("shell-completion") @click.argument("shell", type=click.Choice(["bash", "zsh", "fish"])) def shell_completion(shell: str) -> None: diff --git a/src/slopometry/core/database.py b/src/slopometry/core/database.py index 3f10d87..8ea60e7 100644 --- a/src/slopometry/core/database.py +++ b/src/slopometry/core/database.py @@ -133,7 +133,8 @@ def _create_tables(self) -> None: project_source TEXT, transcript_path TEXT, source TEXT DEFAULT 'claude_code', - parent_session_id TEXT + parent_session_id TEXT, + event_id TEXT ) """) conn.execute(""" @@ -157,6 +158,10 @@ def _create_tables(self) -> None: ON hook_events(session_id, event_type, source) """) + # NOTE: the (source, event_id) unique partial index is created by + # Migration020EnvelopeIdempotency only; _create_tables runs before + # migrations on pre-existing databases that lack the event_id column. + conn.execute(""" CREATE TABLE IF NOT EXISTS experiment_runs ( id TEXT PRIMARY KEY, @@ -398,6 +403,26 @@ def _create_tables(self) -> None: conn.commit() + def has_event(self, source: str, event_id: str) -> bool: + """Check whether an event with the given (source, event_id) is already stored. + + Used by envelope ingestion to make backfills idempotent.""" + with self._get_db_connection() as conn: + row = conn.execute( + "SELECT 1 FROM hook_events WHERE source = ? AND event_id = ? LIMIT 1", + (source, event_id), + ).fetchone() + return row is not None + + def get_max_sequence_number(self, session_id: str) -> int: + """Get the highest stored sequence number for a session (0 when the session has no events).""" + with self._get_db_connection() as conn: + row = conn.execute( + "SELECT COALESCE(MAX(sequence_number), 0) FROM hook_events WHERE session_id = ?", + (session_id,), + ).fetchone() + return row[0] + def save_event(self, event: AbstractHookEvent) -> int: """Save a hook event to the database.""" tool_name = event.tool_call.tool_name if event.tool_call else None @@ -414,8 +439,8 @@ def save_event(self, event: AbstractHookEvent) -> int: tool_name, tool_type, metadata, duration_ms, exit_code, error_message, git_state, working_directory, project_name, project_source, transcript_path, - source, parent_session_id - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + source, parent_session_id, event_id + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( event.session_id, @@ -433,8 +458,9 @@ def save_event(self, event: AbstractHookEvent) -> int: event.project.name if event.project else None, event.project.source.value if event.project else None, event.transcript_location, - event.source.value, + event.source, event.parent_session_id, + event.event_id, ), ) return cursor.lastrowid or 0 @@ -469,7 +495,7 @@ def get_session_events(self, session_id: str) -> list[AbstractHookEvent]: ) source_val = row["source"] if "source" in row.keys() else None - source = AbstractEventSource(source_val) if source_val else AbstractEventSource.CLAUDE_CODE + source = source_val or str(AbstractEventSource.CLAUDE_CODE) parent_session_id = row["parent_session_id"] if "parent_session_id" in row.keys() else None tool_name = row["tool_name"] @@ -477,6 +503,7 @@ def get_session_events(self, session_id: str) -> list[AbstractHookEvent]: events.append( AbstractHookEvent( id=row["id"], + event_id=row["event_id"], session_id=row["session_id"], event_type=AbstractEventType(row["event_type"]), timestamp=datetime.fromisoformat(row["timestamp"]), diff --git a/src/slopometry/core/migrations.py b/src/slopometry/core/migrations.py index 7dc4e23..93b9cb3 100644 --- a/src/slopometry/core/migrations.py +++ b/src/slopometry/core/migrations.py @@ -569,6 +569,56 @@ def up(self, conn: sqlite3.Connection) -> None: raise +class Migration020EnvelopeIdempotency(Migration): + """Enable idempotent envelope ingestion on hook_events. + + - Adds the event_id column (if missing) with a partial unique index on + (source, event_id) so re-ingesting a trace is a no-op. + - Repairs event_type values written by the abandoned 'hook-protocol' + branch, which used a divergent taxonomy (tool_call/tool_result/stop/ + subagent_stop/subagent_start). Every value maps 1:1 back onto this + taxonomy; databases that never ran that branch are unaffected. + + Numbered 020: 017-019 were consumed by diverged local trees. + """ + + _WRONG_DIALECT_RENAMES: dict[str, str] = { + "tool_call": "tool_call_started", + "tool_result": "tool_call_completed", + "stop": "turn_completed", + "subagent_stop": "subagent_completed", + "subagent_start": "subagent_started", + } + + @property + def version(self) -> str: + return "020" + + @property + def description(self) -> str: + return "Add event_id with unique partial index for envelope backfills; repair wrong-dialect event_type values" + + def up(self, conn: sqlite3.Connection) -> None: + cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='hook_events'") + if not cursor.fetchone(): + return + + columns = {row[1] for row in conn.execute("PRAGMA table_info(hook_events)").fetchall()} + + if "event_type" in columns: + for wrong, canonical in self._WRONG_DIALECT_RENAMES.items(): + conn.execute("UPDATE hook_events SET event_type = ? WHERE event_type = ?", (canonical, wrong)) + + if "event_id" not in columns: + conn.execute("ALTER TABLE hook_events ADD COLUMN event_id TEXT") + + conn.execute(""" + CREATE UNIQUE INDEX IF NOT EXISTS idx_hook_events_source_event_id + ON hook_events(source, event_id) + WHERE event_id IS NOT NULL + """) + + class MigrationRunner: """Manages database migrations.""" @@ -591,6 +641,7 @@ def __init__(self, db_path: Path): Migration014AddBehavioralPatternHistory(), Migration015AbstractEventTypeValues(), Migration016AddRetiredReasonToMemories(), + Migration020EnvelopeIdempotency(), ] @contextmanager diff --git a/src/slopometry/core/models/protocol/events.py b/src/slopometry/core/models/protocol/events.py index a621a32..3f38b4f 100644 --- a/src/slopometry/core/models/protocol/events.py +++ b/src/slopometry/core/models/protocol/events.py @@ -15,7 +15,13 @@ class AbstractEventSource(StrEnum): - """Identity of the agent harness or collector that produced the event.""" + """Vocabulary for the harnesses slopometry ships adapters for. + + This enum is NOT a gate: `AbstractHookEvent.source` is an open string so + third-party collectors can ingest events under their own source name + (see `slopometry ingest`). Membership here only means "has a wire-format + adapter registered in `protocol/adapters`". + """ CLAUDE_CODE = "claude_code" OPENCODE = "opencode" @@ -86,13 +92,19 @@ class AbstractHookEvent(BaseModel): default=None, description="Database autoincrement id; None for in-memory events not yet persisted", ) + event_id: str | None = Field( + default=None, + description="Source-unique id from envelope ingestion; (source, event_id) is unique so backfills are idempotent. Live harness paths have no source-side event id.", + ) session_id: str parent_session_id: str | None = Field( default=None, description="Parent session ID for subagent/child sessions; None for top-level", ) event_type: AbstractEventType - source: AbstractEventSource + source: str = Field( + description="Source name of the producing agent; built-ins in AbstractEventSource, open for third-party collectors", + ) timestamp: datetime = Field(default_factory=datetime.now) tool_call: ToolCallPayload | None = None metadata: dict[str, Any] = Field( diff --git a/src/slopometry/core/protocol/ingest.py b/src/slopometry/core/protocol/ingest.py new file mode 100644 index 0000000..d066492 --- /dev/null +++ b/src/slopometry/core/protocol/ingest.py @@ -0,0 +1,97 @@ +"""Idempotent ingestion of `EventEnvelope` batches into slopometry storage. + +This is the third-party counterpart to `protocol/dispatch.py`: dispatch +runs a wire payload through a registered adapter (live harness paths), while +ingest accepts already-canonical envelopes from collectors without adapters. +Session context (git state, project detection) is deliberately not captured: +the repository the ingester runs in is unrelated to the traced session. +""" + +import sqlite3 +from pathlib import Path + +from pydantic import BaseModel + +from slopometry.core.database import EventDatabase +from slopometry.core.models.protocol.events import AbstractHookEvent, ToolCallPayload +from slopometry.core.protocol.schema import EventEnvelope + + +class IngestReport(BaseModel): + """Outcome of ingesting a batch of envelopes.""" + + inserted: int = 0 + skipped_duplicates: int = 0 + + +def envelope_to_event(envelope: EventEnvelope, working_directory: str, sequence_number: int) -> AbstractHookEvent: + """Expand a validated envelope into the canonical stored event model.""" + tool_call = ( + ToolCallPayload( + tool_name=envelope.tool_name, + input=envelope.raw, + duration_ms=envelope.duration_ms, + exit_code=envelope.exit_code, + error_message=envelope.error_message, + ) + if envelope.tool_name + else None + ) + return AbstractHookEvent( + session_id=envelope.session_id, + parent_session_id=envelope.parent_session_id, + event_type=envelope.kind, + source=envelope.source, + timestamp=envelope.occurred_at, + tool_call=tool_call, + metadata=envelope.raw, + working_directory=working_directory, + sequence_number=sequence_number, + event_id=envelope.event_id, + ) + + +def ingest_envelopes( + envelopes: list[EventEnvelope], + db: EventDatabase | None = None, + working_directory: str | None = None, +) -> IngestReport: + """Store envelopes, skipping duplicates identified by (source, event_id). + + Sequence numbers come from the envelope when provided; otherwise a + per-session counter continuing after the highest sequence number already + used (stored or earlier in the batch) is assigned. The live SessionManager + state files are not touched: backfills derive ordering from storage. + """ + db = db or EventDatabase() + working_directory = working_directory or str(Path.cwd()) + report = IngestReport() + next_sequence: dict[str, int] = {} + + for envelope in envelopes: + if db.has_event(envelope.source, envelope.event_id): + report.skipped_duplicates += 1 + continue + + if envelope.seq is not None: + sequence_number = envelope.seq + next_sequence[envelope.session_id] = max( + envelope.seq + 1, + next_sequence.get(envelope.session_id, 1 + db.get_max_sequence_number(envelope.session_id)), + ) + else: + next_sequence.setdefault(envelope.session_id, 1 + db.get_max_sequence_number(envelope.session_id)) + sequence_number = next_sequence[envelope.session_id] + next_sequence[envelope.session_id] = sequence_number + 1 + + try: + event = envelope_to_event(envelope, working_directory, sequence_number) + event.id = db.save_event(event) + except sqlite3.IntegrityError: + # Concurrent ingest of the same (source, event_id); the unique + # partial index makes this safe to treat as a duplicate. + report.skipped_duplicates += 1 + continue + report.inserted += 1 + + return report diff --git a/src/slopometry/core/protocol/schema.py b/src/slopometry/core/protocol/schema.py new file mode 100644 index 0000000..c7db8ba --- /dev/null +++ b/src/slopometry/core/protocol/schema.py @@ -0,0 +1,48 @@ +"""Stable external schema for third-party event ingestion. + +`EventEnvelope` is the public contract for agent tools that have no wire- +format adapter (anything beyond Claude Code and OpenCode). Collectors emit +JSONL envelopes and feed them to `slopometry ingest`; the closed +`AbstractEventType` taxonomy and open source strings keep the contract +harness-independent (see issue #46). +""" + +from datetime import datetime +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +from slopometry.core.models.protocol.events import AbstractEventType + +PROTOCOL_SCHEMA_VERSION = 1 + + +class EventEnvelope(BaseModel): + """One event as ingested over the stable hook protocol. + + Collectors identify events via `source` + `event_id`; the combination is + unique in storage, so re-ingesting a trace file is an idempotent backfill. + """ + + model_config = ConfigDict(extra="allow") + + schema_version: int = Field(default=PROTOCOL_SCHEMA_VERSION, description="Protocol schema version, currently 1") + source: str = Field(min_length=1, description="Agent tool that produced the event, e.g. 'mmkr'") + session_id: str = Field(min_length=1, description="Session the event belongs to") + event_id: str = Field(min_length=1, description="Source-unique event id; required so backfills are always idempotent") + parent_session_id: str | None = Field(default=None, description="Parent session id for subagent sessions") + seq: int | None = Field(default=None, ge=1, description="Source-provided sequence number for the event") + occurred_at: datetime = Field(default_factory=datetime.now, description="When the event occurred at the source") + kind: AbstractEventType = Field(description="Canonical event kind; validated against the closed taxonomy") + tool_name: str | None = Field(default=None, description="Tool name for tool_call_started/tool_call_completed events") + duration_ms: int | None = Field(default=None, ge=0, description="Tool execution duration for tool_call_completed events") + exit_code: int | None = Field(default=None, description="Tool exit code for tool_call_completed events") + error_message: str | None = Field(default=None, description="Tool error message for tool_call_completed events") + raw: dict[str, Any] = Field(default_factory=dict, description="Source-specific payload kept as event metadata") + + @field_validator("schema_version") + @classmethod + def schema_version_must_be_supported(cls, value: int) -> int: + if value != PROTOCOL_SCHEMA_VERSION: + raise ValueError(f"Unsupported schema_version {value}; this slopometry version supports 1") + return value diff --git a/src/slopometry/solo/services/transcript_finder.py b/src/slopometry/solo/services/transcript_finder.py index 4f8f5e7..b7edc26 100644 --- a/src/slopometry/solo/services/transcript_finder.py +++ b/src/slopometry/solo/services/transcript_finder.py @@ -14,7 +14,7 @@ class DiscoveredTranscript: session_id: str transcript_path: Path project_dir: Path - source: AbstractEventSource + source: str class TranscriptFinder: diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 3cc8891..f93214a 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -1,8 +1,11 @@ """Tests for database migrations.""" +import sqlite3 from pathlib import Path from tempfile import TemporaryDirectory +import pytest + from slopometry.core.migrations import MigrationRunner # Derive expected count from the runner itself so adding a migration doesn't @@ -220,3 +223,46 @@ def test_migration_016__is_idempotent_when_column_already_exists(self): cursor = conn.execute("PRAGMA table_info(memories)") columns = [row[1] for row in cursor.fetchall()] assert columns.count("retired_reason") == 1 + + def test_migration_020__adds_event_id_idempotency_and_repairs_wrong_dialect(self): + """Envelope backfills get a unique (source, event_id) index; wrong-dialect values return to the canonical taxonomy.""" + with TemporaryDirectory() as temp_dir: + db_path = Path(temp_dir) / "test.db" + runner = MigrationRunner(db_path) + + with runner._get_db_connection() as conn: + conn.execute(""" + CREATE TABLE hook_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + event_type TEXT NOT NULL, + timestamp TEXT NOT NULL, + source TEXT + ) + """) + conn.execute( + "INSERT INTO hook_events (session_id, event_type, timestamp, source) VALUES ('s1', 'tool_call', '2026-01-01T00:00:00', 'claude_code')" + ) + conn.execute( + "INSERT INTO hook_events (session_id, event_type, timestamp, source) VALUES ('s1', 'turn_completed', '2026-01-01T00:01:00', 'opencode')" + ) + conn.commit() + + runner.run_migrations() + + with runner._get_db_connection() as conn: + rows = conn.execute("SELECT event_type FROM hook_events ORDER BY id").fetchall() + assert rows == [("tool_call_started",), ("turn_completed",)] + + columns = {row[1] for row in conn.execute("PRAGMA table_info(hook_events)").fetchall()} + indexes = {row[1] for row in conn.execute("PRAGMA index_list(hook_events)").fetchall()} + assert "event_id" in columns + assert "idx_hook_events_source_event_id" in indexes + + conn.execute( + "INSERT INTO hook_events (session_id, event_type, timestamp, source, event_id) VALUES ('s2', 'turn_completed', '2026-01-01T02:00:00', 'mmkr', 'evt-1')" + ) + with pytest.raises(sqlite3.IntegrityError): + conn.execute( + "INSERT INTO hook_events (session_id, event_type, timestamp, source, event_id) VALUES ('s2', 'turn_completed', '2026-01-01T02:01:00', 'mmkr', 'evt-1')" + ) diff --git a/tests/test_protocol_events.py b/tests/test_protocol_events.py index 8128129..2ebcc34 100644 --- a/tests/test_protocol_events.py +++ b/tests/test_protocol_events.py @@ -245,15 +245,16 @@ def test_hook_event__rejects_unknown_field(self): legacy_field="ignored", # pyright: ignore[reportCallIssue] ) - def test_hook_event__source_accepts_only_known_enum_values(self): - """AbstractEventSource is a closed enum — free-form strings fail validation.""" - with pytest.raises(ValidationError): - AbstractHookEvent( - session_id="s1", - event_type=AbstractEventType.NOTIFICATION, - source="some_other_agent", # type: ignore[arg-type] - working_directory="/repo", - ) + def test_hook_event__source_accepts_open_third_party_strings(self): + """Source is an open string — third-party collectors don't need an enum extension.""" + event = AbstractHookEvent( + session_id="s1", + event_type=AbstractEventType.NOTIFICATION, + source="some_other_agent", + working_directory="/repo", + ) + + assert event.source == "some_other_agent" def test_hook_event__event_type_accepts_only_known_enum_values(self): """AbstractEventType is a closed enum — wire-format drift fails validation.""" diff --git a/tests/test_protocol_ingest.py b/tests/test_protocol_ingest.py new file mode 100644 index 0000000..ffd0315 --- /dev/null +++ b/tests/test_protocol_ingest.py @@ -0,0 +1,199 @@ +"""Tests for envelope ingestion into slopometry storage.""" + +from datetime import datetime +from unittest.mock import patch + +import pytest +from click.testing import CliRunner + +from slopometry.cli import cli +from slopometry.core.database import EventDatabase +from slopometry.core.models.protocol.events import AbstractEventType +from slopometry.core.protocol.ingest import ingest_envelopes +from slopometry.core.protocol.schema import EventEnvelope + + +@pytest.fixture +def isolated_storage(tmp_path): + """Isolated database for ingestion tests.""" + return EventDatabase(db_path=tmp_path / "test.db") + + +def _mmkr_envelope( + session_id: str, + event_id: str, + kind: str, + seq: int | None = None, + tool_name: str | None = None, + raw: dict | None = None, + **tool_result_fields, +) -> EventEnvelope: + return EventEnvelope( + source="mmkr", + session_id=session_id, + event_id=event_id, + kind=kind, + seq=seq, + tool_name=tool_name, + occurred_at=datetime(2026, 6, 30, 12, 0, 0), + raw=raw or {}, + **tool_result_fields, + ) + + +class TestIngestEnvelopes: + def test_ingest_envelopes__stores_third_party_events_with_canonical_kinds(self, isolated_storage): + db = isolated_storage + envelopes = [ + _mmkr_envelope("s-1", "evt-1", "tool_call_started", seq=1, tool_name="check_issue_responses", raw={"tick": 31}), + _mmkr_envelope( + "s-1", "evt-2", "tool_call_completed", seq=2, tool_name="check_issue_responses", raw={"ok": True} + ), + _mmkr_envelope("s-1", "evt-3", "turn_completed", seq=3), + ] + + report = ingest_envelopes(envelopes, db=db, working_directory="/repo") + + assert report.inserted == 3 + events = db.get_session_events("s-1") + assert [e.event_type for e in events] == [ + AbstractEventType.TOOL_CALL_STARTED, + AbstractEventType.TOOL_CALL_COMPLETED, + AbstractEventType.TURN_COMPLETED, + ] + assert all(e.source == "mmkr" for e in events) + assert [e.sequence_number for e in events] == [1, 2, 3] + assert events[0].tool_call is not None + assert events[0].tool_call.tool_name == "check_issue_responses" + assert events[0].metadata == {"tick": 31} + assert all(e.event_id is not None for e in events) + assert all(e.id is not None for e in events) + + def test_ingest_envelopes__stores_tool_result_timing_and_error_fields(self, isolated_storage): + db = isolated_storage + envelopes = [ + _mmkr_envelope( + "s-1", + "evt-1", + "tool_call_completed", + seq=1, + tool_name="Bash", + duration_ms=120, + exit_code=1, + error_message="command failed", + ), + ] + + ingest_envelopes(envelopes, db=db) + + event = db.get_session_events("s-1")[0] + assert event.tool_call is not None + assert event.tool_call.duration_ms == 120 + assert event.tool_call.exit_code == 1 + assert event.tool_call.error_message == "command failed" + + def test_ingest_envelopes__preserves_open_source_strings_through_storage_round_trip(self, isolated_storage): + db = isolated_storage + envelopes = [_mmkr_envelope("s-7", "evt-1", "turn_completed", seq=1)] + envelopes[0].source = "totally-unknown-agent" + + ingest_envelopes(envelopes, db=db) + reloaded = EventDatabase(db_path=db.db_path).get_session_events("s-7") + + assert [e.source for e in reloaded] == ["totally-unknown-agent"] + + def test_ingest_envelopes__assigns_sequence_numbers_when_envelope_lacks_seq(self, isolated_storage): + db = isolated_storage + envelopes = [ + _mmkr_envelope("s-2", "evt-1", "tool_call_started"), + _mmkr_envelope("s-2", "evt-2", "tool_call_completed"), + ] + + ingest_envelopes(envelopes, db=db) + + events = db.get_session_events("s-2") + assert [e.sequence_number for e in events] == [1, 2] + + def test_ingest_envelopes__interleaves_source_and_assigned_sequences_without_collision(self, isolated_storage): + db = isolated_storage + envelopes = [ + _mmkr_envelope("s-4", "evt-1", "tool_call_started", seq=1), + _mmkr_envelope("s-4", "evt-2", "tool_call_completed"), + _mmkr_envelope("s-4", "evt-3", "turn_completed", seq=7), + _mmkr_envelope("s-4", "evt-4", "notification"), + ] + + ingest_envelopes(envelopes, db=db) + + events = sorted(db.get_session_events("s-4"), key=lambda e: e.event_id) + assert [e.sequence_number for e in events] == [1, 2, 7, 8] + + def test_ingest_envelopes__continues_after_highest_stored_sequence_for_existing_session(self, isolated_storage): + db = isolated_storage + ingest_envelopes([_mmkr_envelope("s-5", "evt-1", "turn_completed", seq=5)], db=db) + + ingest_envelopes([_mmkr_envelope("s-5", "evt-2", "notification")], db=db) + + events = db.get_session_events("s-5") + assert [e.sequence_number for e in events] == [5, 6] + + def test_ingest_envelopes__skips_duplicates_when_backfilling_identical_trace(self, isolated_storage): + db = isolated_storage + envelopes = [_mmkr_envelope("s-3", "evt-1", "turn_completed", seq=1)] + + first = ingest_envelopes(envelopes, db=db) + second = ingest_envelopes(envelopes, db=db) + + assert (first.inserted, first.skipped_duplicates) == (1, 0) + assert (second.inserted, second.skipped_duplicates) == (0, 1) + assert len(db.get_session_events("s-3")) == 1 + + +class TestIngestCli: + def test_ingest__ingests_jsonl_from_stdin_for_third_party_source(self, tmp_path): + jsonl = ( + '{"source": "mmkr", "session_id": "s-cli", "event_id": "e1", "kind": "tool_call_started", "seq": 1}\n' + '{"source": "mmkr", "session_id": "s-cli", "event_id": "e2", "kind": "turn_completed", "seq": 2}\n' + ) + db_path = tmp_path / "test.db" + + with patch("slopometry.core.settings.settings.database_path", db_path): + result = CliRunner().invoke(cli, ["ingest", "--source", "mmkr"], input=jsonl) + + assert result.exit_code == 0 + assert "Ingested 2 events" in result.output + events = EventDatabase(db_path=db_path).get_session_events("s-cli") + assert [e.event_type for e in events] == [AbstractEventType.TOOL_CALL_STARTED, AbstractEventType.TURN_COMPLETED] + + def test_ingest__reads_from_file_when_flag_given(self, tmp_path): + trace_file = tmp_path / "trace.jsonl" + trace_file.write_text('{"source": "mmkr", "session_id": "s-file", "event_id": "e1", "kind": "notification"}\n') + db_path = tmp_path / "test.db" + + with patch("slopometry.core.settings.settings.database_path", db_path): + result = CliRunner().invoke( + cli, ["ingest", "--source", "mmkr", "--file", str(trace_file), "--working-directory", str(tmp_path)] + ) + + assert result.exit_code == 0 + events = EventDatabase(db_path=db_path).get_session_events("s-file") + assert [e.event_type for e in events] == [AbstractEventType.NOTIFICATION] + assert events[0].working_directory == str(tmp_path) + + def test_ingest__fails_loudly_when_envelope_source_mismatches_flag(self, tmp_path): + jsonl = '{"source": "mmkr", "session_id": "s-cli", "event_id": "e1", "kind": "turn_completed"}\n' + + with patch("slopometry.core.settings.settings.database_path", tmp_path / "test.db"): + result = CliRunner().invoke(cli, ["ingest", "--source", "other-agent"], input=jsonl) + + assert result.exit_code == 2 + assert "does not match --source" in result.output + + def test_ingest__fails_loudly_when_envelope_is_invalid(self, tmp_path): + jsonl = '{"source": "mmkr", "session_id": "s-cli", "event_id": "e1", "kind": "tick_complete"}\n' + + with patch("slopometry.core.settings.settings.database_path", tmp_path / "test.db"): + result = CliRunner().invoke(cli, ["ingest", "--source", "mmkr"], input=jsonl) + + assert result.exit_code == 2 + assert "Invalid envelope at line 1" in result.output diff --git a/tests/test_protocol_schema.py b/tests/test_protocol_schema.py new file mode 100644 index 0000000..8659fcd --- /dev/null +++ b/tests/test_protocol_schema.py @@ -0,0 +1,68 @@ +"""Tests for the external envelope schema used by `slopometry ingest`.""" + +import pytest +from pydantic import ValidationError + +from slopometry.core.models.protocol.events import AbstractEventType +from slopometry.core.protocol.schema import EventEnvelope + + +class TestEventEnvelopeValidation: + def test_event_envelope__accepts_minimal_envelope_with_defaults(self): + envelope = EventEnvelope(source="mmkr", session_id="s1", event_id="e1", kind="tool_call_started") + + assert envelope.kind is AbstractEventType.TOOL_CALL_STARTED + assert envelope.schema_version == 1 + assert envelope.seq is None + assert envelope.duration_ms is None + assert envelope.exit_code is None + assert envelope.error_message is None + assert envelope.raw == {} + + def test_event_envelope__accepts_arbitrary_third_party_source_strings(self): + envelope = EventEnvelope(source="my-custom-agent", session_id="s1", event_id="e1", kind="notification") + + assert envelope.source == "my-custom-agent" + + def test_event_envelope__requires_event_id_for_idempotent_backfill(self): + with pytest.raises(ValidationError): + EventEnvelope(source="mmkr", session_id="s1", kind="tool_call_started") + + def test_event_envelope__rejects_unsupported_schema_version(self): + with pytest.raises(ValidationError, match="Unsupported schema_version"): + EventEnvelope(source="mmkr", session_id="s1", event_id="e1", kind="tool_call_started", schema_version=2) + + def test_event_envelope__rejects_unknown_kind_with_helpful_message(self): + with pytest.raises(ValidationError) as exc_info: + EventEnvelope(source="mmkr", session_id="s1", event_id="e1", kind="tick_complete") + + assert "tool_call_started" in str(exc_info.value) + + def test_event_envelope__rejects_empty_source(self): + with pytest.raises(ValidationError): + EventEnvelope(source="", session_id="s1", event_id="e1", kind="tool_call_started") + + def test_event_envelope__rejects_sequence_below_one(self): + with pytest.raises(ValidationError): + EventEnvelope(source="mmkr", session_id="s1", event_id="e1", kind="tool_call_started", seq=0) + + def test_event_envelope__preserves_unknown_fields_for_forward_compatibility(self): + envelope = EventEnvelope( + source="mmkr", session_id="s1", event_id="e1", kind="turn_completed", harness_trace_url="https://example.com" + ) + + assert envelope.model_extra == {"harness_trace_url": "https://example.com"} + + def test_event_envelope__round_trips_through_jsonl_line(self): + payload = ( + '{"source": "mmkr", "schema_version": 1, "session_id": "s-42", "event_id": "tick-31",' + ' "parent_session_id": "s-1", "seq": 31, "occurred_at": "2026-06-30T12:00:00",' + ' "kind": "tool_call_started", "tool_name": "check_issue_responses", "raw": {"tick": 31}}' + ) + + envelope = EventEnvelope.model_validate_json(payload) + + assert envelope.parent_session_id == "s-1" + assert envelope.tool_name == "check_issue_responses" + assert envelope.seq == 31 + assert envelope.raw == {"tick": 31}