diff --git a/.gitignore b/.gitignore index 8dc4d04..77f5187 100644 --- a/.gitignore +++ b/.gitignore @@ -80,6 +80,19 @@ Thumbs.db # Paper-finder build artifacts services/paper-finder/.venv/ services/paper-finder/**/__pycache__/ +tmp_runtime_test/ -# axle_check.py runtime scratch (normalized snippet sent to AXLE) -.axle_last_check.lean +# Log analysis generated outputs +src/core/log_analysis/sample_data/parsed/ +*.db +*.db-shm +*.db-wal + +# Large local log corpora +log_analysis/raw/ +log_analysis/parsed/ + +.neurico/neurico.db +.neurico/*.db +.neurico/*.db-shm +.neurico/*.db-wal \ No newline at end of file diff --git a/src/core/event_store.py b/src/core/event_store.py new file mode 100644 index 0000000..9d80492 --- /dev/null +++ b/src/core/event_store.py @@ -0,0 +1,462 @@ +""" +Structured event and failure storage for NeuriCo workspaces. +This module introduces a local-first event store for browser/visualizer work. It intentionally keeps the +existing JSON/JSONL files for resume compatibility while adding SQLite as the structured query layer. +Files writtern under each workspace: +- .neurico/neurico.db structured SQLite database +- .neurico/events.jsonl append-only event fallback/debug log +- .neurico/failures.jsonl append-only failure log +""" + +from __future__ import annotations +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, Optional +import json +import sqlite3 +import uuid + +def _utc_now_iso() -> str: + """Return an ISO-8601 UTC timestamp.""" + return datetime.now(timezone.utc).isoformat() + +def _json_dumps(data: Optional[Dict[str, Any]]) -> Optional[str]: + if data is None: + return None + return json.dumps(data, ensure_ascii=False, sort_keys=True) + +class EventStore: + """ + Append structured events and failures for one workspace. + Use SQLite for timeline/visualizer queries and JSONL as an easy-to-inspect fallback. + This keeps the refactor local-first and avoids introducing a server database before the browser/visualizer exists. + """ + + def __init__(self, work_dir: Path, run_id: Optional[str] = None) -> None: + self.work_dir = Path(work_dir) + self.neurico_dir = self.work_dir / ".neurico" + self.neurico_dir.mkdir(parents=True, exist_ok=True) + self.db_path = self.neurico_dir / "neurico.db" + self.events_jsonl_path = self.neurico_dir / "events.jsonl" + self.failures_jsonl_path = self.neurico_dir / "failures.jsonl" + self.run_id = run_id or self._load_or_create_run_id() + self._initialize_db() + self.ensure_run(workspace_path=str(self.work_dir)) + + def _load_or_create_run_id(self) -> str: + run_id_path = self.neurico_dir / "run_id" + if run_id_path.exists(): + existing = run_id_path.read_text(encoding="utf-8").strip() + if existing: + return existing + run_id = str(uuid.uuid4()) + run_id_path.write_text(run_id + "\n", encoding="utf-8") + return run_id + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self.db_path) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA foreign_keys=ON") + return conn + + def _initialize_db(self) -> None: + with self._connect() as conn: + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS runs ( + run_id TEXT PRIMARY KEY, + idea_id TEXT, + provider TEXT, + workspace_path TEXT NOT NULL, + status TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + resumed_from_run_id TEXT + ); + CREATE TABLE IF NOT EXISTS events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + run_id TEXT NOT NULL, + timestamp TEXT NOT NULL, + source TEXT NOT NULL, + stage TEXT, + phase TEXT, + event_type TEXT NOT NULL, + status TEXT, + message TEXT, + data_json TEXT, + FOREIGN KEY(run_id) REFERENCES runs(run_id) + ); + CREATE INDEX IF NOT EXISTS idx_events_run_time + ON events ( + run_id, + timestamp + ); + CREATE TABLE IF NOT EXISTS stage_states ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + run_id TEXT NOT NULL, + stage TEXT NOT NULL, + status TEXT NOT NULL, + started_at TEXT, + completed_at TEXT, + success INTEGER, + outputs_json TEXT, + updated_at TEXT NOT NULL, + FOREIGN KEY(run_id) REFERENCES runs(run_id) + ); + CREATE INDEX IF NOT EXISTS idx_stage_states_run_stage + ON stage_states ( + run_id, + stage, + updated_at + ); + CREATE TABLE IF NOT EXISTS failures ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + run_id TEXT NOT NULL, + timestamp TEXT NOT NULL, + source TEXT NOT NULL, + stage TEXT, + phase TEXT, + severity TEXT NOT NULL, + error_type TEXT, + reason TEXT NOT NULL, + recoverable INTEGER NOT NULL, + traceback TEXT, + context_json TEXT, + resolved INTEGER DEFAULT 0, + FOREIGN KEY(run_id) REFERENCES runs(run_id) + ); + CREATE INDEX IF NOT EXISTS idx_failures_run_time + ON failures ( + run_id, + timestamp + ); + CREATE TABLE IF NOT EXISTS agent_logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + run_id TEXT NOT NULL, + timestamp TEXT NOT NULL, + stage TEXT, + stream TEXT, + level TEXT, + message TEXT NOT NULL, + raw_json TEXT, + FOREIGN KEY(run_id) REFERENCES runs(run_id) + ); + CREATE TABLE IF NOT EXISTS trajectory_steps ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + run_id TEXT NOT NULL, + step_index INTEGER NOT NULL, + timestamp TEXT, + source_file TEXT, + line_no INTEGER, + actor TEXT, + event_type TEXT NOT NULL, + raw_event_type TEXT, + stage TEXT, + phase TEXT, + status TEXT, + message TEXT, + command TEXT, + exit_code INTEGER, + tool_name TEXT, + file_path TEXT, + input_text TEXT, + output_text TEXT, + raw_json TEXT, + FOREIGN KEY(run_id) REFERENCES runs(run_id) + ); + + CREATE INDEX IF NOT EXISTS idx_trajectory_steps_run_step + ON trajectory_steps (run_id, step_index); + + CREATE INDEX IF NOT EXISTS idx_trajectory_steps_type + ON trajectory_steps (run_id, event_type); + + CREATE INDEX IF NOT EXISTS idx_trajectory_steps_stage_phase + ON trajectory_steps (run_id, stage, phase); + + CREATE TABLE IF NOT EXISTS run_artifacts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + run_id TEXT NOT NULL, + artifact_path TEXT NOT NULL, + artifact_type TEXT, + source_file TEXT, + created_by_step_index INTEGER, + exists_on_disk INTEGER, + size_bytes INTEGER, + observed_at TEXT NOT NULL, + FOREIGN KEY(run_id) REFERENCES runs(run_id) + ); + + CREATE INDEX IF NOT EXISTS idx_run_artifacts_run + ON run_artifacts (run_id); + + CREATE INDEX IF NOT EXISTS idx_run_artifacts_path + ON run_artifacts (run_id, artifact_path); + """ + ) + def ensure_run( + self, + *, + workspace_path: str, + idea_id: Optional[str] = None, + provider: Optional[str] = None, + status: str = "active", + resumed_from_run_id: Optional[str] = None, + ) -> None: + now = _utc_now_iso() + with self._connect() as conn: + conn.execute( + """ + INSERT INTO runs ( + run_id, idea_id, provider, workspace_path, status, + created_at, updated_at, resumed_from_run_id + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(run_id) DO UPDATE SET + idea_id = COALESCE(excluded.idea_id, runs.idea_id), + provider = COALESCE(excluded.provider, runs.provider), + workspace_path = excluded.workspace_path, + status = excluded.status, + updated_at = excluded.updated_at, + resumed_from_run_id = COALESCE(excluded.resumed_from_run_id, runs.resumed_from_run_id) + """, + ( + self.run_id, + idea_id, + provider, + workspace_path, + status, + now, + now, + resumed_from_run_id, + ), + ) + + def append_event( + self, + *, + source: str, + event_type: str, + stage: Optional[str] = None, + phase: Optional[str] = None, + status: Optional[str] = None, + message: Optional[str] = None, + data: Optional[Dict[str, Any]] = None, + ) -> None: + timestamp = _utc_now_iso() + entry = { + "run_id": self.run_id, + "timestamp": timestamp, + "source": source, + "stage": stage, + "phase": phase, + "event_type": event_type, + "status": status, + "message": message, + "data": data or {}, + } + with self._connect() as conn: + conn.execute( + """ + INSERT INTO events ( + run_id, timestamp, source, stage, phase, event_type, + status, message, data_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + self.run_id, + timestamp, + source, + stage, + phase, + event_type, + status, + message, + _json_dumps(data), + ), + ) + self._append_jsonl(self.events_jsonl_path, entry) + + def append_failure( + self, + *, + source: str, + reason: str, + stage: Optional[str] = None, + phase: Optional[str] = None, + severity:str = "error", + recoverable: bool = True, + error_type: Optional[str] = None, + traceback_text: Optional[str] = None, + context: Optional[Dict[str, Any]] = None, + ) -> None: + timestamp = _utc_now_iso() + entry = { + "run_id": self.run_id, + "timestamp": timestamp, + "source": source, + "stage": stage, + "phase": phase, + "severity": severity, + "error_type": error_type, + "reason": reason, + "recoverable": recoverable, + "traceback": traceback_text, + "context": context or {}, + "resolved": False, + } + with self._connect() as conn: + conn.execute( + """ + INSERT INTO failures ( + run_id, timestamp, source, stage, phase, severity, + error_type, reason, recoverable, traceback, context_json, resolved + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0) + """, + ( + self.run_id, + timestamp, + source, + stage, + phase, + severity, + error_type, + reason, + int(recoverable), + traceback_text, + _json_dumps(context), + ), + ) + self._append_jsonl(self.failures_jsonl_path, entry) + self.append_event( + source=source, + event_type="failure", + stage=stage, + phase=phase, + status="recoverable" if recoverable else "failed", + message=reason, + data={ + "severity": severity, + "error_type": error_type, + "recoverable": recoverable, + }, + ) + + def upsert_stage_state( + self, + *, + stage: str, + status: str, + success: Optional[bool] = None, + started_at: Optional[str] = None, + completed_at: Optional[str] = None, + outputs: Optional[Dict[str, Any]] = None, + ) -> None: + with self._connect() as conn: + conn.execute( + """ + INSERT INTO stage_states ( + run_id, stage, status, started_at, completed_at, + success, outputs_json, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + self.run_id, + stage, + status, + started_at, + completed_at, + None if success is None else int(success), + _json_dumps(outputs), + _utc_now_iso(), + ), + ) + def append_trajectory_step( + self, + *, + step_index: int, + event_type: str, + actor: str, + timestamp: Optional[str] = None, + source_file: Optional[str] = None, + line_no: Optional[int] = None, + raw_event_type: Optional[str] = None, + stage: Optional[str] = None, + phase: Optional[str] = None, + status: Optional[str] = None, + message: Optional[str] = None, + command: Optional[str] = None, + exit_code: Optional[int] = None, + tool_name: Optional[str] = None, + file_path: Optional[str] = None, + input_text: Optional[str] = None, + output_text: Optional[str] = None, + raw: Optional[Dict[str, Any]] = None, + ) -> None: + with self._connect() as conn: + conn.execute( + """ + INSERT INTO trajectory_steps ( + run_id, step_index, timestamp, source_file, line_no, + actor, event_type, raw_event_type, stage, phase, status, + message, command, exit_code, tool_name, file_path, + input_text, output_text, raw_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + self.run_id, + step_index, + timestamp, + source_file, + line_no, + actor, + event_type, + raw_event_type, + stage, + phase, + status, + message, + command, + exit_code, + tool_name, + file_path, + input_text, + output_text, + _json_dumps(raw), + ), + ) + + def append_run_artifact( + self, + *, + artifact_path: str, + artifact_type: Optional[str] = None, + source_file: Optional[str] = None, + created_by_step_index: Optional[int] = None, + exists_on_disk: Optional[bool] = None, + size_bytes: Optional[int] = None, + ) -> None: + with self._connect() as conn: + conn.execute( + """ + INSERT INTO run_artifacts ( + run_id, artifact_path, artifact_type, source_file, + created_by_step_index, exists_on_disk, size_bytes, observed_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + self.run_id, + artifact_path, + artifact_type, + source_file, + created_by_step_index, + None if exists_on_disk is None else int(exists_on_disk), + size_bytes, + _utc_now_iso(), + ), + ) + + + @staticmethod + def _append_jsonl(path: Path, entry: Dict[str, Any]) -> None: + with open(path, "a", encoding="utf-8") as f: + f.write(json.dumps(entry, ensure_ascii=False) + "\n") diff --git a/src/core/log_analysis/__init__.py b/src/core/log_analysis/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/core/log_analysis/cli.py b/src/core/log_analysis/cli.py new file mode 100644 index 0000000..f10d82e --- /dev/null +++ b/src/core/log_analysis/cli.py @@ -0,0 +1,47 @@ +import argparse +import json +from pathlib import Path + +from core.log_analysis.run import find_run_repo +from core.log_analysis.ingest import load_prompt_texts, load_transcript_events +from core.log_analysis.parser.prompt_parser import parse_task_spec +from core.log_analysis.trajectory.trajectory_builder import build_trajectory +from core.log_analysis.datastore.event_store_writer import EventStoreWriter + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--raw-root", required=True) + parser.add_argument("--parsed-root", required=True) + args = parser.parse_args() + raw_root = Path(args.raw_root) + parsed_root = Path(args.parsed_root) + parsed_root.mkdir(parents=True, exist_ok=True) + + repos = find_run_repo(raw_root) + + for repo in repos: + prompt_texts = load_prompt_texts(repo) + raw_events = load_transcript_events(repo) + task = parse_task_spec(repo.run_id, prompt_texts) + trajectory = build_trajectory(repo, task, raw_events) + out_dir = parsed_root / repo.title_slug + out_dir.mkdir(parents=True, exist_ok=True) + trajectory_json_path = out_dir / "trajectory.json" + trajectory_events_path = out_dir / "trajectory_events.jsonl" + trajectory_json_path.write_text( + trajectory.model_dump_json(indent=2), + encoding="utf-8", + ) + with open(trajectory_events_path, "w", encoding="utf-8") as f: + for step in trajectory.steps: + f.write(json.dumps(step.model_dump(mode="json"), ensure_ascii=False) + "\n") + writer = EventStoreWriter(db_work_dir=out_dir) + writer.write(trajectory) + print( + f"Parsed {repo.run_id}: " + f"{len(trajectory.steps)} steps, " + f"{len(trajectory.artifacts)} artifacts, " + f"{len(trajectory.failures)} failures" + ) +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/src/core/log_analysis/datastore/__init__.py b/src/core/log_analysis/datastore/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/core/log_analysis/datastore/event_store_writer.py b/src/core/log_analysis/datastore/event_store_writer.py new file mode 100644 index 0000000..b74f823 --- /dev/null +++ b/src/core/log_analysis/datastore/event_store_writer.py @@ -0,0 +1,80 @@ +from pathlib import Path + +from core.event_store import EventStore +from core.log_analysis.models import RunTrajectory + + +class EventStoreWriter: + """ + Adapter that writes high-level RunTrajectory objects into EventStore. + + EventStore is the low-level SQLite/JSONL storage layer. + EventStoreWriter is the bridge from parsed trajectory models to database rows. + """ + + def __init__(self, db_work_dir: Path, run_id: str | None = None): + self.db_work_dir = Path(db_work_dir) + self.run_id = run_id + + def write(self, trajectory: RunTrajectory) -> None: + """ + Persist one parsed trajectory into the workspace-local EventStore. + + The EventStore run_id is set to trajectory.run_id so database rows can be + joined across runs, trajectory_steps, run_artifacts, failures, and events. + """ + event_store = EventStore( + self.db_work_dir, + run_id=self.run_id or trajectory.run_id, + ) + + event_store.ensure_run( + workspace_path=str(trajectory.root_dir), + idea_id=trajectory.run_id, + status=trajectory.status, + ) + + for step in trajectory.steps: + event_store.append_trajectory_step( + step_index=step.step_index, + timestamp=step.timestamp, + source_file=step.source_file, + line_no=step.line_no, + actor=step.actor, + event_type=step.event_type, + raw_event_type=step.raw_event_type, + stage=step.stage, + phase=step.phase, + status=step.status, + message=step.message, + command=step.command, + exit_code=step.exit_code, + tool_name=step.tool_name, + file_path=step.file_path, + input_text=step.input_text, + output_text=step.output_text, + raw=step.raw_json, + ) + + for artifact in trajectory.artifacts: + event_store.append_run_artifact( + artifact_path=artifact.artifact_path, + artifact_type=artifact.artifact_type, + source_file=artifact.source_file, + created_by_step_index=artifact.created_by_step_index, + exists_on_disk=artifact.exists_on_disk, + size_bytes=artifact.size_bytes, + ) + + for failure in trajectory.failures: + event_store.append_failure( + source="log_analysis", + reason=failure.reason, + stage=failure.stage, + phase=failure.phase, + severity=failure.severity, + recoverable=failure.recoverable, + error_type=failure.error_type, + traceback_text=failure.traceback, + context=failure.context, + ) \ No newline at end of file diff --git a/src/core/log_analysis/ingest.py b/src/core/log_analysis/ingest.py new file mode 100644 index 0000000..35e2950 --- /dev/null +++ b/src/core/log_analysis/ingest.py @@ -0,0 +1,39 @@ +import json +from pathlib import Path +from .models import RawTranscriptEvent, RunRepo + +def read_text_file(path: Path) -> str: + return Path(path).read_text(encoding="utf-8", errors="replace") + +def load_prompt_texts(repo: RunRepo) -> dict[str, str]: + result: dict[str, str] = {} + for path in repo.prompt_files: + result[str(path)] = read_text_file(path) + return result + +def load_transcript_events(repo: RunRepo) -> list[RawTranscriptEvent]: + events: list[RawTranscriptEvent] = [] + for path in repo.transcript_files: + with open(path, "r", encoding="utf-8", errors="replace") as f: + for line_no, line in enumerate(f, start=1): + line = line.strip() + if not line: + continue + try: + raw = json.loads(line) + except json.JSONDecodeError: + raw = { + "type": "json_parse_error", + "message": line, + } + events.append( + RawTranscriptEvent( + run_id=repo.run_id, + source_file=str(path), + line_no=line_no, + raw_event_type=raw.get("type"), + timestamp=raw.get("timestamp"), + raw=raw, + ) + ) + return events diff --git a/src/core/log_analysis/models.py b/src/core/log_analysis/models.py new file mode 100644 index 0000000..d9039e4 --- /dev/null +++ b/src/core/log_analysis/models.py @@ -0,0 +1,86 @@ +from pathlib import Path +from typing import Any, Literal, Optional +from pydantic import BaseModel, Field + +Actor = Literal["agent", "tool", "system", "user", "orchestrator", "unknown"] + +class RunRepo(BaseModel): + run_id: str + title_slug: str + root_dir: Path + prompt_files: list[Path] = Field(default_factory=list) + transcript_files: list[Path] = Field(default_factory=list) + artifact_files: list[Path] = Field(default_factory=list) + +class TaskSpec(BaseModel): + run_id: str + title: Optional[str] = None + domain: Optional[str] = None + hypothesis: Optional[str] = None + expected_phases: list[str] = Field(default_factory=list) + expected_deliverables: list[str] = Field(default_factory=list) + source_files: list[str] = Field(default_factory=list) + +class RawTranscriptEvent(BaseModel): + run_id: str + source_file: str + line_no: int + raw_event_type: Optional[str] = None + timestamp: Optional[str] = None + raw: dict[str, Any] + +class TrajectoryStep(BaseModel): + run_id: str + step_index: int = 0 + timestamp: Optional[str] = None + source_file: Optional[str] = None + line_no: Optional[int] = None + + actor: Actor = "unknown" + event_type: str + raw_event_type: Optional[str] = None + + stage: Optional[str] = None + phase: Optional[str] = None + status: Optional[str] = None + + message: Optional[str] = None + command: Optional[str] = None + exit_code: Optional[int] = None + tool_name: Optional[str] = None + file_path: Optional[str] = None + + input_text: Optional[str] = None + output_text: Optional[str] = None + raw_json: Optional[dict[str, Any]] = None + +class ArtifactRecord(BaseModel): + run_id: str + artifact_path: str + artifact_type: Optional[str] = None + source_file: Optional[str] = None + created_by_step_index: Optional[int] = None + exists_on_disk: Optional[bool] = None + size_bytes: Optional[int] = None + +class FailureRecord(BaseModel): + run_id: str + step_index: Optional[int] = None + stage: Optional[str] = None + phase: Optional[str] = None + severity: str = "error" + error_type: Optional[str] = None + reason: str + recoverable: bool = False + traceback: Optional[str] = None + context: dict[str, Any] = Field(default_factory=dict) + +class RunTrajectory(BaseModel): + run_id: str + title_slug: str + root_dir: Path + task: Optional[TaskSpec] = None + steps: list[TrajectoryStep] = Field(default_story=list) + artifacts: list[ArtifactRecord] = Field(default_story=list) + failures: list[FailureRecord] = Field(default_story=list) + status: str = "unknown" diff --git a/src/core/log_analysis/parser/__init__.py b/src/core/log_analysis/parser/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/core/log_analysis/parser/event_normalizer.py b/src/core/log_analysis/parser/event_normalizer.py new file mode 100644 index 0000000..839a356 --- /dev/null +++ b/src/core/log_analysis/parser/event_normalizer.py @@ -0,0 +1,340 @@ +from typing import Any + +from core.log_analysis.models import RawTranscriptEvent, TrajectoryStep, FailureRecord + + +IGNORED_RAW_TYPES = { + "thread.started", + "turn.started", + "turn.completed", +} + + +def should_skip_raw_event( + raw_type: str | None, + item_type: str | None, + message: Any = None, +) -> bool: + """ + Skip transcript lifecycle records that do not add useful trajectory content. + + Phase 1 should preserve meaningful agent/tool actions, but avoid flooding the + trajectory with thread/turn bookkeeping events. + """ + if raw_type in IGNORED_RAW_TYPES: + return True + + # Skip pure lifecycle item events only when they have no useful item type or text. + if item_type is None and raw_type in {"item.started", "item.completed", "item.updated"} and not message: + return True + + return False + + +def extract_file_path(raw: dict[str, Any], item: dict[str, Any]) -> str | None: + """ + Extract a file path from known Codex transcript file-change shapes. + + Different transcript versions may store the path under different keys. + Keep this defensive so Phase 1 can parse many historical logs. + """ + direct_path = ( + item.get("path") + or item.get("file_path") + or item.get("file") + or item.get("filename") + or raw.get("path") + or raw.get("file_path") + or raw.get("file") + or raw.get("filename") + ) + + if direct_path: + return str(direct_path) + + # Some logs store file changes as a list of objects. + changes = item.get("changes") or raw.get("changes") + if isinstance(changes, list) and changes: + first = changes[0] + if isinstance(first, dict): + path = ( + first.get("path") + or first.get("file_path") + or first.get("file") + or first.get("filename") + ) + if path: + return str(path) + + return None + + +def normalize_event(raw_event: RawTranscriptEvent) -> tuple[list[TrajectoryStep], list[FailureRecord]]: + """ + Convert one raw transcript event into zero or more normalized trajectory steps. + + This is the main Phase 1 normalization layer: + - agent_message becomes plan / claim / revision / final_summary / agent_message + - command_execution becomes a tool command step + - file_change becomes an artifact-related step + - web_search becomes a tool_call + - todo_list becomes todo_update + - unknown but non-empty records become raw_event for debugging + """ + raw = raw_event.raw + raw_type = raw.get("type") + + item = raw.get("item") or raw.get("data") or {} + if not isinstance(item, dict): + item = {} + + item_type = item.get("type") or raw.get("item_type") + + message = ( + raw.get("message") + or raw.get("text") + or item.get("text") + or item.get("message") + or item.get("content") + ) + + if should_skip_raw_event(raw_type, item_type, message): + return [], [] + + steps: list[TrajectoryStep] = [] + failures: list[FailureRecord] = [] + + if item_type == "agent_message": + steps.append( + TrajectoryStep( + run_id=raw_event.run_id, + timestamp=raw_event.timestamp, + source_file=raw_event.source_file, + line_no=raw_event.line_no, + actor="agent", + event_type=classify_agent_message(str(message or "")), + raw_event_type=raw_type, + status=_status_from_raw_type(raw_type), + message=str(message) if message is not None else None, + raw_json=raw, + ) + ) + + elif item_type == "command_execution": + command = item.get("command") or item.get("cmd") or raw.get("command") or raw.get("cmd") + output = ( + item.get("output") + or item.get("stdout") + or item.get("stderr") + or raw.get("output") + or raw.get("stdout") + or raw.get("stderr") + ) + exit_code = item.get("exit_code", raw.get("exit_code")) + + status = _status_from_raw_type(raw_type) + if exit_code not in (None, 0): + status = "failed" + + steps.append( + TrajectoryStep( + run_id=raw_event.run_id, + timestamp=raw_event.timestamp, + source_file=raw_event.source_file, + line_no=raw_event.line_no, + actor="tool", + event_type="command_execution", + raw_event_type=raw_type, + status=status, + command=str(command) if command is not None else None, + exit_code=exit_code, + output_text=str(output) if output is not None else None, + raw_json=raw, + ) + ) + + if exit_code not in (None, 0): + failures.append( + FailureRecord( + run_id=raw_event.run_id, + reason=f"Command failed with exit code {exit_code}: {command}", + error_type="command_failed", + recoverable=True, + context={ + "source_file": raw_event.source_file, + "line_no": raw_event.line_no, + "command": command, + }, + ) + ) + + elif item_type == "file_change": + # File-change records are important for reconstructing artifacts such as + # planning.md, REPORT.md, scripts, figures, and results. Codex logs may + # store file paths under different keys, so use extract_file_path(). + file_path = extract_file_path(raw, item) + + steps.append( + TrajectoryStep( + run_id=raw_event.run_id, + timestamp=raw_event.timestamp, + source_file=raw_event.source_file, + line_no=raw_event.line_no, + actor="agent", + event_type="file_change", + raw_event_type=raw_type, + status=_status_from_raw_type(raw_type), + file_path=file_path, + message=str(message) if message is not None else None, + raw_json=raw, + ) + ) + + elif item_type == "web_search": + query = item.get("query") or raw.get("query") or message + + steps.append( + TrajectoryStep( + run_id=raw_event.run_id, + timestamp=raw_event.timestamp, + source_file=raw_event.source_file, + line_no=raw_event.line_no, + actor="tool", + event_type="tool_call", + tool_name="web_search", + raw_event_type=raw_type, + status=_status_from_raw_type(raw_type), + input_text=str(query) if query is not None else None, + raw_json=raw, + ) + ) + + elif item_type == "todo_list": + steps.append( + TrajectoryStep( + run_id=raw_event.run_id, + timestamp=raw_event.timestamp, + source_file=raw_event.source_file, + line_no=raw_event.line_no, + actor="agent", + event_type="todo_update", + raw_event_type=raw_type, + status=_status_from_raw_type(raw_type), + message=str(message) if message is not None else None, + raw_json=raw, + ) + ) + + else: + steps.append( + TrajectoryStep( + run_id=raw_event.run_id, + timestamp=raw_event.timestamp, + source_file=raw_event.source_file, + line_no=raw_event.line_no, + actor="system", + event_type="raw_event", + raw_event_type=raw_type, + status=_status_from_raw_type(raw_type), + message=str(message) if message is not None else None, + raw_json=raw, + ) + ) + + return steps, failures + + +# NOTE: +# This classifier is intentionally rule-based and conservative for Phase 1. +# The goal is not perfect semantic labeling yet; it is to create stable, +# readable trajectory categories for visualization. We avoid labeling every +# "completed" message as final_summary because many are only local progress +# updates, such as dependency installation or command completion. +def classify_agent_message(text: str) -> str: + """ + Classify agent natural-language messages into coarse trajectory event types. + + Keep this conservative: + - "final_summary" should only mean the run or major stage is actually ending. + - Ordinary progress messages like "package installation completed" should stay + as "agent_message", "claim", or "plan". + """ + lowered = text.lower() + + final_markers = [ + "final report", + "final summary", + "task complete", + "research complete", + "experiment complete", + "all deliverables", + "completed the full", + "the work is complete", + "i have completed the", + "successfully completed the research", + ] + + if any(marker in lowered for marker in final_markers): + return "final_summary" + + revision_markers = [ + "fix", + "retry", + "instead", + "adjust", + "modify", + "change approach", + "fallback", + "recover", + ] + + if any(marker in lowered for marker in revision_markers): + return "revision" + + plan_markers = [ + "i will", + "i’ll", + "plan", + "next", + "now i will", + "i'm going to", + "i am going to", + ] + + if any(marker in lowered for marker in plan_markers): + return "plan" + + claim_markers = [ + "found", + "result", + "shows", + "indicates", + "suggests", + "confirms", + "the workspace is correct", + "gpu", + "available", + ] + + if any(marker in lowered for marker in claim_markers): + return "claim" + + return "agent_message" + + +def _status_from_raw_type(raw_type: Any) -> str | None: + if not raw_type: + return None + + raw_type = str(raw_type) + + if raw_type.endswith(".started"): + return "started" + + if raw_type.endswith(".completed"): + return "completed" + + if raw_type.endswith(".updated"): + return "updated" + + return None \ No newline at end of file diff --git a/src/core/log_analysis/parser/prompt_parser.py b/src/core/log_analysis/parser/prompt_parser.py new file mode 100644 index 0000000..9739d21 --- /dev/null +++ b/src/core/log_analysis/parser/prompt_parser.py @@ -0,0 +1,71 @@ +import re +from core.log_analysis.models import TaskSpec + +DEFAULT_PHASES = [ + "motivation_novelty", + "planning", + "implementation", + "analysis", + "documentation", + "validation", +] + +DEFAULT_DELIVERABLES = [ + "planning.md", + "REPORT.md", + "README.md", + "resources.md", + "literature_review.md", + "papers/", + "datasets/", + "code/", + "results/", + "figures/", +] + +def _extract_after_heading(text: str, headings: list[str]) -> str | None: + for heading in headings: + pattern = rf"{re.escape(heading)}\s*:?\s*\n+(.+?)(?:\n\n|\Z)" + match = re.search(pattern, text, flags=re.IGNORECASE | re.DOTALL) + if match: + return match.group(1).strip() + return None + +def parse_task_spec(run_id: str, prompt_texts: dict[str, str]) -> TaskSpec: + combined = "\n\n".join(prompt_texts.values()) + title = _extract_after_heading( + combined, + ["RESEARCH TITLE", "## RESEARCH TITLE", "Research Title"], + ) + domain = _extract_after_heading( + combined, + ["RESEARCH DOMAIN", "## RESEARCH DOMAIN", "Research Domain"], + ) + hypothesis = _extract_after_heading( + combined, + [ + "RESEARCH HYPOTHESIS", + "HYPOTHESIS / RESEARCH QUESTION", + "## HYPOTHESIS / RESEARCH QUESTION", + ], + ) + expected_phases = [ + phase for phase in DEFAULT_PHASES + if phase.replace("_", " ").lower() in combined.lower() + or phase.lower() in combined.lower() + ] + if not expected_phases: + expected_phases = DEFAULT_PHASES + expected_deliverables = [ + item for item in DEFAULT_DELIVERABLES + if item.lower() in combined.lower() + ] + return TaskSpec( + run_id=run_id, + title=title, + domain=domain, + hypothesis=hypothesis, + expected_phases=expected_phases, + expected_deliverables=expected_deliverables, + source_files=list(prompt_texts.keys()), + ) diff --git a/src/core/log_analysis/parser/transcript_parser.py b/src/core/log_analysis/parser/transcript_parser.py new file mode 100644 index 0000000..a699a74 --- /dev/null +++ b/src/core/log_analysis/parser/transcript_parser.py @@ -0,0 +1,5 @@ +from core.log_analysis.models import RawTranscriptEvent +def parse_codex_events( + raw_events: list[RawTranscriptEvent], +) -> list[RawTranscriptEvent]: + return raw_events \ No newline at end of file diff --git a/src/core/log_analysis/run.py b/src/core/log_analysis/run.py new file mode 100644 index 0000000..017102c --- /dev/null +++ b/src/core/log_analysis/run.py @@ -0,0 +1,49 @@ +from pathlib import Path +from .models import RunRepo + +PROMPT_PATTERNS = [ + "*prompt*.txt", + "*instructions*.txt", +] + +TRANSCRIPT_PATTERNS = [ + "*transctipt*.jsonl", + "*.jsonl", +] + +def find_run_repo(raw_root: Path) -> list[RunRepo]: + raw_root = Path(raw_root) + repos: list[RunRepo] = [] + for task_dir in sorted(raw_root.iterdir()): + if not task_dir.is_dir(): + continue + prompt_files : list[Path] = [] + for pattern in PROMPT_PATTERNS: + prompt_files.extend(task_dir.glob(pattern)) + transcript_files: list[Path] = [] + for pattern in TRANSCRIPT_PATTERNS: + transcript_files.extend(task_dir.glob(pattern)) + prompt_files = sorted(set(prompt_files)) + transcript_files = sorted(set(transcript_files)) + + if not prompt_files and not transcript_files: + continue + + artifact_files = [ + path for path in task_dir.rglob("*") + if path.is_file() + and ".git" not in path.parts + and path.name != ".DS_Store" + ] + + repos.append( + RunRepo( + run_id=task_dir.name, + title_slug=task_dir.name, + root_dir=task_dir, + prompt_files=prompt_files, + transcript_files=transcript_files, + artifact_files=artifact_files, + ) + ) + return repos diff --git a/src/core/log_analysis/trajectory/__init__.py b/src/core/log_analysis/trajectory/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/core/log_analysis/trajectory/trajectory_builder.py b/src/core/log_analysis/trajectory/trajectory_builder.py new file mode 100644 index 0000000..e0c98b6 --- /dev/null +++ b/src/core/log_analysis/trajectory/trajectory_builder.py @@ -0,0 +1,424 @@ +from pathlib import Path +from typing import Iterable + +from core.log_analysis.models import ( + ArtifactRecord, + FailureRecord, + RunTrajectory, + TaskSpec, + TrajectoryStep, +) +from core.log_analysis.parser.event_normalizer import normalize_event + + +def build_trajectory( + repo, + task: TaskSpec | None, + raw_events: Iterable, +) -> RunTrajectory: + """ + Build one structured trajectory from a run/repo bundle and raw transcript events. + + Phase 1 responsibilities: + - normalize raw transcript rows into trajectory steps + - assign stable step_index + - infer coarse stage and phase labels + - collect observed artifacts + - infer a rough run status + + This should not do deep quality evaluation yet. Pattern detection and artifact + validation belong to later phases. + """ + steps: list[TrajectoryStep] = [] + failures: list[FailureRecord] = [] + + for raw_event in raw_events: + normalized_steps, normalized_failures = normalize_event(raw_event) + steps.extend(normalized_steps) + failures.extend(normalized_failures) + + # Stable order: source file then line number. + # This works for transcript JSONL files where line order is the event order. + steps.sort(key=lambda s: (s.source_file or "", s.line_no or 0)) + + for idx, step in enumerate(steps, start=1): + step.step_index = idx + step.stage = step.stage or infer_stage(step.source_file or "") + step.phase = step.phase or infer_phase(step) + + artifacts = collect_artifacts(repo, steps) + status = infer_run_status(steps, failures, artifacts) + + return RunTrajectory( + run_id=repo.run_id, + title_slug=getattr(repo, "title_slug", repo.run_id), + root_dir=repo.root_dir, + task=task, + steps=steps, + artifacts=artifacts, + failures=failures, + status=status, + ) + + +def infer_stage(source_file: str) -> str | None: + """ + Infer coarse pipeline stage from source filename. + + This is intentionally filename-based for Phase 1 because the run folders + already contain stage-specific transcript names. + """ + name = source_file.lower() + + if "resource_finder" in name: + return "resource_finder" + + if "execution" in name or "research" in name: + return "experiment_runner" + + if "paper_writer" in name: + return "paper_writer" + + return None + + +# Phase inference is intentionally heuristic in Phase 1. +# We infer coarse phases from commands/messages/file paths so the visualizer can +# group long trajectories into readable sections. This is not a scientific +# judgment yet; later phases can add a stronger phase detector. +def infer_phase(step: TrajectoryStep) -> str | None: + text = " ".join( + value + for value in [ + step.message, + step.command, + step.output_text, + step.file_path, + ] + if value + ).lower() + + if ( + text.strip() in {"pwd", "/bin/bash -lc pwd"} + or "rg --files" in text + or "git status" in text + or "ls -la" in text + or "find ." in text + or "date -iseconds" in text + or "date -i" in text + ): + return "workspace_check" + + if not text: + return None + + # Prompt/context loading at the beginning of a transcript. + if "reading prompt from stdin" in text: + return "prompt_read" + + # Agent/system skill inspection. These are not research artifacts; they are + # capability/context review steps before the agent acts. + if ( + ".codex/skills/" in text + or ".claude/skills/" in text + or "skill.md" in text + or "/skills/" in text + ): + return "capability_review" + + # Workspace/project checks. These reduce empty phases for shell bookkeeping. + if ( + text.strip() in {"pwd", "/bin/bash -lc pwd"} + or "rg --files" in text + or "git status" in text + or "ls -la" in text + or "find ." in text + or "date -iseconds" in text + or "cat .resource_finder_complete" in text + ): + return "workspace_check" + + # Resource-finder setup / collection actions. + if ( + "mkdir -p papers datasets code" in text + or "paper-finder" in text + or "download every paper" in text + or "paper-finder returned" in text + or "ranked papers" in text + or "selected set" in text + or "relevance" in text + or "download" in text and "paper" in text + ): + return "resource_collection" + + # Resource-finder waiting/progress messages. + if ( + "still waiting on paper-finder" in text + or "local paper-finder call is still running" in text + or "expected diligent-search latency" in text + or "query remains active" in text + ): + return "resource_collection" + + # Experiment execution / smoke runs / model loading. + if ( + "smoke test" in text + or "smoke run" in text + or "model download" in text + or "checkpoint" in text + or "loading the qwen" in text + or "main experiment finished" in text + or "token-level kl" in text + or "real model logits" in text + or "python src/run_" in text + or "run_divergence_experiment.py" in text + or "--examples-per-source" in text + or "--batch-size" in text + or "--bootstrap-iterations" in text + ): + return "experimentation" + + # Result inspection / comparison / metrics analysis. + if ( + "structured comparison" in text + or "predictor metrics" in text + or "row count" in text + or "results/" in text + or "figures/" in text + or "summary.json" in text + or "metrics" in text + ): + return "analysis" + + # Documentation-writing progress messages and final docs. + if ( + "readme" in text + or "report.md" in text + or "code walkthrough" in text + or "code_walkthrough.md" in text + or "final readme" in text + or "final report" in text + or "execution note" in text + ): + return "documentation" + + # Todo updates usually represent planning/checkpoint management. + if step.event_type == "todo_update": + return "planning" + + if text.strip() in {"pwd", "/bin/bash -lc pwd"} or " pwd" in text: + return "workspace_check" + + if "rg --files" in text or "ls " in text or "find " in text: + return "workspace_check" + + if "nvidia-smi" in text or "cuda" in text or "gpu" in text: + return "environment_setup" + + if "sed -n" in text and ( + "literature_review.md" in text + or "resources.md" in text + or "readme.md" in text + or "dataset_summary.json" in text + ): + return "resource_review" + + if any( + marker in text + for marker in [ + "literature_review.md", + "resources.md", + "datasets/readme.md", + "code/readme.md", + "papers/", + "datasets/", + "code/", + "dataset_summary.json", + ] + ): + return "resource_review" + + if "planning.md" in text or "motivation" in text or "novelty" in text: + return "planning" + + + if ( + "python src/run_" in text + or "run_divergence_experiment.py" in text + or "run_experiment" in text + or "--examples-per-source" in text + or "--batch-size" in text + or "--bootstrap-iterations" in text + ): + return "experimentation" + + if ( + "python src/" in text + or "python -m" in text + or "python ./" in text + or "run_" in text + or "src/" in text + ): + return "implementation" + + + if "python - <<'py'" in text or 'python - <<"' in text: + if any(marker in text for marker in ["pandas", "json", "results/", "figures/", "summary"]): + return "analysis" + return "implementation" + + + if ( + "uv add" in text + or "pip install" in text + or "pyproject.toml" in text + or "python --version" in text + or "uv --version" in text + or "py_compile" in text + or "test -d .venv" in text + ): + return "environment_setup" + + if ( + "results/" in text + or "analysis" in text + or "figure" in text + or "figures/" in text + or ".csv" in text + or ".json" in text + ): + return "analysis" + + if "report.md" in text or "readme.md" in text or "paper_draft" in text or ".tex" in text: + return "documentation" + + if "validation" in text or "reproduce" in text or "reproducibility" in text: + return "validation" + + return None + + +def collect_artifacts(repo, steps: list[TrajectoryStep]) -> list[ArtifactRecord]: + """ + Collect artifacts from two sources: + 1. Files that exist under the run folder. + 2. File paths mentioned by file_change trajectory steps. + + Phase 1 records observed artifacts only. It does not yet judge whether they + are complete or scientifically valid. + """ + artifacts: dict[str, ArtifactRecord] = {} + + for path in getattr(repo, "artifact_files", []): + path = Path(path) + + try: + rel = path.relative_to(repo.root_dir) + artifact_path = str(rel) + except ValueError: + artifact_path = str(path) + + artifacts[artifact_path] = ArtifactRecord( + run_id=repo.run_id, + artifact_path=artifact_path, + artifact_type=infer_artifact_type(path), + exists_on_disk=path.exists(), + size_bytes=path.stat().st_size if path.exists() and path.is_file() else None, + ) + + for step in steps: + if not step.file_path: + continue + + if step.file_path not in artifacts: + artifacts[step.file_path] = ArtifactRecord( + run_id=repo.run_id, + artifact_path=step.file_path, + artifact_type=infer_artifact_type(Path(step.file_path)), + source_file=step.source_file, + created_by_step_index=step.step_index, + exists_on_disk=None, + size_bytes=None, + ) + + return list(artifacts.values()) + + +def infer_artifact_type(path: Path) -> str | None: + name = path.name.lower() + suffix = path.suffix.lower() + + if name in { + "report.md", + "readme.md", + "planning.md", + "resources.md", + "literature_review.md", + }: + return "markdown_report" + + if name in { + "resource_finder_prompt.txt", + "research_prompt.txt", + "paper_writer_prompt.txt", + "session_instructions.txt", + }: + return "prompt" + + if "transcript" in name and suffix == ".jsonl": + return "transcript" + + if suffix == ".py": + return "code" + + if suffix in {".json", ".jsonl", ".csv", ".gz", ".parquet", ".pkl"}: + return "data_or_results" + + if suffix in {".png", ".jpg", ".jpeg", ".svg"}: + return "figure" + + if suffix == ".pdf": + return "paper_or_pdf" + + if suffix in {".txt", ".log"}: + return "log_or_text" + + if suffix in {".tex", ".bib", ".sty"}: + return "paper_draft" + + return "other" + + +def infer_run_status( + steps: list[TrajectoryStep], + failures: list[FailureRecord], + artifacts: list[ArtifactRecord], +) -> str: + """ + Infer a rough Phase 1 run status. + + This is intentionally lightweight. Later artifact validation can replace this + with stronger checks such as "REPORT.md exists and references real results." + """ + has_final = any(s.event_type == "final_summary" for s in steps) + + has_report_artifact = any( + artifact.artifact_path.lower().endswith("report.md") + for artifact in artifacts + ) + + has_report_step = any( + (s.file_path or "").lower().endswith("report.md") + for s in steps + ) + + if (has_final or has_report_artifact or has_report_step) and failures: + return "completed_with_warnings" + + if has_final or has_report_artifact or has_report_step: + return "completed" + + if failures: + return "failed_or_incomplete" + + return "unknown"