"""task-group 맥락 문서 — 그룹의 모든 task 가 공유하는 배경을 워커 입력에 싣는다. `.okstra/briefs//group-context.md` 한 파일이 그룹이 존재하는 이유, 성과 척도, 그룹 전체 금지선, 티켓 간 관계를 담는다. 브리프는 티켓 하나의 범위를 말하므로 그룹 단위의 "왜" 는 어느 브리프에도 없다(2026-09-04 실측, fontsninja `cache` 그룹: 브리프 15개 중 가용성 사고를 말한 것이 0개였고 option-selection 이 분모를 페이지당 요청 수로 바꿔 판단했다). 파일은 두 소유자가 나눠 쓴다. 위의 네 절은 사용자가 쓰고, 파일 끝의 마커 영역 (`MEMORY_BEGIN`…`MEMORY_END`, `## Task Memory`)은 okstra 가 report-finalize 마다 다시 쓴다 — 그룹의 각 task 가 마지막 run 에서 낸 결론(headline·decisions· follow-up·레코드 경로)이다. 형제 task 의 run 은 이 영역을 패킷의 `## Task-Group Memory` 로 받는다(2026-09-07: cache 그룹 15개 티켓 중 첫 티켓의 run 8개 결론이 나머지 14개에 닿는 통로가 없었다). 네 소비자가 이 모듈 하나를 쓴다. - `okstra group-context init` — 템플릿에서 뼈대를 쓴다. okstra 가 메모리만 써 둔 파일이면 그 위에 사람 절 뼈대를 끼워 넣고, 사람 절이 있는 파일은 덮지 않는다. - `validators/validate-brief.py` — 프론트매터 `type: group-context` 를 만나면 `validate_group_context` 로 검사한다. 사람 절이 하나도 없는 메모리 전용 파일은 네 절 규칙을 적용하지 않는다. - prepare(`run.py`) — 파일이 있으면 같은 검사를 통과해야 진행하고, 통과하면 `instruction-set/task-group-context.md` 로 복사해 analysis packet 에 싣는다. 패킷은 `split_memory_region` 으로 사람 절과 메모리를 갈라 따로 배치한다. - report-finalize `record-group-memory` — `record_task_memory` 로 이 run 의 항목을 영역에 쓴다. 파일이 없으면 메모리 전용 파일을 만든다. """ from __future__ import annotations import argparse import datetime as dt import re import sys from dataclasses import dataclass, replace from pathlib import Path from typing import Any, Mapping from okstra_project import list_project_tasks from okstra_project.phase_pointer import STATUS_TERMINAL from .brief_frontmatter import read_brief_frontmatter from .ids import slugify_task_segment from .paths import find_asset_root from .run_context import dir_flock GROUP_CONTEXT_FILENAME = "group-context.md" GROUP_CONTEXT_TYPE = "group-context" GROUP_CONTEXT_GENERATOR = "okstra-brief-gen" # 파일이 없을 때 finalize 가 메모리 전용 파일을 만들며 적는 생성자. MEMORY_GENERATOR = "okstra-report-finalize" GENERATORS = (GROUP_CONTEXT_GENERATOR, MEMORY_GENERATOR) INSTRUCTION_SET_FILENAME = "task-group-context.md" REQUIRED_FRONTMATTER_KEYS = ("type", "task-group", "created", "generator") SECTIONS = ( "Why This Group Exists", "Definition of Better", "Group-Wide Constraints", "Ticket Relations", ) # 나머지 두 절은 `_(none)_` 이 허용된다 — 금지선이나 관계가 없는 그룹도 있다. MUST_BE_FILLED = SECTIONS[:2] TEMPLATE_RELATIVE = ("templates", "reports", "group-context.template.md") _HTML_COMMENT_RE = re.compile(r"", re.DOTALL) _NONE_MARKERS = {"_(none)_", ""} # --- okstra 소유 영역 ---------------------------------------------------------- MEMORY_BEGIN = "" MEMORY_END = "" MEMORY_NOTE = ( "" ) MEMORY_HEADING = "## Task Memory" MEMORY_EMPTY = "_(no runs recorded yet)_" # 실측(2026-09-07 cache 그룹 dev-10626): decisions 한 줄이 250~400자, 캡 없이 한 task 가 # 3.5KB 였다. 15개 task 그룹이 패킷에 실리므로 task 당 1.5KB 안팎으로 잡는다. HEADLINE_MAX = 240 ITEM_MAX = 160 DECISIONS_MAX = 3 FOLLOW_UPS_MAX = 3 WATCH_OUT_MAX = 2 LOCK_FILENAME = ".group-context.lock" _FOLLOW_UP_SEP = "; " _LATEST_RE = re.compile(r"^- latest: (?P\S+) #(?P\S+) · (?P\S+) · next: (?P.*)$") @dataclass(frozen=True) class MemoryEntry: """한 task 의 마지막 run 결론. `decisions` 만 이전 항목과 합쳐 누적한다.""" task_id: str task_type: str seq: str date: str next_phase: str headline: str decisions: tuple[str, ...] follow_ups: tuple[str, ...] record: str watch_out: tuple[str, ...] = () source: str = "run" def memory_entry_from_record( data: Mapping[str, Any], *, task_id: str, task_type: str, seq: str, record: str, next_phase: str, today: dt.date, ) -> MemoryEntry: """final-report 레코드에서 항목을 뽑는다. 없는 필드는 빈 값이지 실패가 아니다.""" summary = data.get("humanSummary") summary = summary if isinstance(summary, Mapping) else {} header = data.get("header") header = header if isinstance(header, Mapping) else {} created = header.get("createdAt") date = created[:10] if isinstance(created, str) and len(created) >= 10 else today.isoformat() follow_ups = tuple( _clip(_one_line(row["title"]), ITEM_MAX) for row in (data.get("followUpTasks") or []) if isinstance(row, Mapping) and row.get("origin") != "phase-continuation" and isinstance(row.get("title"), str) and row["title"].strip() ) return MemoryEntry( task_id=slugify_task_segment(task_id), task_type=task_type, seq=seq, date=date, next_phase=next_phase, headline=_clip(_one_line(summary.get("headline")), HEADLINE_MAX), decisions=tuple( _clip(_one_line(item), ITEM_MAX) for item in (summary.get("decisions") or []) if isinstance(item, str) and item.strip() ), follow_ups=follow_ups, record=record, watch_out=tuple( _clip(_one_line(item), ITEM_MAX) for item in (summary.get("blockers") or []) if isinstance(item, str) and item.strip() ), ) def _one_line(value: Any) -> str: return " ".join(str(value).split()) if isinstance(value, str) else "" def _clip(text: str, limit: int) -> str: return text if len(text) <= limit else text[: limit - 1].rstrip() + "…" def split_memory_region(text: str) -> tuple[str, str, str]: """(마커 앞, 마커 사이, 마커 뒤). 영역이 없으면 사이와 뒤는 빈 문자열. begin 만 있으면 파일 끝까지를 영역으로 본다. end 만 있으면 영역이 없는 것으로 본다 — 어느 쪽이든 다음 쓰기가 균형 잡힌 영역으로 다시 쓴다. """ begin = text.find(MEMORY_BEGIN) if begin < 0: return text, "", "" inner_start = begin + len(MEMORY_BEGIN) end = text.find(MEMORY_END, inner_start) if end < 0: return text[:begin], text[inner_start:], "" return text[:begin], text[inner_start:end], text[end + len(MEMORY_END):] def parse_memory_entries(region: str) -> list[MemoryEntry]: """영역 본문 → 항목. `### ` 가 항목을 연다. 모르는 줄은 버린다.""" entries: list[MemoryEntry] = [] current: dict[str, Any] | None = None in_decisions = in_watch_out = False for raw in region.splitlines(): line = raw.rstrip() if line.startswith("### "): if current is not None: entries.append(_entry_from_fields(current)) current = {"task_id": line[4:].strip(), "decisions": [], "watch_out": []} in_decisions = in_watch_out = False continue if current is None: continue if line.startswith(" - ") and (in_decisions or in_watch_out): current["decisions" if in_decisions else "watch_out"].append(line[4:].strip()) continue in_decisions = in_watch_out = False match = _LATEST_RE.match(line) if match: current.update( task_type=match.group("type"), seq=match.group("seq"), date=match.group("date"), next_phase=match.group("next").strip(), ) elif line.startswith("- direct: "): current.update(source="direct", date=line[len("- direct: "):].strip()) elif line.startswith("- headline: "): current["headline"] = line[len("- headline: "):].strip() elif line == "- decisions:": in_decisions = True elif line.startswith("- open follow-ups: "): value = line[len("- open follow-ups: "):].strip() current["follow_ups"] = [] if value in _NONE_MARKERS else value.split(_FOLLOW_UP_SEP) elif line == "- watch out:": in_watch_out = True elif line.startswith("- record: "): current["record"] = line[len("- record: "):].strip().strip("`") if current is not None: entries.append(_entry_from_fields(current)) return entries def _entry_from_fields(fields: Mapping[str, Any]) -> MemoryEntry: record = fields.get("record", "") return MemoryEntry( task_id=fields.get("task_id", ""), task_type=fields.get("task_type", ""), seq=fields.get("seq", ""), date=fields.get("date", ""), next_phase=fields.get("next_phase", ""), headline=fields.get("headline", ""), decisions=tuple(fields.get("decisions", [])), follow_ups=tuple(fields.get("follow_ups", [])), record="" if record in _NONE_MARKERS else record, watch_out=tuple(fields.get("watch_out", [])), source=fields.get("source", "run"), ) def render_memory_entries(entries: list[MemoryEntry]) -> str: """항목들 → `### ` 블록. 빈 목록은 `MEMORY_EMPTY` 한 줄.""" if not entries: return MEMORY_EMPTY + "\n" blocks: list[str] = [] for entry in entries: lines = [ f"### {entry.task_id}", (f"- direct: {entry.date}" if entry.source == "direct" else f"- latest: {entry.task_type} #{entry.seq} · {entry.date} · next: {entry.next_phase or '_(none)_'}"), f"- headline: {entry.headline or '_(none)_'}", ] if entry.source == "direct": lines.append("- Verification: direct work; no cross-verification performed for this record.") if entry.decisions: lines.append("- decisions:") lines.extend(f" - {item}" for item in entry.decisions) if entry.watch_out: lines.append("- watch out:") lines.extend(f" - {item}" for item in entry.watch_out) lines.append( "- open follow-ups: " + (_FOLLOW_UP_SEP.join(entry.follow_ups) if entry.follow_ups else "_(none)_") ) lines.append(f"- record: `{entry.record}`" if entry.record else "- record: _(none)_") blocks.append("\n".join(lines) + "\n") return "\n".join(blocks) def render_memory_region(entries: list[MemoryEntry], queue: list[QueueRow] | None = None) -> str: parts = [MEMORY_BEGIN, MEMORY_NOTE, MEMORY_HEADING, ""] queue_text = render_queue(queue or []) if queue_text: parts.extend([queue_text.rstrip("\n"), ""]) parts.extend([render_memory_entries(entries).rstrip("\n"), MEMORY_END]) return "\n".join(parts) + "\n" def merge_memory_entry(entries: list[MemoryEntry], entry: MemoryEntry) -> list[MemoryEntry]: """같은 task 의 항목을 갈아 끼우고 decisions 는 새 것 우선으로 합쳐 상한까지 남긴다.""" previous = next((e for e in entries if e.task_id == entry.task_id), None) decisions: list[str] = [] for item in (*entry.decisions, *(previous.decisions if previous else ())): if item and item not in decisions: decisions.append(item) merged = replace( entry, decisions=tuple(decisions[:DECISIONS_MAX]), follow_ups=entry.follow_ups[:FOLLOW_UPS_MAX], watch_out=entry.watch_out[:WATCH_OUT_MAX], ) rest = [e for e in entries if e.task_id != entry.task_id] return sorted([*rest, merged], key=lambda e: (e.date, e.task_id), reverse=True) def memory_only_document(task_group: str, created: str) -> str: """사람 절 없이 프론트매터·제목·빈 영역만 있는 파일. finalize 가 파일이 없을 때 만든다.""" slug = slugify_task_segment(task_group) return ( "---\n" f"type: {GROUP_CONTEXT_TYPE}\n" f"task-group: {slug}\n" f"created: {created}\n" f"generator: {MEMORY_GENERATOR}\n" "---\n" "\n" f"# Task-Group Context: {slug}\n" "\n" + render_memory_region([]) ) def record_task_memory( project_root: Path, task_group: str, entry: MemoryEntry, *, today: dt.date ) -> tuple[Path, list[QueueRow]]: """이 run 의 항목을 그룹 문서의 okstra 영역에 쓰고 (경로, 시작 순서 큐)를 돌려준다. 파일이 없으면 만든다. 사람 절은 바이트 그대로 둔다. 같은 그룹의 finalize 가 동시에 돌 수 있으므로 디렉터리 flock 아래에서 읽고-바꾸고-쓴다. """ target = group_context_file(project_root, task_group) target.parent.mkdir(parents=True, exist_ok=True) with dir_flock(target.parent, LOCK_FILENAME): text = ( target.read_text(encoding="utf-8") if target.is_file() else memory_only_document(task_group, today.isoformat()) ) before, region, after = split_memory_region(text) entries = merge_memory_entry(parse_memory_entries(region), entry) queue = group_queue(project_root, task_group, entries) if not before.endswith("\n"): before += "\n" if before.strip() and not before.endswith("\n\n"): before += "\n" target.write_text(before + render_memory_region(entries, queue) + after, encoding="utf-8") return target, queue def refresh_group_queue(project_root: Path, task_group: str) -> Path | None: """그룹 문서의 시작 순서만 지금 값으로 다시 그린다. 문서가 없으면 아무것도 안 한다. 기록된 항목과 사람 절은 바이트 그대로 두고 큐만 다시 계산한다. 이 투영을 쓰는 곳은 원래 report-finalize 의 `record-group-memory` 하나뿐이라, 리포트 없이 끝난 run 이나 사람이 손으로 바꾼 상태는 다음 finalize 까지 문서에 닿지 않았다(2026-09-10 실측, cache 그룹 dev-10635: `workStatus` 를 done 으로 적었는데 큐는 `[in progress] — requirements-discovery (blocked)` 그대로). 같은 그룹의 finalize 와 겹칠 수 있으므로 같은 flock 아래에서 읽고-쓴다. """ target = group_context_file(project_root, task_group) if not target.is_file(): return None with dir_flock(target.parent, LOCK_FILENAME): text = target.read_text(encoding="utf-8") before, region, after = split_memory_region(text) if not region: return None entries = parse_memory_entries(region) queue = group_queue(project_root, task_group, entries) updated = before + render_memory_region(entries, queue) + after if updated != text: target.write_text(updated, encoding="utf-8") return target # --- 시작 순서 ---------------------------------------------------------------- # # 그룹의 task 는 순서가 있을 수 있다. 순서는 brief-gen 이 매긴 브리프 순번 # (`DEV-10627-2-…` 의 `2`)이다 — 사용자가 분할 때 정한 순서다. 브리프의 Related # Task Graph 에 적힌 `depends-on` / `blocks` / `blocked-by` 간선은 재정렬하지 않고 # `waits for` 로 붙여 보이기만 한다: 실측(2026-09-07 cache 그룹) 에서 그래프는 # 15번 가드 티켓이 1번을 `blocks` 한다고 적었지만 사용자는 1번을 먼저 끝냈다. # 큐는 영역 머리에 실리고, 이 task 가 끝나는 closeout 은 큐의 다음 task 를 이름한다. # `okstra set-work-status` 가 task-manifest 에 적는 값 중 큐 상태를 덮는 것. # 나머지(`todo` / `in-progress` / `blocked`)는 큐의 세 상태로 옮길 때 파생값보다 # 나은 정보가 없어 덮지 않는다. WORK_STATUS_DONE = "done" QUEUE_DONE = "done" QUEUE_IN_PROGRESS = "in progress" QUEUE_NOT_STARTED = "not started" _ORDINAL_RE = re.compile(r"^(?P[A-Za-z]+-\d+)-(?P\d+)-") _WAIT_RELATIONS = {"depends-on", "blocked-by"} _BLOCK_RELATIONS = {"blocks"} _QUEUE_INTRO = ( "Start order (brief ordinal; `waits for` quotes `depends-on` / `blocks` edges " "from the briefs' Related Task Graph whose named ticket is not done):" ) _QUEUE_LINE_RE = re.compile(r"^\d+\. \[(?P[^\]]*)\] (?P\S+)") @dataclass(frozen=True) class QueueRow: task_id: str # slugified brief-id — the task directory name brief_id: str ticket_id: str brief: str # project-relative brief path status: str # QUEUE_* progress: str # " #" for a recorded task, else "" waits_for: tuple[str, ...] = () # ticket ids the graph says come first and are not done # 같은 `brief-id` 를 가진 다른 브리프 파일들. 하나의 task 를 두 파일이 # 주장하는 상태이므로 큐는 정본 한 줄만 싣고 나머지를 여기 이름한다. duplicate_briefs: tuple[str, ...] = () def ticket_id_from_brief_id(brief_id: str) -> str: """순번이 붙은 브리프 식별자에서 이슈 식별자를 복원한다.""" match = _ORDINAL_RE.match(brief_id) return match.group("ticket") if match else brief_id def group_briefs(project_root: Path, task_group: str) -> list[dict[str, Any]]: """그룹 디렉터리의 브리프(프론트매터 `type: brief`)와 각 브리프의 대기 간선.""" group_dir = group_context_file(project_root, task_group).parent if not group_dir.is_dir(): return [] briefs: list[dict[str, Any]] = [] for path in sorted(group_dir.rglob("*.md")): if is_group_context_file(path) or not path.is_file(): continue frontmatter = read_brief_frontmatter(path) if frontmatter.get("type") != "brief": continue brief_id = frontmatter.get("brief-id") or path.stem match = _ORDINAL_RE.match(brief_id) briefs.append({ "brief_id": brief_id, "ticket_id": frontmatter.get("ticket-id") or ticket_id_from_brief_id(brief_id), "ordinal": int(match.group("ordinal")) if match else None, "brief": path.relative_to(project_root).as_posix(), "waits_for": _wait_edges(path.read_text(encoding="utf-8")), }) return briefs def _wait_edges(text: str) -> dict[str, set[str]]: """Related Task Graph 표 → {ticket: 먼저 끝나야 하는 ticket 들}.""" waits: dict[str, set[str]] = {} for line in _section_bodies(text).get("Related Task Graph", "").splitlines(): cells = [cell.strip() for cell in line.strip().strip("|").split("|")] if len(cells) < 3: continue source, relation, target = cells[0], cells[1], cells[2] if relation in _WAIT_RELATIONS: waits.setdefault(source, set()).add(target) elif relation in _BLOCK_RELATIONS: waits.setdefault(target, set()).add(source) return waits def catalog_progress(project_root: Path, task_group: str) -> dict[str, tuple[str, str]]: """카탈로그에 있는 그룹 task → (큐 상태, 진행 표기). task-manifest 가 정본이다. 메모리 항목은 2026-09-07 이후의 finalize 만 쓰므로 그 전에 끝난 task 는 항목이 없다(실측: cache 그룹 dev-10626 — run 13개, release-handoff 완료인데 큐는 `[not started]`). 카탈로그 항목은 `list_project_tasks` 가 task-manifest 의 phase·상태·다음 phase 포인터로 덮어 돌려주므로 여기서 다시 읽지 않는다. """ wanted = slugify_task_segment(task_group) out: dict[str, tuple[str, str]] = {} 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 not task_id: continue pointer = entry.get("nextRecommendedPhase") pointer = pointer if isinstance(pointer, Mapping) else {} phase = str(entry.get("currentPhase") or entry.get("taskType") or "") state = str(entry.get("currentPhaseState") or entry.get("latestRunStatus") or "") progress = f"{phase} ({state})" if phase and state else phase status = QUEUE_DONE if pointer.get("status") == STATUS_TERMINAL else QUEUE_IN_PROGRESS # 사용자가 `okstra set-work-status done` 으로 끝났다고 선언한 task # 는 끝난 것이다. 파생 포인터만 보면 리포트 없이 끝난 run 이 영원히 # `[in progress]` 로 남는다 — 실측(2026-09-10, fontsninja-v3-site cache # 그룹 dev-10635): 사람이 AWS 콘솔에서 작업을 마치고 `done` 을 적었는데, # 그 run 이 리포트 조립 실패로 끝나 포인터가 `blocked` 에 멈춰 있었다. # 파생 표기는 지우지 않는다 — 왜 done 인지가 그 차이에 남는다. if str(entry.get("workStatus") or "") == WORK_STATUS_DONE: status = QUEUE_DONE progress = f"{progress} · marked done" if progress else "marked done" elif entry.get("workStatus") == "todo": status = QUEUE_NOT_STARTED elif entry.get("workStatus") in ("in-progress", "blocked"): status = QUEUE_IN_PROGRESS if entry.get("latestWorkRecordPath"): progress = f"{progress} · direct work recorded" if progress else "direct work recorded" out[task_id] = (status, progress) return out def group_queue( project_root: Path, task_group: str, entries: list[MemoryEntry] ) -> list[QueueRow]: """브리프를 순번 순서로 늘어놓고 카탈로그, 그 다음 기록된 항목으로 상태를 매긴다. 순번이 없는 브리프는 뒤에 id 순이다. 그래프 간선은 순서를 바꾸지 않고 아직 끝나지 않은 선행 ticket 을 `waits_for` 로 붙인다. 카탈로그에 있는 task 는 task-manifest 가 상태를 정하고, 메모리 항목은 카탈로그가 모르는 task 에만 쓰인다(`catalog_progress`). """ briefs = group_briefs(project_root, task_group) if not briefs: return [] by_ticket = {brief["ticket_id"]: brief for brief in briefs} waits: dict[str, set[str]] = {} for brief in briefs: for ticket, before in brief["waits_for"].items(): waits.setdefault(ticket, set()).update((before & by_ticket.keys()) - {ticket}) briefs, duplicates = _fold_duplicate_briefs(briefs) ordered = sorted(briefs, key=lambda b: (b["ordinal"] is None, b["ordinal"] or 0, b["brief_id"])) by_task = {entry.task_id: entry for entry in entries} recorded = catalog_progress(project_root, task_group) def _status(brief: dict[str, Any]) -> tuple[str, str]: task_id = slugify_task_segment(brief["brief_id"]) if task_id in recorded: return recorded[task_id] entry = by_task.get(task_id) if entry is None: return QUEUE_NOT_STARTED, "" done = entry.next_phase.endswith("(terminal)") return (QUEUE_DONE if done else QUEUE_IN_PROGRESS), f"{entry.task_type} #{entry.seq}" status_by_ticket = {brief["ticket_id"]: _status(brief) for brief in ordered} rows: list[QueueRow] = [] for brief in ordered: status, progress = status_by_ticket[brief["ticket_id"]] open_waits = tuple(sorted( ticket for ticket in waits.get(brief["ticket_id"], set()) if status_by_ticket[ticket][0] != QUEUE_DONE )) rows.append(QueueRow( slugify_task_segment(brief["brief_id"]), brief["brief_id"], brief["ticket_id"], brief["brief"], status, progress, open_waits, duplicates.get(brief["brief_id"], ()), )) return rows def _fold_duplicate_briefs( briefs: list[dict[str, Any]], ) -> tuple[list[dict[str, Any]], dict[str, tuple[str, ...]]]: """같은 `brief-id` 를 주장하는 브리프들을 정본 하나로 접는다. `brief-id` 는 task 디렉터리 이름(`task_id`)의 출처라, 두 파일이 같은 id 를 달면 하나의 task 를 둘이 주장하는 상태다. 접지 않으면 큐가 같은 task 를 두 줄로 싣고 번호가 브리프 수보다 커진다 — `next_in_group` 과 closeout 도 같은 task 를 두 번 가리킨다(2026-09-10 실측, fontsninja-v3-site `cache` 그룹: 개정 전 브리프를 `.superseded-<날짜>.md` 로 같은 디렉터리에 남겨 15개 그룹의 큐가 16번까지 갔다). 정본은 파일 이름이 곧 `brief-id` 인 파일이다 — brief-gen 이 쓰는 이름이고, 같은 id 를 단 다른 이름은 사본이다. 그런 파일이 없거나 여럿이면 경로 순서로 첫 번째다. 거절하지 않는 이유는 이 큐를 report-finalize 와 위저드가 읽기 때문이다: 브리프 디렉터리 정리가 안 됐다고 run 발행을 막을 일은 아니다. 대신 남은 사본을 `duplicate_briefs` 로 실어 큐가 그 사실을 말한다. """ by_id: dict[str, list[dict[str, Any]]] = {} for brief in briefs: by_id.setdefault(brief["brief_id"], []).append(brief) kept: list[dict[str, Any]] = [] duplicates: dict[str, tuple[str, ...]] = {} for brief_id, group in by_id.items(): if len(group) == 1: kept.append(group[0]) continue ordered_group = sorted(group, key=lambda b: b["brief"]) canonical = next( (b for b in ordered_group if Path(b["brief"]).stem == brief_id), ordered_group[0], ) kept.append(canonical) duplicates[brief_id] = tuple( b["brief"] for b in ordered_group if b is not canonical ) return kept, duplicates def next_in_group(queue: list[QueueRow]) -> QueueRow | None: """시작 순서에서 아직 시작하지 않은 첫 task.""" return next((row for row in queue if row.status == QUEUE_NOT_STARTED), None) def render_queue(queue: list[QueueRow]) -> str: if not queue: return "" lines = [_QUEUE_INTRO, ""] for number, row in enumerate(queue, start=1): tail = f" — {row.progress}" if row.progress else "" if row.waits_for: tail += f" (waits for {', '.join(row.waits_for)})" lines.append(f"{number}. [{row.status}] {row.task_id}{tail}") # 사본을 이름한다. 이 줄이 없으면 큐가 조용히 한 파일을 무시하고, 그 # 파일을 고친 사람은 자기 편집이 어디로 갔는지 알 길이 없다. for extra in row.duplicate_briefs: lines.append( f" - ignored duplicate of this brief-id: `{extra}` — one task, " "two brief files. Move the copy out of the briefs directory." ) return "\n".join(lines) + "\n" def has_human_sections(text: str) -> bool: """마커 밖에 `## ` 절이 하나라도 있는가. 없으면 메모리 전용 문서다.""" before, _, after = split_memory_region(text) return any(line.startswith("## ") for line in (before + after).splitlines()) def briefs_root(project_root: Path) -> Path: return Path(project_root) / ".okstra" / "briefs" def group_context_file(project_root: Path, task_group: str) -> Path: """그룹 맥락 문서의 정본 경로. 디렉터리는 브리프와 같은 slug 를 쓴다.""" return briefs_root(project_root) / slugify_task_segment(task_group) / GROUP_CONTEXT_FILENAME def is_group_context_file(path: Path) -> bool: return path.name == GROUP_CONTEXT_FILENAME def validate_group_context(path: Path, root: Path) -> list[str]: """결함 목록. 비어 있으면 통과. 결함마다 사람이 고칠 자리를 이름한다.""" raw = path.read_text(encoding="utf-8") errors: list[str] = [] frontmatter = read_brief_frontmatter(path) if not frontmatter: return ["frontmatter: missing or malformed (a `---` block must open line 1)"] missing = [key for key in REQUIRED_FRONTMATTER_KEYS if key not in frontmatter] if missing: errors.append(f"frontmatter missing keys: {missing}") if frontmatter.get("type") != GROUP_CONTEXT_TYPE: errors.append( f"frontmatter type must be {GROUP_CONTEXT_TYPE!r}, got {frontmatter.get('type')!r}" ) if frontmatter.get("generator") not in GENERATORS: errors.append( f"frontmatter generator must be one of {list(GENERATORS)}, " f"got {frontmatter.get('generator')!r}" ) errors.extend(_path_errors(path, root, frontmatter.get("task-group", ""))) # okstra 가 쓴 영역은 검사하지 않는다. 사람 절이 하나라도 있으면 네 절 규칙. # 마커가 HTML 주석이라 영역을 가른 뒤에 주석을 지운다. if has_human_sections(raw): human_text, _, human_tail = split_memory_region(raw) errors.extend(_section_errors(_HTML_COMMENT_RE.sub("", human_text + human_tail))) return errors def _path_errors(path: Path, root: Path, task_group: str) -> list[str]: errors: list[str] = [] if path.name != GROUP_CONTEXT_FILENAME: errors.append(f"file must be named {GROUP_CONTEXT_FILENAME!r}, got {path.name!r}") try: relative = path.resolve().relative_to(Path(root).resolve()) except ValueError: return errors + [f"file is not under the briefs root {root}"] if len(relative.parts) != 2: errors.append( "file must sit directly under its task-group directory " f"(`//{GROUP_CONTEXT_FILENAME}`), got {relative}" ) return errors expected = slugify_task_segment(task_group) if task_group and relative.parts[0] != expected: errors.append( f"task-group directory segment {relative.parts[0]!r} does not match the " f"slugified frontmatter task-group {expected!r}" ) return errors def _section_errors(text: str) -> list[str]: bodies = _section_bodies(text) errors: list[str] = [] for heading in SECTIONS: if heading not in bodies: errors.append(f"missing section `## {heading}`") continue lines = [line.strip() for line in bodies[heading].splitlines() if line.strip()] placeholders = [line for line in lines if _is_template_placeholder(line)] if placeholders: errors.append( f"section `## {heading}` still carries a template placeholder " f"({placeholders[0][:40]}...); fill it or delete the file" ) continue if heading in MUST_BE_FILLED and all(line in _NONE_MARKERS for line in lines): errors.append(f"section `## {heading}` must be filled; `_(none)_` is not accepted here") return errors def _section_bodies(text: str) -> dict[str, str]: """`## ` 제목 → 본문. `## ` 만 절을 닫는다(validate-brief `section_body` 와 같은 규칙).""" bodies: dict[str, str] = {} current: str | None = None buffer: list[str] = [] for line in text.splitlines(): if line.startswith("## "): if current is not None: bodies[current] = "\n".join(buffer) current = line[3:].strip() buffer = [] elif current is not None: buffer.append(line) if current is not None: bodies[current] = "\n".join(buffer) return bodies def _is_template_placeholder(line: str) -> bool: bare = line.lstrip("-").strip() return bare.startswith("<") and bare.endswith(">") def render_skeleton(task_group: str, created: str) -> str: root = find_asset_root(TEMPLATE_RELATIVE) if root is None: raise FileNotFoundError( "group-context template not found in any okstra asset root: " + "/".join(TEMPLATE_RELATIVE) ) template = root.joinpath(*TEMPLATE_RELATIVE).read_text(encoding="utf-8") slug = slugify_task_segment(task_group) return template.replace("", slug).replace("", created) def init_group_context(project_root: Path, task_group: str, *, today: dt.date) -> Path: """뼈대를 쓴다. 사람 절이 있는 파일은 `FileExistsError` — 채운 문서를 덮어쓰지 않는다. okstra 가 메모리만 써 둔 파일이면 그 영역을 그대로 두고 사람 절 뼈대를 앞에 끼운다. """ target = group_context_file(project_root, task_group) skeleton = render_skeleton(task_group, today.isoformat()) if target.exists(): existing = target.read_text(encoding="utf-8") if has_human_sections(existing): raise FileExistsError(str(target)) _, region, after = split_memory_region(existing) human, _, _ = split_memory_region(skeleton) entries = parse_memory_entries(region) target.write_text(human + render_memory_region(entries) + after, encoding="utf-8") return target target.parent.mkdir(parents=True, exist_ok=True) target.write_text(skeleton, encoding="utf-8") return target def _init_command(args: argparse.Namespace) -> int: project_root = Path(args.project_root).resolve() try: target = init_group_context(project_root, args.task_group, today=dt.date.today()) except FileExistsError as exc: print( f"group-context: already exists: {exc}\n" "Edit that file instead; delete it first if you want a fresh skeleton.", file=sys.stderr, ) return 2 print(f"group context skeleton: {target}") print("Fill these sections (one `<...>` placeholder line each):") for heading in SECTIONS: note = "required" if heading in MUST_BE_FILLED else "`_(none)_` allowed" print(f" - ## {heading} ({note})") print( "Validate with: python3 ~/.okstra/lib/validators/validate-brief.py " f"{target} --briefs-root {briefs_root(project_root)}" ) print( "Every run of task-group " f"`{slugify_task_segment(args.task_group)}` refuses to prepare while a placeholder " "line remains; delete the file if the group needs no context." ) print( f"The trailing `{MEMORY_HEADING}` region is okstra's — redrawn after every " "report-finalize, at `okstra set-work-status`, and before each run copies " "this file; leave it alone." ) return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="okstra group-context", description=( "Create the task-group context skeleton beside the group's briefs " "(`.okstra/briefs//group-context.md`)." ), ) sub = parser.add_subparsers(dest="command", required=True) init = sub.add_parser("init", help="write the skeleton from the template; refuses an existing file") init.add_argument("--project-root", required=True, help="project root that holds `.okstra/`") init.add_argument("--task-group", required=True, help="task-group name; slugified for the directory") init.set_defaults(func=_init_command) return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) return args.func(args) if __name__ == "__main__": sys.exit(main())