"""Build the compact analysis-worker input packet for a task run.""" from __future__ import annotations import re from pathlib import Path 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 when scoring options", ), "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, fix_history_text: str = "", stage_ledger_json: str = "", stage_ledger_notice: str = "", ) -> str: """Return the primary compact input for Claude/Codex/Antigravity analysers.""" brief_text = task_brief_path.read_text(encoding="utf-8") 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), ) ) parts.extend(_brief_block(brief_text)) 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(_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 _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, ) -> 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`" ) 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. Treat it as a blocker for any new stage number and", "report it; do not resolve it by choosing one of the two yourself.", "", "```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()}", "", "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.", "Report this as a blocker instead.", "", ] 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:", "## ", "