"""Build the compact analysis-worker input packet for a task run.""" from __future__ import annotations import re from pathlib import Path from . import group_context BRIEF_SECTIONS = ( "Identity", "Request Summary", "Current Context", "Evidence and Source Materials", "Task-Type Focus", "Constraints and Risks", "Out of Scope", "Configuration References and Expected Values", "Deployment Manifests and Expected Values", "Questions for Workers", "Expected Outputs", "Task Continuity Notes", "Available MCP Servers", ) CANONICAL_BRIEF_SECTIONS = ( "Source Material", "Context", "Problem / Symptom", "Desired Outcome", "Expected Behavior", "Preserved Behavior", "Expected Outcome", "External Gates", "Constraints", "Scan Scope", "Priority Lenses", "Related Artifacts", "Related Task Graph", "Open Questions", "Reporter Confirmations", "Augmentation", ) + BRIEF_SECTIONS PROFILE_SECTIONS = ( "Primary focus areas", "Expected output emphasis", "Non-goals", ) WORKER_PROFILE_SECTIONS_BY_TASK_TYPE = { # `Brief consumption` carries the reporter-confirmation rules — do not infer an # unconfirmed `intent-check:` / `conversion-block:`, carry it forward as a # blocker instead. Those bind the analysis worker, and the packet is the only # profile text a worker receives, so a phase that omits the section ships a # worker that can turn an unconfirmed intent into a settled requirement. "requirements-discovery": ( "Brief consumption", "Worker discovery procedure", ), "error-analysis": ( "Brief consumption", "Worker diagnosis procedure", ), "implementation-option-selection": ( "Brief consumption", "Worker direction-selection procedure", "Pre-selection context exploration", "Option evaluation rules", ), "implementation-planning": ( "Brief consumption", "Worker planning procedure", "Pre-planning context exploration", "Design principles applied to realization review", ), "final-verification": ( "Worker verification procedure", ), "improvement-discovery": ( "Brief consumption", "Worker candidate procedure", # Every analyser examines every resolved lens but leads with its own, and # keeps duplicate / conflicting relations. Held only in the lead's profile, # the workers converge on the same easy lens and the cross-check thins out # while the aggregate still looks complete. "Worker diversity rule", ), } CLARIFICATION_SECTIONS = ( "Clarification Items", "Clarification Response Carried In From Previous Run", ) _FRONTMATTER_RE = re.compile(r"\A---\n(?P.*?)\n---\n", re.DOTALL) def build_analysis_packet( *, task_key: str, task_type: str, task_brief_path: Path, analysis_profile_path: Path, reference_expectations_path: Path, clarification_response_path: Path | None, directive: str, instruction_set_relative_path: str, group_context_path: Path | None = None, fix_history_text: str = "", stage_ledger_json: str = "", stage_ledger_notice: str = "", prior_planning_summary: str = "", direct_work_text: str = "", ) -> str: """Return the primary compact input for Claude/Codex/Antigravity analysers. `fix_history_text`, `stage_ledger_json` / `stage_ledger_notice`, and `prior_planning_summary` are pre-rendered by their own modules (`fix_cycles`, `stage_ledger`, `prior_planning`) because each is assembled from disk state this file does not own. Each arrives already narrowed, and an empty string means the caller found nothing to carry — the matching block is then omitted rather than rendered saying "none". `group_context_path` is the task-group context prepare copied into the instruction set (`task-group-context.md`); `None` means the group has no such document and the packet carries no `## Task-Group Context` section. """ brief_text = task_brief_path.read_text(encoding="utf-8") group_context_text = _read_optional(group_context_path) if group_context_path else "" profile_text = analysis_profile_path.read_text(encoding="utf-8") reference_text = _read_optional(reference_expectations_path) clarification_text = ( _read_optional(clarification_response_path) if clarification_response_path else "" ) parts = [_packet_frontmatter(brief_text, task_key)] parts.extend( _intro_block( task_key, task_type, instruction_set_relative_path, bool(clarification_response_path), bool(group_context_text), ) ) # 그룹 문서는 사람 절과 okstra 메모리 영역으로 갈라 싣는다. 사람 절("왜")은 # 브리프 발췌보다 앞 — 짧고, 46KB 에서 잘리는 읽기(agy view_file)에서도 # 도착해야 한다. 형제 task 의 결론은 브리프 뒤 — 브리프를 밀어내면 안 된다. group_human, group_memory, _ = group_context.split_memory_region( _strip_frontmatter(group_context_text) ) parts.extend(_group_context_block(group_human)) parts.extend(_brief_block(brief_text)) if direct_work_text: parts.extend(["", direct_work_text, ""]) parts.extend(_group_memory_block( group_memory, _own_task_segment(task_key), "\n".join((brief_text, group_human, clarification_text, directive, fix_history_text, prior_planning_summary, stage_ledger_json)), )) parts.extend(_profile_block(task_type, profile_text)) parts.extend(_reference_block(reference_text)) parts.extend(_fix_history_block(fix_history_text)) parts.extend(_stage_ledger_block(stage_ledger_json)) parts.extend(_stage_ledger_unavailable_block(stage_ledger_notice)) parts.extend(_prior_planning_block(prior_planning_summary)) parts.extend(_clarification_block(clarification_text)) parts.extend(_directive_block(directive)) body = "\n".join(part.rstrip() for part in parts).rstrip() + "\n" return _with_section_index(body) _HEADING_RE = re.compile(r"\A#{1,2} \S") _FENCE_RE = re.compile(r"\A\s*(```|~~~)") # 목차 자신이 차지하는 줄 수 중 표제 행을 뺀 나머지. 행 수는 표제 개수로 # 정해지므로 삽입 전에 전체 이동량을 계산할 수 있다. _INDEX_PREAMBLE_LINES = 5 def reference_source_extracts( packet_text: str, task_type: str, *, brief_text: str, profile_text: str, reference_text: str, clarification_text: str, ) -> str: """합성 입력 안에 원문이 함께 있을 때 일치하는 발췌만 참조로 바꾼다.""" lines = packet_text.splitlines() headings = _heading_lines(lines) if headings and headings[0][1] == "## Section Index": start = headings[0][0] - 2 end = start + _INDEX_PREAMBLE_LINES + len(headings) - 1 unindexed = "\n".join(lines[:start] + lines[end:]) + "\n" # 생성기가 만든 목차임을 재구성으로 확인한다. 수정된 목차가 있으면 # 원문 전체를 유지해 내용 손실이나 잘못된 줄번호를 만들지 않는다. if _with_section_index(unindexed) != packet_text: return packet_text lines = unindexed.splitlines() blocks = ( ("Task brief", brief_text, _brief_block(brief_text)), ("Analysis profile", profile_text, _profile_block(task_type, profile_text)), ("Reference expectations", reference_text, _reference_block(reference_text)), ("Clarification response", clarification_text, _clarification_block(clarification_text)), ) changed = False for label, source_text, parts in blocks: expected = "\n".join(part.rstrip() for part in parts).strip() if not source_text.strip() or not expected: continue heading = expected.splitlines()[0] reference = f'{heading}\n\nRead the same extract in "Source: {label}" below.' if len(reference) >= len(expected): continue for number, text in _heading_lines(lines): start = number - 1 end = start + len(expected.splitlines()) if text == heading and "\n".join(lines[start:end]) == expected: lines[start:end] = reference.splitlines() changed = True break return "\n".join(lines).rstrip() + "\n" if changed else packet_text def _with_section_index(body: str) -> str: """Prefix the rendered packet with every section's line range. The packet's own structure is the only thing a worker can navigate it by, and a worker cannot navigate what it cannot see. Measured 2026-08-21: the antigravity CLI's `view_file` truncates at 46,080 bytes per call, carries no line-range parameter, and reports only ` lines, bytes` — the same string whether it returned the file or 40% of it. On a 114KB packet that first call ended at line 510, so `## Stage Ledger` (572) and `## Clarification Carry-In Extract` (680) never arrived and the worker planned from the pre-migration brief that fills the first 42KB. Every evidence line it cited was under 400. With the ranges here a truncated reader can fetch the block it is missing by range instead of re-reading the head; the lead had to write those ranges by hand to get a usable retry. """ lines = body.splitlines() anchor = next( (i for i, line in enumerate(lines) if line.startswith("# OKSTRA Analysis Packet")), None, ) if anchor is None: return body headings = _heading_lines(lines) if not headings: return body shift = _INDEX_PREAMBLE_LINES + len(headings) total = len(lines) + shift starts = [number + shift for number, _ in headings] rows = [ f"- {start}-{end} `{text}`" for start, end, (_, text) in zip( starts, [nxt - 1 for nxt in starts[1:]] + [total], headings ) ] block = [ "", "## Section Index", "", f"This packet is {total} lines. Read a section by line range with" " `sed -n ',p'` rather than opening the whole file: a" " whole-file read is truncated by some worker CLIs without reporting" " it, and the sections that describe current state are last.", "", *rows, ] assert len(block) == shift, (len(block), shift) return "\n".join(lines[:anchor + 1] + block + lines[anchor + 1:]) + "\n" def _heading_lines(lines: list[str]) -> list[tuple[int, str]]: """Every `#`/`##` heading outside a fence, as (1-based line, text). Fences are tracked because the Stage Ledger and the carried-in clarification rows embed JSON and shell text where a `#` starts a comment, not a section. """ out: list[tuple[int, str]] = [] fenced = False for number, line in enumerate(lines, start=1): if _FENCE_RE.match(line): fenced = not fenced continue if fenced or not _HEADING_RE.match(line): continue if line.startswith("# OKSTRA Analysis Packet"): continue out.append((number, line.strip())) return out def _packet_frontmatter(brief_text: str, task_key: str) -> str: frontmatter = _extract_frontmatter(brief_text) if frontmatter: return ( f"---\n{frontmatter}\n" 'packetVersion: "1.0"\n' "packetRole: analysis-worker-primary\n---" ) return ( "---\n" f'title: OKSTRA Analysis Packet - {task_key}\n' 'packetVersion: "1.0"\n' "packetRole: analysis-worker-primary\n" "---" ) def _intro_block( task_key: str, task_type: str, instruction_set: str, has_clarification: bool, has_group_context: bool = False, ) -> list[str]: lines = [ f"# OKSTRA Analysis Packet - {task_key}", "", "## Packet Role", "", "- This packet is the primary required reading for analysis workers.", "- It extracts task-specific material from the source files listed below.", "- Read source files only for evidence verification or missing detail.", "", "## Source Files", "", f"- Task brief: `{instruction_set}/task-brief.md`", f"- Analysis profile: `{instruction_set}/analysis-profile.md`", f"- Analysis material: `{instruction_set}/analysis-material.md`", f"- Reference expectations: `{instruction_set}/reference-expectations.md`", ] if task_type == "final-verification": lines.append( f"- Verification target: `{instruction_set}/verification-target.md`" ) if has_clarification: lines.append( f"- Clarification response: `{instruction_set}/clarification-response.md`" ) if has_group_context: lines.append( f"- Task-group context: `{instruction_set}/task-group-context.md`" ) return lines GROUP_CONTEXT_PREFACE = ( "Background shared by every task in this task-group, copied from the group's " "`group-context.md`. It is not this task's requirement ledger: the " "`## Task-Specific Brief Extract` below decides what this run must deliver, " "while this section decides how the result is measured and what no task in the " "group may break. Where the two conflict, raise a `Clarification Items` row; " "never resolve the conflict silently in either direction." ) GROUP_MEMORY_PREFACE = ( "What the other tasks of this task-group concluded in their latest run, " "written by okstra at each run's report-finalize into the group's " "`group-context.md`. A headline is a claim; its evidence is the record at " "`record`. This section is not this task's requirement ledger, and a sibling's " "decision does not bind this run: where a sibling's conclusion conflicts with " "the `## Task-Specific Brief Extract`, raise a `Clarification Items` row rather " "than re-deriving what the sibling already settled or silently overriding it. " "Every sibling has a headline and record pointer below. Full details are " "included for explicit task references and their referenced siblings, and " "for entries with watch-outs or no record pointer. For an index-only entry, " "read its section in the Task-group context source or its record when its " "headline may affect this task or the relationship is unclear. Omitted " "details do not mean the sibling has no decisions or follow-ups." ) def _group_context_block(human_text: str) -> list[str]: """사람 절만 싣는다. 메모리 전용 문서(사람 `## ` 절 없음)는 절을 만들지 않는다.""" if not any(line.startswith("## ") for line in human_text.splitlines()): return [] return [ "", "## Task-Group Context", "", GROUP_CONTEXT_PREFACE, "", human_text.strip(), ] def _own_task_segment(task_key: str) -> str: return group_context.slugify_task_segment(re.split(r"[:/]", task_key)[-1]) _TASK_REFERENCE_RE = re.compile(r"[a-z0-9](?:[a-z0-9_.-]*[a-z0-9])?", re.IGNORECASE) def _group_memory_block( region: str, own_task_segment: str, reference_text: str, ) -> list[str]: """형제 전체의 색인을 유지하며 참조 연결과 주의사항이 있는 항목을 펼친다.""" entries = [ entry for entry in group_context.parse_memory_entries(region) if entry.task_id != own_task_segment ] if not entries: return [] tokens = set(_TASK_REFERENCE_RE.findall(reference_text.casefold())) own_ids = {own_task_segment, group_context.ticket_id_from_brief_id(own_task_segment)} rendered = [group_context.render_memory_entries([entry]).strip() for entry in entries] entry_tokens = [set(_TASK_REFERENCE_RE.findall(text.casefold())) for text in rendered] selected: set[int] = set() while True: previous_count = len(selected) for index, entry in enumerate(entries): aliases = {entry.task_id.casefold(), group_context.ticket_id_from_brief_id(entry.task_id).casefold()} if index in selected: continue if (aliases & tokens or own_ids & entry_tokens[index] or entry.watch_out or not entry.record): selected.add(index) tokens.update(entry_tokens[index]) if len(selected) == previous_count: break lines = ["", "## Task-Group Memory", "", GROUP_MEMORY_PREFACE, ""] for index, entry in enumerate(entries): if index in selected: lines.extend([rendered[index], ""]) else: lines.extend([ f"### {entry.task_id}", f"- headline: {entry.headline or '_(none)_'}", f"- record: `{entry.record}`", "- Detail: index-only; read the source if relevant or uncertain.", "", ]) if entry.source == "direct": lines.extend(["- Source: direct work; no cross-verification performed for this record.", ""]) return lines def _brief_block(brief_text: str) -> list[str]: return [ "", "## Task-Specific Brief Extract", "", _extract_sections(brief_text, CANONICAL_BRIEF_SECTIONS), ] def _profile_block(task_type: str, profile_text: str) -> list[str]: section_names = ( PROFILE_SECTIONS + WORKER_PROFILE_SECTIONS_BY_TASK_TYPE.get(task_type, ()) ) profile_sections = _extract_sections(profile_text, section_names) profile_bullets = _extract_bullet_sections(profile_text, section_names) profile_focus = "\n\n".join( part for part in (profile_sections, profile_bullets) if part and not part.startswith("- No matching") ) return [ "", "## Phase Focus Extract", "", f"- Task Type: `{task_type}`", "", _extract_profile_prelude(profile_text), "", profile_focus or "- No matching source sections were available.", ] def _reference_block(reference_text: str) -> list[str]: body = reference_text.strip() or "- No reference expectations content was available." return ["", "## Reference Expectations", "", body] def _fix_history_block(fix_history_text: str) -> list[str]: if not fix_history_text.strip(): return [] return [ "", "## Fix History", "", "Past bug-fix cycles registered on this task. Treat the open cycle's", "symptom as a prior-defect signal when analysing.", "", fix_history_text, "", ] def _stage_ledger_block(stage_ledger_json: str) -> list[str]: if not stage_ledger_json.strip(): return [] return [ "", "## Stage Ledger", "", "Facts about this task's stages as they stand on disk — not a plan.", "A stage whose `status` is `done` is already implemented and will not", "be executed again, so do not rewrite its plan body.", "", "`stages` lists every stage the latest plan declares, so every number", "in it is taken. Never reuse or renumber one: a new stage takes the", "next number after the highest listed here, and reworking a completed", "stage means cancelling it and adding a new number, never editing it", "in place. `sourcePlan` is the plan the completed stages were built", "against; `latestPlan` is the plan this list came from. When the two", "differ, the completed work followed the former and the numbering", "authority is the latter.", "", "A `planDivergence` entry means the two plans disagree about a", "completed stage, or that the plan the completed stages were built", "against could not be read so that comparison never ran. Treat either", "as a blocker for any new stage number and report it; do not resolve", "it by choosing one of the two yourself. Completed stages built", "against an earlier plan are not a divergence.", "", "```json", stage_ledger_json.strip(), "```", "", ] def _stage_ledger_unavailable_block(notice: str) -> list[str]: """원장을 못 읽었다는 사실을 packet 에 남긴다. 블록을 생략하면 저작 쪽은 "이전 계획이 없다" 로 읽는다 — 프로파일이 그렇게 지시한다. 계획이 있는데 못 읽은 경우에 그 침묵은 거짓이고, 이미 점유된 번호를 새 stage 에 다시 내주는 경로가 된다. """ if not notice.strip(): return [] return [ "", "## Stage Ledger", "", "This task's stage facts could NOT be read, so no ledger is included.", "Do not read this as 'the task has no prior plan' — a plan exists and", "this run could not parse it.", "", f"- Reason: {notice.strip()}", "", "Continue this planning run by reading that report and the implementation", "carry records. Repair an unambiguous dependency notation error in the", "new planning report, preserving every existing stage number and title", "and the bodies of completed stages. Record each original and corrected", "value with its source path. Do not overwrite the prior report or fall", "back to an older plan. If the intended dependency cannot be established", "from those records, report that uncertainty as a blocker; do not guess.", "Validate the corrected Stage Map before treating its dependencies as facts.", "", "Do not assign a number to any new stage in this state. A number taken", "by a plan this run could not read would collide with the completed", "work under it, and that collision stays silent until integration.", "Until the corrected map validates, keep new stage numbering blocked.", "", ] def _prior_planning_block(prior_planning_summary: str) -> list[str]: """직전 계획 run 의 맥락. 직전 run 이 없으면 블록 자체가 없다. 머리글이 이 블록의 용도를 스스로 말한다. 그러지 않으면 워커는 표에 실린 계획을 승인된 것으로 읽고 그대로 옮겨 적는다 — 이 블록이 실리는 run 은 직전 계획이 **승인에 도달하지 못해** 다시 도는 run 이므로 그 독해는 틀렸다. """ if not prior_planning_summary.strip(): return [] return [ "", "## Prior Planning Run", "", "What this task's previous implementation-planning run recorded. It is", "here so that stages the previous plan already settled stay textually", "stable across the re-run, and so settled analysis is not derived from", "the brief a second time under new `P-*` numbers.", "", "It is NOT an approved plan and NOT an instruction to copy. The", "previous run did not reach approval — that is why this run exists —", "so every line below is still open to revision. Where the brief, the", "directive, or a clarification answer disagrees with it, they win, and", "a stage you are changing is rewritten rather than carried. Read the", "report at the path below for anything this summary leaves out.", "", prior_planning_summary.strip(), "", ] def _clarification_block(clarification_text: str) -> list[str]: if not clarification_text.strip(): return [] return [ "", "## Clarification Carry-In Extract", "", _extract_sections(clarification_text, CLARIFICATION_SECTIONS), ] def _directive_block(directive: str) -> list[str]: if not directive: return [] return ["", "## Directive", "", directive.strip()] def _extract_frontmatter(text: str) -> str: match = _FRONTMATTER_RE.match(text) return match.group("body").rstrip() if match else "" def _extract_profile_prelude(text: str) -> str: lines = [] for line in text.splitlines(): if line.startswith(("{{INCLUDE:", "## ", "