diff --git a/apps/sol/routes.py b/apps/sol/routes.py index eed4b70bf..877d12187 100644 --- a/apps/sol/routes.py +++ b/apps/sol/routes.py @@ -143,16 +143,29 @@ def _parse_use_events( return result -def _parse_use_file(use_file: Path) -> dict[str, Any] | None: - """Parse a use JSONL file and extract metadata. +def _get_use_day(use_file: Path) -> str: + """Get the logical day for a use from its request event. - Returns dict with: id, name, start, status, prompt, facet, failed, - runtime_seconds, thinking_count, tool_count, cost, model, provider, - error_message. - Returns None if file cannot be parsed. + Prefers the ``day`` field from the request event (the day being processed) + over the use_id timestamp (when the agent actually ran). This ensures + overnight think uses appear under the day they processed. """ - from think.cortex_client import get_use_end_state + use_id = use_file.stem.replace("_active", "") + try: + with open(use_file, "r") as f: + first_line = f.readline().strip() + if first_line: + request_event = json.loads(first_line) + req_day = request_event.get("day") + if req_day: + return req_day + except (json.JSONDecodeError, IOError): + pass + return _use_id_to_day(use_id) + +def _parse_active_use_file(use_file: Path) -> dict[str, Any] | None: + """Parse an active use JSONL file for the day listing.""" try: with open(use_file, "r") as f: lines = f.readlines() @@ -168,82 +181,65 @@ def _parse_use_file(use_file: Path) -> dict[str, Any] | None: if request_event.get("event") != "request": return None - is_active = "_active.jsonl" in use_file.name - use_id = use_file.stem.replace("_active", "") + thinking_count = 0 + tool_count = 0 + model = None + provider = request_event.get("provider") - # Parse events using shared helper - event_data = _parse_use_events(lines[1:]) + for line in lines[1:]: + line = line.strip() + if not line: + continue + try: + event = json.loads(line) + except json.JSONDecodeError: + continue - use_info: dict[str, Any] = { + event_type = event.get("event") + if event_type == "thinking": + thinking_count += 1 + elif event_type == "tool_start": + tool_count += 1 + elif event_type == "start": + model = event.get("model") + provider = provider or event.get("provider") + + output_file = None + if request_event.get("output"): + out_path = _resolve_output_path(request_event, state.journal_root) + if out_path and out_path.exists(): + req_day = request_event.get("day") + day_dir = Path(state.journal_root) / req_day if req_day else None + try: + if day_dir and out_path.is_relative_to(day_dir): + output_file = str(out_path.relative_to(day_dir)) + else: + output_file = str(out_path.relative_to(state.journal_root)) + except ValueError: + output_file = None + + use_id = use_file.stem.replace("_active", "") + return { "id": use_id, "name": request_event["name"], "start": request_event.get("ts", 0), - "status": "running" if is_active else "completed", + "status": "running", "prompt": request_event.get("prompt", ""), "facet": request_event.get("facet"), "failed": False, "runtime_seconds": None, - "thinking_count": event_data["thinking_count"], - "tool_count": event_data["tool_count"], + "thinking_count": thinking_count, + "tool_count": tool_count, "cost": None, - "model": event_data["model"], - "provider": request_event.get("provider") or event_data.get("provider"), - "error_message": event_data["error_message"], + "model": model, + "provider": provider, + "error_message": None, + "output_file": output_file, } - - # Check for output file (generators only) - output_file = None - req_output = request_event.get("output") - if req_output: - out_path = _resolve_output_path(request_event, state.journal_root) - if out_path and out_path.exists(): - req_day = request_event.get("day") - day_dir = Path(state.journal_root) / req_day if req_day else None - if day_dir and out_path.is_relative_to(day_dir): - output_file = str(out_path.relative_to(day_dir)) - else: - output_file = str(out_path.relative_to(state.journal_root)) - use_info["output_file"] = output_file - - # For completed uses, determine end state and calculate cost - if not is_active: - end_state = get_use_end_state(use_id) - use_info["failed"] = end_state in ("error", "unknown") - - # Calculate runtime from finish or error timestamp - end_ts = event_data["finish_ts"] or event_data["error_ts"] - if end_ts and use_info["start"]: - use_info["runtime_seconds"] = (end_ts - use_info["start"]) / 1000.0 - - # Calculate cost - use_info["cost"] = calc_agent_cost(event_data["model"], event_data["usage"]) - - return use_info - except (json.JSONDecodeError, IOError): + except (json.JSONDecodeError, OSError): return None -def _get_use_day(use_file: Path) -> str: - """Get the logical day for a use from its request event. - - Prefers the ``day`` field from the request event (the day being processed) - over the use_id timestamp (when the agent actually ran). This ensures - overnight think uses appear under the day they processed. - """ - use_id = use_file.stem.replace("_active", "") - try: - with open(use_file, "r") as f: - first_line = f.readline().strip() - if first_line: - request_event = json.loads(first_line) - req_day = request_event.get("day") - if req_day: - return req_day - except (json.JSONDecodeError, IOError): - pass - return _use_id_to_day(use_id) - - def _get_uses_for_day(day: str, facet_filter: str | None = None) -> list[dict]: """Get all talent uses for a specific day. @@ -260,7 +256,7 @@ def _get_uses_for_day(day: str, facet_filter: str | None = None) -> list[dict]: if not talents_dir.exists(): return [] - uses = [] + uses: list[dict[str, Any]] = [] # Read day index for completed uses day_index_path = talents_dir / f"{day}.jsonl" @@ -280,17 +276,30 @@ def _get_uses_for_day(day: str, facet_filter: str | None = None) -> list[dict]: if facet_filter is not None and entry.get("facet") != facet_filter: continue - # Locate the actual file for full parsing - use_id = entry.get("use_id", "") - name = entry["name"] - safe_name = name.replace(":", "--") - use_file = talents_dir / safe_name / f"{use_id}.jsonl" - if not use_file.exists(): + use_id = entry.get("use_id") or entry.get("agent_id") + if not use_id: continue - use_info = _parse_use_file(use_file) - if use_info: - uses.append(use_info) + status = entry.get("status") + uses.append( + { + "id": use_id, + "name": entry.get("name"), + "start": entry.get("ts"), + "status": status, + "prompt": entry.get("prompt"), + "facet": entry.get("facet"), + "failed": status in ("error", "unknown"), + "runtime_seconds": entry.get("runtime_seconds"), + "thinking_count": entry.get("thinking_count"), + "tool_count": entry.get("tool_count"), + "cost": entry.get("cost"), + "model": entry.get("model"), + "provider": entry.get("provider"), + "error_message": entry.get("error_message"), + "output_file": entry.get("output_file"), + } + ) except OSError as exc: logging.warning("Failed to read use day index %s: %s", day_index_path, exc) @@ -301,7 +310,7 @@ def _get_uses_for_day(day: str, facet_filter: str | None = None) -> list[dict]: if _get_use_day(use_file) != day: continue - use_info = _parse_use_file(use_file) + use_info = _parse_active_use_file(use_file) if not use_info: continue @@ -311,7 +320,7 @@ def _get_uses_for_day(day: str, facet_filter: str | None = None) -> list[dict]: uses.append(use_info) # Sort by start time (newest first) - uses.sort(key=lambda x: x["start"], reverse=True) + uses.sort(key=lambda x: x.get("start") or 0, reverse=True) return uses diff --git a/tests/test_app_sol.py b/tests/test_app_sol.py index a572f8ffe..8ad7f2742 100644 --- a/tests/test_app_sol.py +++ b/tests/test_app_sol.py @@ -323,6 +323,168 @@ class TestApiOutputFile: assert resp.status_code == 404 +@pytest.fixture +def sol_listing_client(tmp_path, monkeypatch): + """Create a sol app client backed by a temporary journal.""" + from flask import Flask + + from apps.sol.routes import sol_bp + from convey import state + + app = Flask(__name__) + app.register_blueprint(sol_bp) + + talents_dir = tmp_path / "talents" + talents_dir.mkdir() + + monkeypatch.setattr(state, "journal_root", str(tmp_path)) + monkeypatch.setattr("apps.sol.routes.get_facets", lambda: {}) + monkeypatch.setattr("apps.sol.routes._build_talents_meta", lambda: {}) + + return app.test_client(), talents_dir + + +def _write_day_index(talents_dir: Path, day: str, entries: list[dict]) -> Path: + path = talents_dir / f"{day}.jsonl" + lines = [json.dumps(entry) + "\n" for entry in entries] + path.write_text("".join(lines), encoding="utf-8") + return path + + +class TestApiTalentsDayListing: + """Tests for day-index-backed talent listing.""" + + def test_index_only_entry_returns_full_summary(self, sol_listing_client): + """A complete day-index entry is enough without a per-use file.""" + client, talents_dir = sol_listing_client + day = "20990101" + entry = { + "use_id": "4070908800001", + "name": "flow", + "day": day, + "facet": "work", + "ts": 4070908800000, + "status": "error", + "runtime_seconds": 12.3, + "provider": "google", + "model": "gemini-2.5-flash", + "schedule": "daily", + "thinking_count": 4, + "tool_count": 2, + "cost": 0.0123, + "error_message": "rate limited", + "output_file": "talents/flow.md", + "prompt": "Summarize the day", + } + _write_day_index(talents_dir, day, [entry]) + + resp = client.get(f"/app/sol/api/talents/{day}") + + assert resp.status_code == 200 + uses = resp.get_json()["uses"] + assert len(uses) == 1 + assert uses[0] == { + "id": "4070908800001", + "name": "flow", + "start": 4070908800000, + "status": "error", + "prompt": "Summarize the day", + "facet": "work", + "failed": True, + "runtime_seconds": 12.3, + "thinking_count": 4, + "tool_count": 2, + "cost": 0.0123, + "model": "gemini-2.5-flash", + "provider": "google", + "error_message": "rate limited", + "output_file": "talents/flow.md", + } + + def test_legacy_agent_id_entry_returns_with_blank_new_fields( + self, sol_listing_client + ): + """Legacy agent_id day-index entries are visible with missing fields blank.""" + client, talents_dir = sol_listing_client + day = "20990102" + agent_id = "4070995200001" + _write_day_index( + talents_dir, + day, + [ + { + "agent_id": agent_id, + "name": "entities", + "day": day, + "facet": "personal", + "ts": 4070995200000, + "status": "completed", + "runtime_seconds": 8.4, + "provider": "google", + "model": "gemini-2.5-flash-lite", + } + ], + ) + + resp = client.get(f"/app/sol/api/talents/{day}") + + assert resp.status_code == 200 + use = resp.get_json()["uses"][0] + assert use["id"] == agent_id + assert use["failed"] is False + for field in ( + "thinking_count", + "tool_count", + "cost", + "error_message", + "output_file", + "prompt", + ): + assert use[field] is None + + def test_current_thin_entry_returns_without_rewriting_index( + self, sol_listing_client + ): + """Current thin use_id entries return with missing fields blank.""" + client, talents_dir = sol_listing_client + day = "20990103" + index_path = _write_day_index( + talents_dir, + day, + [ + { + "use_id": "4071081600001", + "name": "knowledge_graph", + "day": day, + "facet": None, + "ts": 4071081600000, + "status": "completed", + "runtime_seconds": 9.1, + "provider": "anthropic", + "model": "claude-sonnet-4-5", + "schedule": "daily", + } + ], + ) + before = index_path.read_bytes() + + resp = client.get(f"/app/sol/api/talents/{day}") + + assert resp.status_code == 200 + use = resp.get_json()["uses"][0] + assert use["id"] == "4071081600001" + for field in ( + "thinking_count", + "tool_count", + "cost", + "error_message", + "output_file", + "prompt", + ): + assert use[field] is None + assert index_path.read_bytes() == before + + class TestApiUpdatedDays: """Tests for api_updated_days endpoint.""" diff --git a/think/cortex.py b/think/cortex.py index 4736255a7..ef744fe09 100644 --- a/think/cortex.py +++ b/think/cortex.py @@ -29,7 +29,9 @@ from pathlib import Path from typing import Any, Dict, Optional from think.callosum import CallosumConnection +from think.models import calc_agent_cost from think.runner import _atomic_symlink +from think.talent import get_output_path from think.talents import TALENT_EXECUTION_MODULE from think.utils import get_journal, get_project_root, get_rev, now_ms @@ -662,6 +664,40 @@ class CortexService: except Exception as e: self.logger.error(f"Failed to complete talent file {use_id}: {e}") + def _summarize_output_file(self, request: Dict[str, Any]) -> str | None: + """Return the API-facing output path if it exists at completion time.""" + if not request.get("output"): + return None + + try: + if request.get("output_path"): + out_path = Path(request["output_path"]) + else: + req_day = request.get("day") + if not req_day: + return None + day_dir = self.talents_dir.parent / req_day + req_env = request.get("env") or {} + out_path = get_output_path( + day_dir, + request["name"], + segment=request.get("segment"), + output_format=request.get("output"), + facet=request.get("facet"), + stream=req_env.get("SOL_STREAM"), + ) + + if not out_path.exists(): + return None + + req_day = request.get("day") + day_dir = self.talents_dir.parent / req_day if req_day else None + if day_dir and out_path.is_relative_to(day_dir): + return str(out_path.relative_to(day_dir)) + return str(out_path.relative_to(self.talents_dir.parent)) + except (OSError, ValueError, KeyError): + return None + def _append_day_index( self, use_id: str, request: Dict[str, Any], completed_path: Path ) -> None: @@ -677,30 +713,43 @@ class CortexService: start_ts = request.get("ts", 0) - # Read last few lines to find finish/error event for runtime + thinking_count = 0 + tool_count = 0 + finish_usage = None + error_message = None + model = None runtime_seconds = None status = "completed" try: with open(completed_path, "r") as f: lines = f.readlines() - for line in reversed(lines[-10:]): + for line in lines: line = line.strip() if not line: continue try: event = json.loads(line) event_type = event.get("event") + if event_type == "thinking": + thinking_count += 1 + elif event_type == "tool_start": + tool_count += 1 + elif event_type == "start": + model = event.get("model") + if event_type == "finish": + status = "completed" + finish_usage = event.get("usage") end_ts = event.get("ts", 0) if end_ts and start_ts: runtime_seconds = round((end_ts - start_ts) / 1000.0, 1) - break if event_type == "error": status = "error" + msg = event.get("error", "") + error_message = msg[:200] if msg else None end_ts = event.get("ts", 0) if end_ts and start_ts: runtime_seconds = round((end_ts - start_ts) / 1000.0, 1) - break except json.JSONDecodeError: continue except Exception: @@ -715,8 +764,14 @@ class CortexService: "status": status, "runtime_seconds": runtime_seconds, "provider": request.get("provider"), - "model": request.get("model"), + "model": model, "schedule": request.get("schedule"), + "thinking_count": thinking_count, + "tool_count": tool_count, + "cost": calc_agent_cost(model, finish_usage), + "error_message": error_message if status == "error" else None, + "output_file": self._summarize_output_file(request), + "prompt": request.get("prompt", ""), } day_index_path = self.talents_dir / f"{day}.jsonl"