"""Read-side recap for a task: assemble run-to-run phase transitions, append a recap Q&A log, and write agent-authored notes. Writes only under /recap/recap-log.jsonl and /notes/; never touches task-manifest / catalog / timeline. """ from __future__ import annotations import argparse import datetime as dt import json import sys from pathlib import Path from okstra_ctl import next_phase from okstra_ctl.clarification_items import sidecar_answers, user_response_sidecars from okstra_ctl.ids import slugify_task_segment from okstra_ctl.incremental_scope import preview_link_availability_for_report from okstra_ctl.json_boundary import JsonBoundaryError, load_owned_object from okstra_ctl.jsonl import append_jsonl from okstra_ctl.paths import task_timeline_file from okstra_project import read_task_key from okstra_ctl.run_context import dir_flock from okstra_ctl.final_report_paths import timeline_report_record_rel from okstra_ctl.task_target import resolve_task_root, project_rel NOTE_KINDS = ("verification-evidence", "decision-draft", "analysis-note") def _load_timeline(task_root: Path) -> list[dict]: path = task_timeline_file(task_root) if not path.is_file(): return [] try: data = load_owned_object(path, artifact="task timeline") except (JsonBoundaryError, OSError): return [] runs = data.get("runs", []) return runs if isinstance(runs, list) else [] _STRUCTURAL_CHANGE_NOTE = ( "An answer that overturns the selected option, restructures the Stage Map, " 'or changes the recommended approach requires --full-reason "". The back-trace resolves stages; it cannot judge whether the ' "plan's shape survived, so that call is the lead's and must be declared." ) def _latest_planning_report(runs: list[dict], project_root: Path) -> Path | None: """The most recent `implementation-planning` report still on disk.""" for run in reversed(runs): if not isinstance(run, dict): continue if run.get("taskType") != "implementation-planning": continue relative = timeline_report_record_rel(run) if not relative: continue report = project_root / relative if report.is_file(): return report return None def rerun_readiness(project_root: Path, runs: list[dict]) -> dict | None: """What the next clarification re-run needs, assembled from disk. Every field is derivable before the run starts, and each one used to live somewhere else: the flag value in the lead prompt (read only *after* the run begins), the answered ids in sidecars, the re-verification mode nowhere at all until `incremental-scope --preview`. Scattering them is why the user had to ask for each one instead of being told. ``None`` when nothing is waiting to be carried — no planning report, or no answered clarification beside it. """ report = _latest_planning_report(runs, project_root) if report is None: return None answered = sorted(sidecar_answers(report)) if not answered: return None return { "sourceReport": project_rel(report, project_root), "answeredClarifications": answered, "answeredClarificationsCsv": ",".join(answered), "sidecars": [ project_rel(sidecar, project_root) for sidecar in user_response_sidecars(report) ], "reverifyPreview": preview_link_availability_for_report( report, set(answered) ), "structuralChangeNote": _STRUCTURAL_CHANGE_NOTE, } def assemble_recap(task_root: Path, project_root: Path) -> dict: runs = _load_timeline(task_root) transitions = [] prev_phase = "" latest_states: dict = {} for idx, run in enumerate(runs): snap = run.get("workflowSnapshot", {}) if isinstance(run, dict) else {} cur = snap.get("currentPhase", "") transitions.append({ "index": idx, "runTimestamp": run.get("runTimestamp", ""), "taskType": run.get("taskType", ""), "status": run.get("status", ""), "fromPhase": prev_phase, "toPhase": cur, "lastCompletedPhase": snap.get("lastCompletedPhase", ""), "nextRecommendedPhase": next_phase.promote( snap.get("nextRecommendedPhase") ), "reportRecordPath": timeline_report_record_rel(run), }) prev_phase = cur # latestPhaseStates 는 항상 가장 최근 run 의 상태를 반영해야 한다. # 빈/누락 snapshot 일 때 이전 run 값으로 폴백하면 오래된 phase map 을 # 최신 상태인 양 보고하게 된다. latest_states = snap.get("phaseStates", {}) or {} return { "taskKey": read_task_key(task_root), "runCount": len(runs), "transitions": transitions, "latestPhaseStates": latest_states, # `null` when nothing is waiting to be carried — the key is always # present so a consumer can tell "no re-run pending" from "this recap # predates the block". "rerunReadiness": rerun_readiness(project_root, runs), } def append_recap_entry(task_root, project_root, *, kind, mode, question, answer_summary, citations, now) -> dict: recap_dir = task_root / "recap" recap_dir.mkdir(parents=True, exist_ok=True) log_path = recap_dir / "recap-log.jsonl" entry = { "ts": now.isoformat(), "kind": kind, "mode": mode, "question": question, "answerSummary": answer_summary, "citations": list(citations), } # 같은 task 에 대한 동시 record 호출이 한 JSONL 줄을 쪼개지 않도록 # open+write+close(flush) 전체를 dir_flock 으로 직렬화한다 (consumers.jsonl # 과 같은 append-only idiom — 데이터 fd 가 아닌 별도 .lock 파일을 잠가야 # flush 가 락 안에서 끝난다). with dir_flock(recap_dir, ".recap-log.lock"): append_jsonl(log_path, entry, ensure_ascii=False, compact=False) return {"logPath": project_rel(log_path, project_root), "entry": entry} def _note_path(notes_dir: Path, slug: str, date: str) -> Path: # 같은 날 같은 slug 로 노트를 여러 개 남기면 덮어쓰지 않도록 -2, -3 을 붙인다. base = notes_dir / f"{slug}-{date}.md" if not base.exists(): return base seq = 2 while (candidate := notes_dir / f"{slug}-{date}-{seq}.md").exists(): seq += 1 return candidate def _note_frontmatter(*, task_key, kind, created_at, purpose, scope_note) -> str: return ( "---\n" f"task-key: {task_key}\n" f"kind: {kind}\n" "author: agent\n" f"created-at: {created_at}\n" f"purpose: {purpose}\n" f"scope-note: {scope_note}\n" "---\n\n" ) def write_note(task_root, project_root, *, kind, slug, purpose, scope_note, body, now) -> dict: notes_dir = task_root / "notes" notes_dir.mkdir(parents=True, exist_ok=True) created_at = now.date().isoformat() slug_seg = slugify_task_segment(slug) if not slug_seg: raise ValueError("slug must contain at least one alphanumeric character") path = _note_path(notes_dir, slug_seg, created_at) task_key = read_task_key(task_root) frontmatter = _note_frontmatter( task_key=task_key, kind=kind, created_at=created_at, purpose=purpose, scope_note=scope_note) path.write_text(frontmatter + body.rstrip("\n") + "\n", encoding="utf-8") note_rel = project_rel(path, project_root) return { "notePath": note_rel, "taskKey": task_key, "kind": kind, "createdAt": created_at, # 노트는 inert — 다음 run 이 자동으로 읽지 않는다. 이 인자를 그대로 # `okstra run ...` 에 붙여야만 instruction-set 에 주입된다. "clarificationResponseArg": f"--clarification-response {note_rel}", } def _add_common(sp) -> None: # skill 이 쓰는 ` --project-root ` 형태를 받으려면 # 공통 옵션을 부모가 아닌 각 서브파서에 등록해야 한다. sp.add_argument("--project-root", default="") sp.add_argument("--cwd", default=".") def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="okstra recap") sub = parser.add_subparsers(dest="command", required=True) p_asm = sub.add_parser("assemble", help="assemble run-to-run phase transitions") p_asm.add_argument("target", help="task root path or task-key") _add_common(p_asm) p_rec = sub.add_parser("record", help="append a recap Q&A/summary log entry") p_rec.add_argument("target", help="task root path or task-key") _add_common(p_rec) p_rec.add_argument("--kind", choices=["summary", "qa"], required=True) p_rec.add_argument("--mode", choices=["artifact", "code"], required=True) p_rec.add_argument("--question", default="") p_rec.add_argument("--answer", required=True) p_rec.add_argument("--citation", action="append", default=[]) p_note = sub.add_parser( "note", help="write an agent-authored note into /notes/") p_note.add_argument("target", help="task root path or task-key") _add_common(p_note) p_note.add_argument("--kind", choices=NOTE_KINDS, required=True) p_note.add_argument("--slug", required=True, help="short topic slug for the filename") p_note.add_argument("--purpose", required=True, help="one line — what it is and which run/decision it feeds") p_note.add_argument("--scope-note", required=True, dest="scope_note", help="one line — what it is NOT (e.g. not a user decision)") body_src = p_note.add_mutually_exclusive_group(required=True) body_src.add_argument("--body", help="inline markdown body") body_src.add_argument("--body-file", dest="body_file", help="path to a file holding the markdown body") args = parser.parse_args(argv) task_root, project_root = resolve_task_root( args.target, args.project_root, args.cwd) if args.command == "assemble": result = assemble_recap(task_root, project_root) elif args.command == "note": body = (Path(args.body_file).read_text(encoding="utf-8") if args.body_file else args.body) result = write_note( task_root, project_root, kind=args.kind, slug=args.slug, purpose=args.purpose, scope_note=args.scope_note, body=body, now=dt.datetime.now(dt.timezone.utc)) else: result = append_recap_entry( task_root, project_root, kind=args.kind, mode=args.mode, question=args.question, answer_summary=args.answer, citations=args.citation, now=dt.datetime.now(dt.timezone.utc)) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))