"""Read-side recap for a task or a task-group: assemble run-to-run phase transitions (task) or the group's start order and each task's latest conclusion (task-group), append a recap Q&A log, and write agent-authored notes. Writes only under /recap/recap-log.jsonl, /notes/, and .okstra/tasks//.recap/recap-log.jsonl; never touches task-manifest / catalog / timeline / group-context.md. """ from __future__ import annotations import argparse import datetime as dt import json import sys from pathlib import Path from typing import Any, Mapping from okstra_ctl import group_context, 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 ( ResolverError, find_task_root, list_project_tasks, read_task_key, read_task_manifest, resolve_project_root, tasks_root, ) 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 from okstra_ctl.timeline_runs import current_run_facts NOTE_KINDS = ("verification-evidence", "decision-draft", "analysis-note") GROUP_RECAP_DIRNAME = ".recap" 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: manifest = read_task_manifest(task_root) or {} # timeline 항목의 status / workflowSnapshot / reportRecordPath 는 준비 시점 # 값이다. 각 run 의 종료 상태는 그 run 의 run-manifest 에서 덮어쓴다. runs = [ current_run_facts(project_root, run) if isinstance(run, dict) else run for run in _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), "workStatus": manifest.get("workStatus", ""), "latestWorkRecordPath": manifest.get("latestWorkRecordPath", ""), "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 group_recap_dir(project_root: Path, task_group: str) -> Path: """그룹 recap 로그의 자리 — task 디렉터리들과 형제인 점 디렉터리. `.okstra/tasks//` 아래를 훑는 스캐너(backfill 의 run-manifest 색인, container 의 배포 상태 조회)는 task 디렉터리를 `runs/` · `container/` 의 존재로 판정하므로 이 디렉터리를 task 로 오인하지 않는다. """ return tasks_root(project_root) / slugify_task_segment(task_group) / GROUP_RECAP_DIRNAME def _group_catalog(project_root: Path, task_group: str) -> dict[str, dict]: """카탈로그의 그룹 task 를 task 디렉터리 이름으로 색인한다.""" wanted = slugify_task_segment(task_group) out: dict[str, dict] = {} for entry in list_project_tasks(project_root): if slugify_task_segment(str(entry.get("taskGroup") or "")) != wanted: continue task_id = slugify_task_segment(str(entry.get("taskId") or "")) if task_id: out[task_id] = entry return out def _memory_block(entry: group_context.MemoryEntry) -> dict[str, Any]: return { "date": entry.date, "source": entry.source, "taskType": entry.task_type, "seq": entry.seq, "nextPhase": entry.next_phase, "headline": entry.headline, "decisions": list(entry.decisions), "watchOut": list(entry.watch_out), "followUps": list(entry.follow_ups), "record": entry.record, } def _queue_row(row: group_context.QueueRow) -> dict[str, Any]: """큐 행을 카멜케이스 키로. `waits_for` 는 튜플이라 JSON 밖에서도 리스트로 맞춘다.""" return { "taskId": row.task_id, "briefId": row.brief_id, "ticketId": row.ticket_id, "brief": row.brief, "status": row.status, "progress": row.progress, "waitsFor": list(row.waits_for), } def assemble_group_recap(project_root: Path, task_group: str) -> dict: """그룹의 시작 순서와 task 별 최신 결론을 세 출처에서 합친다. 브리프 디렉터리가 목록과 순서를, 카탈로그(task-manifest)가 실제 phase·상태를, 그룹 문서의 Task Memory 영역이 결론(headline·decisions·watch out·follow-ups)을 준다. 어느 하나만으로는 틀린다 — 카탈로그는 미시작 브리프를 모르고, 메모리는 2026-09-07 이전에 끝난 task 를 모른다. 숫자 집계(CPU·에러)는 rollup 의 것이라 여기서 세지 않는다. """ context_file = group_context.group_context_file(project_root, task_group) text = context_file.read_text(encoding="utf-8") if context_file.is_file() else "" _, region, _ = group_context.split_memory_region(text) entries = group_context.parse_memory_entries(region) queue = group_context.group_queue(project_root, task_group, entries) following = group_context.next_in_group(queue) by_memory = {entry.task_id: entry for entry in entries} by_catalog = _group_catalog(project_root, task_group) order: list[str] = [row.task_id for row in queue] order.extend(task_id for task_id in by_catalog if task_id not in order) order.extend(task_id for task_id in by_memory if task_id not in order) tasks: list[dict[str, Any]] = [] for task_id in order: catalog = by_catalog.get(task_id) memory = by_memory.get(task_id) if catalog is None and memory is None: continue # 브리프만 있고 시작하지 않은 task — 큐에만 실린다 task_root = find_task_root(project_root, str(catalog.get("taskKey") or "")) if catalog else None pointer = next_phase.promote(catalog.get("nextRecommendedPhase")) if catalog else next_phase.make() tasks.append({ "taskId": task_id, "taskKey": str(catalog.get("taskKey") or "") if catalog else "", "currentPhase": str(catalog.get("currentPhase") or "") if catalog else "", "currentPhaseState": str(catalog.get("currentPhaseState") or "") if catalog else "", "latestRunStatus": str(catalog.get("latestRunStatus") or "") if catalog else "", "workStatus": str(catalog.get("workStatus") or "") if catalog else "", "latestWorkRecordPath": str(catalog.get("latestWorkRecordPath") or "") if catalog else "", "nextRecommendedPhase": pointer, "runCount": len(_load_timeline(task_root)) if task_root else 0, "reportPath": ( str(catalog.get("latestReportRecordPath") or "") if catalog else "" ) or (memory.record if memory else ""), "memory": _memory_block(memory) if memory else None, }) return { "taskGroup": task_group, "groupContextPath": project_rel(context_file, project_root) if context_file.is_file() else "", "hasHumanSections": group_context.has_human_sections(text) if text else False, "briefCount": len(queue), "taskCount": len(tasks), "queue": [_queue_row(row) for row in queue], "nextInGroup": _queue_row(following) if following else None, "tasks": tasks, } def append_recap_entry(task_root, project_root, *, kind, mode, question, answer_summary, citations, now) -> dict: return _append_entry( task_root / "recap", project_root, kind=kind, mode=mode, question=question, answer_summary=answer_summary, citations=citations, now=now) def append_group_recap_entry(project_root, task_group, *, kind, mode, question, answer_summary, citations, now) -> dict: return _append_entry( group_recap_dir(project_root, task_group), project_root, kind=kind, mode=mode, question=question, answer_summary=answer_summary, citations=citations, now=now) def _append_entry(recap_dir: Path, project_root, *, kind, mode, question, answer_summary, citations, now) -> dict: 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=".") _CLI_EPILOG = r"""Usage: okstra recap assemble ( | --task-group ) [--project-root ] [--cwd ] okstra recap record ( | --task-group ) --kind \ --mode [--question ] --answer [--citation ...] okstra recap note \ --kind --slug \ --purpose --scope-note (--body | --body-file ) `assemble` is read-only. For a task it prints a JSON summary of phase transitions across runs; with --task-group it prints the group's start order (briefs in ordinal order, each done / in progress / not started from the catalog) and every recorded task's latest conclusion from the group document's Task Memory. `record` appends one line to /recap/recap-log.jsonl, or with --task-group to .okstra/tasks//.recap/recap-log.jsonl; it never mutates other artifacts. `note` writes an agent-authored markdown note to /notes/ and prints its path plus the `--clarification-response` argument for feeding it into a later run. """ _CLI_DESCRIPTION = "Assemble a task's or a task-group's recap, append a recap log entry, or write an agent note." def _resolve_group_scope(args: argparse.Namespace) -> Path: """`--task-group` 호출의 프로젝트 루트. 그룹은 task-key 해석을 거치지 않는다.""" try: return Path(resolve_project_root(explicit_root=args.project_root, cwd=args.cwd)).resolve() except ResolverError as exc: raise SystemExit(f"project root resolution failed: {exc}") from exc def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description=_CLI_DESCRIPTION, epilog=_CLI_EPILOG, formatter_class=argparse.RawDescriptionHelpFormatter, prog="okstra recap") sub = parser.add_subparsers(dest="command", required=True) p_asm = sub.add_parser( "assemble", help="assemble a task's phase transitions or a task-group's start order") p_asm.add_argument("target", nargs="?", default="", help="task root path or task-key") p_asm.add_argument("--task-group", default="", dest="task_group", help="recap the whole task-group instead of one task") _add_common(p_asm) p_rec = sub.add_parser("record", help="append a recap Q&A/summary log entry") p_rec.add_argument("target", nargs="?", default="", help="task root path or task-key") p_rec.add_argument("--task-group", default="", dest="task_group", help="log under the task-group's .recap/ instead of one task") _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) group_scoped = args.command in ("assemble", "record") and bool(args.task_group) if args.command in ("assemble", "record") and bool(args.target) == group_scoped: parser.error(f"{args.command} takes exactly one of or --task-group") if group_scoped: project_root = _resolve_group_scope(args) if args.command == "assemble": result = assemble_group_recap(project_root, args.task_group) else: result = append_group_recap_entry( project_root, args.task_group, 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 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:]))