"""Read-side inventory of wrapper sidecar logs. `/.okstra/tasks/**/runs/*/prompts/*.log` 를 스캔해 크기·mtime 을 수집하고 task-type/worker/seq 를 path 에서 파싱한 뒤, top-largest 목록과 per-task 합계를 만든다. skill markdown 이 raw `find`(OS별 분기) + 손-집계로 하던 일을 코드 SSOT 로 옮긴 것 — read-only(삭제하지 않는다). cleanup 명령 제시는 호출자(skill)에 남긴다. """ from __future__ import annotations import argparse import json import sys from pathlib import Path from .fixed_text import line from okstra_project import resolve_project_root from okstra_ctl.json_boundary import load_owned_object def _project_id(project_root: Path) -> str: try: return load_owned_object( project_root / ".okstra" / "project.json", artifact="project configuration", ).get("projectId", "") except (OSError, ValueError): return "" def _parse(parts: tuple, stem: str) -> dict: group = parts[0] if len(parts) > 0 else "" task_id = parts[1] if len(parts) > 1 else "" phase = "" if "runs" in parts: i = parts.index("runs") if i + 1 < len(parts): phase = parts[i + 1] worker = stem.split("-worker-prompt-")[0] if "-worker-prompt-" in stem else "" seq = stem.rsplit("-", 1)[-1] if "-" in stem else "" return {"taskGroup": group, "taskId": task_id, "phase": phase, "worker": worker, "seq": seq} def _per_task(files: list[dict]) -> list[dict]: acc: dict[str, dict] = {} for f in files: tk = f["taskKey"] if not tk: continue a = acc.get(tk) if a is None: a = acc[tk] = {"taskKey": tk, "fileCount": 0, "totalBytes": 0, "oldestEpoch": f["mtimeEpoch"], "newestEpoch": f["mtimeEpoch"]} a["fileCount"] += 1 a["totalBytes"] += f["sizeBytes"] a["oldestEpoch"] = min(a["oldestEpoch"], f["mtimeEpoch"]) a["newestEpoch"] = max(a["newestEpoch"], f["mtimeEpoch"]) return sorted(acc.values(), key=lambda a: a["totalBytes"], reverse=True) def _prompt_pair(log_path: Path, transcript_bytes: int) -> dict: prompt_path = log_path.with_suffix(".md") if not prompt_path.is_file(): return { "transcriptPath": str(log_path), "transcriptBytes": transcript_bytes, "promptPath": "", "promptBytes": 0, "transcriptToPromptRatio": None, } try: prompt_bytes = prompt_path.stat().st_size except OSError: prompt_bytes = 0 ratio = ( round(transcript_bytes / prompt_bytes, 2) if prompt_bytes > 0 else None ) return { "transcriptPath": str(log_path), "transcriptBytes": transcript_bytes, "promptPath": str(prompt_path), "promptBytes": prompt_bytes, "transcriptToPromptRatio": ratio, } def scan_logs(logs_root: Path, project_root: Path, top: int = 20) -> dict: project_id = _project_id(project_root) files: list[dict] = [] if logs_root.is_dir(): for p in logs_root.rglob("*.log"): parts = p.relative_to(logs_root).parts if "runs" not in parts or "prompts" not in parts: continue try: size = p.stat().st_size mtime = int(p.stat().st_mtime) except OSError: continue meta = _parse(parts, p.stem) tk = f"{project_id}:{meta['taskGroup']}:{meta['taskId']}" if meta["taskGroup"] else "" files.append({ "path": str(p), "sizeBytes": size, "mtimeEpoch": mtime, "taskKey": tk, **meta, **_prompt_pair(p, size), }) files.sort(key=lambda f: f["sizeBytes"], reverse=True) return { "logsRoot": str(logs_root), "topLargest": files[:top], "perTask": _per_task(files), "totals": { "fileCount": len(files), "totalBytes": sum(f["sizeBytes"] for f in files), "taskCount": len({f["taskKey"] for f in files if f["taskKey"]}), "promptBytes": sum(f["promptBytes"] for f in files), "transcriptBytes": sum(f["transcriptBytes"] for f in files), "pairedFileCount": sum(1 for f in files if f["promptPath"]), }, } def render_result_text(result: dict) -> str: parts = ["# Okstra Log Inventory\n\n", line("Project root", result.get("projectRoot")), line("Logs root", result.get("logsRoot"))] totals = result.get("totals") if isinstance(result.get("totals"), dict) else {} for key, label in (("fileCount", "File count"), ("totalBytes", "Total bytes"), ("taskCount", "Task count"), ("promptBytes", "Prompt bytes"), ("transcriptBytes", "Transcript bytes"), ("pairedFileCount", "Paired file count")): parts.append(line(label, totals.get(key))) for row in result.get("topLargest", []): if isinstance(row, dict): parts.extend(("\n## Log\n\n", line("Task key", row.get("taskKey")), line("Phase", row.get("phase")), line("Worker", row.get("worker")), line("Sequence", row.get("seq")), line("Size bytes", row.get("sizeBytes")), line("Modified epoch", row.get("mtimeEpoch")), line("Path", row.get("path")))) for row in result.get("perTask", []): if isinstance(row, dict): parts.extend(("\n## Task total\n\n", line("Task key", row.get("taskKey")), line("File count", row.get("fileCount")), line("Total bytes", row.get("totalBytes")), line("Oldest epoch", row.get("oldestEpoch")), line("Newest epoch", row.get("newestEpoch")))) return "".join(parts) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( prog="okstra log-report", description="Inventory wrapper sidecar logs by size and task (read-only).") parser.add_argument("--project-root", default="") parser.add_argument("--cwd", default=".") parser.add_argument("--top", type=int, default=20, help="topLargest entry count") parser.add_argument("--json", action="store_true", help="emit JSON (always on)") parser.add_argument("--text", action="store_true", help="emit fixed text fields") args = parser.parse_args(argv) project_root = resolve_project_root(explicit_root=args.project_root, cwd=args.cwd) result = {"ok": True, "projectRoot": str(project_root)} result.update(scan_logs(project_root / ".okstra" / "tasks", project_root, top=args.top)) output = render_result_text(result) if args.text else json.dumps(result, ensure_ascii=False, indent=2) print(output, end="" if args.text else "\n") return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))