"""LLM client interfaces.""" from __future__ import annotations import re from typing import Dict, Optional, Protocol from packages.artifacts.design import DesignArtifact, MetadataPlanItem, V1_ALLOWLIST from packages.artifacts.qa_report import QAReportArtifact, TestScenario from packages.artifacts.requirements import ClarificationQuestion, RequirementsArtifact from packages.artifacts.review import ReviewArtifact, ReviewFinding from packages.config import Settings, get_settings from packages.integrations.doc_validation import DocCheckResult, DocReference from packages.integrations.duplicates import DuplicateCheckResult class LLMClient(Protocol): async def analyze_requirements(self, prompt: str) -> RequirementsArtifact: ... async def design_solution(self, prompt: str) -> DesignArtifact: ... async def review_code(self, prompt: str) -> ReviewArtifact: ... async def fix_code(self, prompt: str) -> Dict[str, str]: ... async def generate_qa_report(self, prompt: str) -> QAReportArtifact: ... async def check_duplicate(self, prompt: str) -> DuplicateCheckResult: ... async def check_already_implemented(self, prompt: str) -> DocCheckResult: ... class MockLLMClient: """Heuristic analyst/architect for offline/local use.""" async def check_already_implemented(self, prompt: str) -> DocCheckResult: m = re.search( r"NEW ticket:.*?Summary:\s*(.+?)\nDescription:\s*(.+?)(?:\n\nCONFLUENCE|\nCONFLUENCE)", prompt, re.S, ) new_summary = (m.group(1).strip() if m else "") new_desc = (m.group(2).strip() if m else "") new_tokens = _significant_tokens(f"{new_summary} {new_desc}") field_tokens = {t for t in new_tokens if t.endswith("__c") or "__c" in t} best: Optional[tuple[float, str, str, str]] = None # score, title, url, snippet for hm in re.finditer( r"- Title:\s*(.+?)\n\s*URL:\s*(.+?)\n\s*Snippet:\s*(.+?)(?:\n- Title:|\Z)", prompt, re.S, ): title, url, snippet = hm.group(1).strip(), hm.group(2).strip(), hm.group(3).strip() if url == "(none)": url = "" blob_tokens = _significant_tokens(f"{title} {snippet}") if not blob_tokens or not new_tokens: continue overlap = len(new_tokens & blob_tokens) / max(len(new_tokens), 1) field_hit = bool(field_tokens and (field_tokens & blob_tokens)) implemented_signal = any( w in f"{title} {snippet}".lower() for w in ("implemented", "already exists", "deployed", "in production", "delivered") ) score = overlap + (0.35 if field_hit else 0.0) + (0.2 if implemented_signal else 0.0) if best is None or score > best[0]: best = (score, title, url, snippet) if best and (best[0] >= 0.45 or field_tokens and best[0] >= 0.35): return DocCheckResult( already_implemented=True, rationale=f"Mock LLM: Confluence overlap {best[0]:.2f} with '{best[1]}'", confidence=min(0.95, 0.5 + best[0] / 2), references=[DocReference(title=best[1], url=best[2], snippet=best[3][:200])], ) return DocCheckResult( already_implemented=False, rationale="Mock LLM: no Confluence hit strongly matches existing functionality", confidence=0.4, ) async def check_duplicate(self, prompt: str) -> DuplicateCheckResult: """Mark duplicate when a candidate summary shares distinctive tokens with NEW.""" m = re.search(r"NEW ticket:.*?Summary:\s*(.+?)(?:\nDescription:|\n\n)", prompt, re.S) new_summary = (m.group(1).strip() if m else "") new_tokens = _significant_tokens(new_summary) if not new_tokens: return DuplicateCheckResult(is_duplicate=False, rationale="No comparable summary", confidence=0.2) best_key = None best_score = 0.0 for m in re.finditer( r"- Key:\s*(\S+)\n\s*Summary:\s*(.+?)(?:\n\s*Description:|\n\s*Status:)", prompt, re.S, ): key, summary = m.group(1).strip(), m.group(2).strip() cand_tokens = _significant_tokens(summary) if not cand_tokens: continue overlap = len(new_tokens & cand_tokens) / max(len(new_tokens), 1) if overlap > best_score: best_score = overlap best_key = key if best_key and best_score >= 0.5: return DuplicateCheckResult( is_duplicate=True, original_key=best_key, rationale=f"Mock LLM: summary overlap {best_score:.2f} with {best_key}", confidence=min(0.95, 0.5 + best_score / 2), ) return DuplicateCheckResult( is_duplicate=False, rationale="Mock LLM: no strong summary overlap", confidence=0.4, ) async def analyze_requirements(self, prompt: str) -> RequirementsArtifact: jira_key = _extract(prompt, r"Jira key:\s*(\S+)", "UNKNOWN") summary = _extract(prompt, r"Summary:\s*(.+)", "Untitled") description = _section(prompt, "Description:") ac_block = _section(prompt, "Acceptance criteria:") comments_block = _section(prompt, "Comments:") acceptance = [ l.strip("- ").strip() for l in ac_block.splitlines() if l.strip() and "(none provided)" not in l.lower() ] # Treat substantive human comments as clarification answers (ignore bot lines) human_lines = [ ln for ln in comments_block.splitlines() if ln.strip() and "agent nova needs clarification" not in ln.lower() and not ln.lower().startswith("- agent-nova") and "(none)" not in ln.lower() ] human_text = "\n".join(human_lines) has_human_detail = len(human_text.strip()) > 60 if has_human_detail and not acceptance: acceptance = ["As clarified in Jira comments"] unclear_signals = [ len(description.strip()) < 40 and not has_human_detail, not acceptance, "improve" in summary.lower() and not acceptance and not has_human_detail, ("TBD" in description.upper() or "TODO" in description.upper()) and not has_human_detail, ] req_text = f"{summary}\n{description}\n{ac_block}".lower() out_of_scope: list[str] = [] if "production" in req_text and "deploy" in req_text: out_of_scope.append("Production deployment is out of Agent Nova v1 scope") if "sharing" in req_text and ("owd" in req_text or "redesign" in req_text): out_of_scope.append("Sharing/OWD redesign is outside the v1 metadata allowlist") if any(unclear_signals) and not (acceptance and len(description.strip()) >= 40): questions = [] n = 1 if not acceptance: questions.append( ClarificationQuestion( number=n, question="What are the acceptance criteria for this story?", rationale="Missing AC", ) ) n += 1 if len(description.strip()) < 40: questions.append( ClarificationQuestion( number=n, question="Please describe the desired Salesforce behavior, objects, and personas.", rationale="Thin description", ) ) n += 1 if not questions: questions.append( ClarificationQuestion( number=1, question="Please provide additional detail so implementation can proceed.", ) ) return RequirementsArtifact( jira_key=jira_key, summary=summary.strip(), acceptance_criteria=acceptance, clarification_questions=questions, is_clear=False, confidence=0.35, referenced_documentation=_refs(prompt), out_of_scope_items=out_of_scope, notes="Mock LLM flagged incomplete requirements", ) return RequirementsArtifact( jira_key=jira_key, summary=summary.strip(), acceptance_criteria=acceptance or ["As described in the Jira story"], clarification_questions=[], is_clear=True, confidence=0.82, referenced_documentation=_refs(prompt), out_of_scope_items=out_of_scope, notes="Mock LLM accepted requirements as clear", ) async def design_solution(self, prompt: str) -> DesignArtifact: jira_key = _extract(prompt, r"Jira key:\s*(\S+)", "UNKNOWN") summary = _extract(prompt, r"Summary:\s*(.+)", "Untitled").strip() ac_block = _section(prompt, "Acceptance criteria:") # Score requirements content only — ignore allowlist boilerplate in the prompt. req_text = f"{summary}\n{ac_block}".lower() plan: list[MetadataPlanItem] = [] risks: list[str] = [] out_of = False # Heuristic object guess obj = "Account" for candidate in ("Account", "Contact", "Opportunity", "Lead", "Case"): if candidate.lower() in req_text: obj = candidate break if "field" in req_text or "__c" in req_text or "picklist" in req_text or "custom" in req_text: api = "Customer_Tier__c" if "tier" in req_text else "Custom_Field__c" m = re.search(r"([A-Za-z][A-Za-z0-9_]*__c)", f"{summary}\n{ac_block}") if m: api = m.group(1) note_bits = ["Custom field from requirements"] if "picklist" in req_text: note_bits = ["Picklist"] label_m = re.search( r"(?:field\s+)?label\s+(?:as|is|:)\s+([A-Za-z][A-Za-z0-9 ]+)", f"{summary}\n{ac_block}", re.I, ) if label_m: note_bits.append(f"label {label_m.group(1).strip().rstrip('.')}") vals_m = re.search( r"(?:picklist\s+)?values?\s*(?:are\s+|:)?\s*([A-Za-z0-9][A-Za-z0-9 ,/\-]{2,})", f"{summary}\n{ac_block}", re.I, ) if vals_m: vals = re.split(r"[.;]", vals_m.group(1), maxsplit=1)[0] note_bits.append(f"values {vals.strip()}") def_m = re.search( r"(?:set\s+)?([A-Za-z][A-Za-z0-9 \-]+?)\s+as\s+the\s+default", f"{summary}\n{ac_block}", re.I, ) if def_m: note_bits.append(f"default {def_m.group(1).strip()}") plan.append( MetadataPlanItem( metadata_type="CustomField", api_name=api, action="create", object=obj, notes="; ".join(note_bits), ) ) if "validation" in req_text: vr_name = f"{obj}_Validation" vr_m = re.search( r"([A-Za-z][A-Za-z0-9_]*_Required(?:_For_[A-Za-z0-9_]+)?)", f"{summary}\n{ac_block}", ) if vr_m: vr_name = vr_m.group(1) vr_notes = "Validation rule from acceptance criteria" if "customer" in req_text: field_api = plan[0].api_name if plan else "Custom_Field__c" vr_notes = f"Block save when Type is Customer and {field_api} is blank" plan.append( MetadataPlanItem( metadata_type="ValidationRule", api_name=vr_name, action="create", object=obj, notes=vr_notes, ) ) if "permission set" in req_text or "permissionset" in req_text: plan.append( MetadataPlanItem( metadata_type="PermissionSet", api_name=f"{obj}_Field_Access", action="create", notes="Grant edit access for intended personas", ) ) if "flow" in req_text and "record" in req_text: plan.append( MetadataPlanItem( metadata_type="Flow", api_name=f"{obj}_RecordTriggered", action="create", object=obj, notes="Record-triggered Flow only (v1)", ) ) if "apex" in req_text or "trigger" in req_text: if "trigger" in req_text: plan.append( MetadataPlanItem( metadata_type="ApexTrigger", api_name=f"{obj}Trigger", action="create", object=obj, ) ) plan.append( MetadataPlanItem( metadata_type="ApexClass", api_name=f"{obj}Handler", action="create", object=obj, notes="with sharing handler", ) ) if "lwc" in req_text or "lightning" in req_text: plan.append( MetadataPlanItem( metadata_type="LightningComponentBundle", api_name=f"{obj[0].lower() + obj[1:]}Component", action="create", object=obj, ) ) if "sharing" in req_text and ("owd" in req_text or "redesign" in req_text): out_of = True risks.append("Sharing/OWD redesign is outside the v1 allowlist — escalate") if re.search(r"\bprofiles?\b", req_text) and "permission set" not in req_text: out_of = True risks.append("Profile changes are outside the v1 allowlist — escalate") if "production" in req_text and "deploy" in req_text: risks.append("Production deployment is out of autonomous scope") if not plan: plan.append( MetadataPlanItem( metadata_type="CustomField", api_name="TBD__c", action="create", object=obj, notes="Default plan placeholder — refine with clearer requirements", ) ) for item in plan: if item.metadata_type not in V1_ALLOWLIST: out_of = True risks.append(f"Metadata type outside v1 allowlist: {item.metadata_type}") return DesignArtifact( jira_key=jira_key, summary=summary, approach=( f"Implement allowlisted Salesforce metadata on {obj} per acceptance criteria; " "prefer configuration (fields/validation/permission sets) before Apex." ), metadata_plan=plan, dependencies=["Existing object layout / FLS for target personas"], risks_or_escalations=risks, out_of_allowlist=out_of, confidence=0.75 if plan and not out_of else 0.45, notes="Mock Solution Architect design", ) async def review_code(self, prompt: str) -> ReviewArtifact: jira_key = _extract(prompt, r"Jira key:\s*(\S+)", "UNKNOWN") cycle = int(_extract(prompt, r"Review cycle:\s*(\d+)", "0")) files_block = _section(prompt, "Generated files:") metadata_block = _section(prompt, "Metadata plan:") findings: list[ReviewFinding] = [] for path, content in _parse_file_blocks(files_block).items(): if path.endswith(".cls") and "without sharing" in content: findings.append( ReviewFinding( severity="high", category="security", file=path, message="Apex class should use 'with sharing' by default", fixable=True, ) ) if path.endswith(".cls") and "@istest" not in content.lower(): if any("ApexClass" in line for line in metadata_block.splitlines()): findings.append( ReviewFinding( severity="high", category="test_coverage", file=path, message="Missing @IsTest companion class for Apex production code", fixable=True, ) ) if path.endswith(".trigger") and "bulk" not in content.lower(): findings.append( ReviewFinding( severity="medium", category="bulkification", file=path, message="Trigger should document bulk-safe handling", fixable=True, ) ) static_block = _section(prompt, "Static analysis violations:") for line in static_block.splitlines(): if line.strip().startswith("-") and "(none)" not in line.lower(): findings.append( ReviewFinding( severity="high", category="static_analysis", file=None, message=line.strip("- ").strip(), fixable=True, ) ) blocking = [f for f in findings if f.severity in ("critical", "high")] passed = len(blocking) == 0 static_status = "failed" if any(f.category == "static_analysis" for f in findings) else "passed" if "(none)" in static_block.lower() or not static_block.strip(): static_status = "passed" return ReviewArtifact( jira_key=jira_key, passed=passed, cycle=cycle, findings=findings, summary=( "Code review passed — no blocking findings" if passed else f"Code review failed with {len(blocking)} blocking finding(s)" ), static_analysis_status=static_status, notes="Mock Code Reviewer", ) async def fix_code(self, prompt: str) -> Dict[str, str]: files = _parse_file_blocks(_section(prompt, "Files to fix:")) findings_block = _section(prompt, "Findings to address:") updated: Dict[str, str] = {} for path, content in files.items(): new_content = content if path.endswith(".cls"): if "without sharing" in new_content: new_content = new_content.replace("without sharing", "with sharing") if "@istest" not in new_content.lower() and "test_coverage" in findings_block.lower(): cls_name = _extract(new_content, r"class\s+(\w+)", "Handler") test_name = f"{cls_name}Test" test_path = path.replace(f"/{cls_name}.cls", f"/{test_name}.cls") test_body = f"""@IsTest private class {test_name} {{ @IsTest static void testRun() {{ {cls_name}.run(); }} }} """ updated[test_path] = test_body meta_path = test_path.replace(".cls", ".cls-meta.xml") updated[meta_path] = f""" 59.0 Active """ if path.endswith(".trigger") and "bulk" not in new_content.lower(): new_content = new_content.replace( "// Agent Nova stub", "// Agent Nova stub — bulk-safe: processes Trigger.new in single pass", ) if new_content != content: updated[path] = new_content return updated async def generate_qa_report(self, prompt: str) -> QAReportArtifact: jira_key = _extract(prompt, r"Jira key:\s*(\S+)", "UNKNOWN") ac_block = _section(prompt, "Acceptance criteria:") apex_block = _section(prompt, "Apex test results:") acceptance = [ l.strip("- ").strip() for l in ac_block.splitlines() if l.strip() and "(none)" not in l.lower() ] matrix: list[TestScenario] = [] manual: list[str] = [] for i, ac in enumerate(acceptance or ["Story behaves as specified"], start=1): matrix.append( TestScenario( name=f"positive_{i}", type="positive", description=f"Verify: {ac}", acceptance_criterion=ac, ) ) matrix.append( TestScenario( name=f"negative_{i}", type="negative", description=f"Invalid input should not satisfy: {ac}", acceptance_criterion=ac, ) ) manual.append(f"Manual UI check for: {ac}") apex_failures: list[str] = [] tests_run = 0 tests_passed = 0 passed = True if "failures:" in apex_block.lower(): for line in apex_block.splitlines(): if line.strip().startswith("-") and "none" not in line.lower(): apex_failures.append(line.strip("- ").strip()) if apex_failures: passed = False if "tests_run:" in apex_block.lower(): tests_run = int(_extract(apex_block, r"tests_run:\s*(\d+)", "0")) tests_passed = int(_extract(apex_block, r"tests_passed:\s*(\d+)", "0")) if tests_run > 0 and tests_passed < tests_run: passed = False return QAReportArtifact( jira_key=jira_key, passed=passed, test_matrix=matrix, apex_tests_run=tests_run, apex_tests_passed=tests_passed, apex_failures=apex_failures, manual_scenarios=manual, summary="QA passed — acceptance criteria covered" if passed else "QA failed — see failures", notes="Mock QA Engineer", ) class OpenAILLMClient: def __init__(self, settings: Optional[Settings] = None) -> None: self.settings = settings or get_settings() def _chat(self): from langchain_openai import ChatOpenAI return ChatOpenAI( model=self.settings.openai_model, api_key=self.settings.openai_api_key or None, temperature=0, ) async def check_duplicate(self, prompt: str) -> DuplicateCheckResult: from pydantic import BaseModel, Field class _Out(BaseModel): is_duplicate: bool original_key: str | None = None rationale: str = "" confidence: float = Field(default=0.5, ge=0.0, le=1.0) llm = self._chat().with_structured_output(_Out) result: _Out = await llm.ainvoke(prompt) key = (result.original_key or "").strip() or None if result.is_duplicate and not key: return DuplicateCheckResult( is_duplicate=False, rationale="Model marked duplicate without original_key", confidence=result.confidence, ) return DuplicateCheckResult( is_duplicate=bool(result.is_duplicate and key), original_key=key if result.is_duplicate else None, rationale=result.rationale or "", confidence=result.confidence, ) async def check_already_implemented(self, prompt: str) -> DocCheckResult: from pydantic import BaseModel, Field class _Ref(BaseModel): title: str url: str | None = None snippet: str | None = None class _Out(BaseModel): already_implemented: bool rationale: str = "" confidence: float = Field(default=0.5, ge=0.0, le=1.0) references: list[_Ref] = Field(default_factory=list) llm = self._chat().with_structured_output(_Out) result: _Out = await llm.ainvoke(prompt) refs = [ DocReference(title=r.title, url=(r.url or "").strip(), snippet=(r.snippet or "")[:300]) for r in result.references if r.title ] if result.already_implemented and not refs: # Soften: require at least one reference when claiming already implemented return DocCheckResult( already_implemented=False, rationale="Model marked already_implemented without references", confidence=result.confidence, ) return DocCheckResult( already_implemented=bool(result.already_implemented and refs), rationale=result.rationale or "", confidence=result.confidence, references=refs, ) async def analyze_requirements(self, prompt: str) -> RequirementsArtifact: from pydantic import BaseModel, Field class _Out(BaseModel): summary: str acceptance_criteria: list[str] = Field(default_factory=list) clarification_questions: list[ClarificationQuestion] = Field(default_factory=list) is_clear: bool confidence: float referenced_documentation: list[str] = Field(default_factory=list) out_of_scope_items: list[str] = Field(default_factory=list) notes: str | None = None llm = self._chat().with_structured_output(_Out) result: _Out = await llm.ainvoke(prompt) jira_key = _extract(prompt, r"Jira key:\s*(\S+)", "UNKNOWN") return RequirementsArtifact( jira_key=jira_key, summary=result.summary, acceptance_criteria=result.acceptance_criteria, clarification_questions=result.clarification_questions, is_clear=result.is_clear, confidence=result.confidence, referenced_documentation=result.referenced_documentation, out_of_scope_items=result.out_of_scope_items, notes=result.notes, ) async def design_solution(self, prompt: str) -> DesignArtifact: from pydantic import BaseModel, Field class _Item(BaseModel): metadata_type: str api_name: str action: str = "create" object: str | None = None notes: str | None = None class _Out(BaseModel): summary: str approach: str metadata_plan: list[_Item] = Field(default_factory=list) dependencies: list[str] = Field(default_factory=list) risks_or_escalations: list[str] = Field(default_factory=list) out_of_allowlist: bool = False confidence: float = 0.7 notes: str | None = None llm = self._chat().with_structured_output(_Out) result: _Out = await llm.ainvoke(prompt) jira_key = _extract(prompt, r"Jira key:\s*(\S+)", "UNKNOWN") return DesignArtifact( jira_key=jira_key, summary=result.summary, approach=result.approach, metadata_plan=[ MetadataPlanItem( metadata_type=i.metadata_type, api_name=i.api_name, action=i.action, object=i.object, notes=i.notes, ) for i in result.metadata_plan ], dependencies=result.dependencies, risks_or_escalations=result.risks_or_escalations, out_of_allowlist=result.out_of_allowlist, confidence=result.confidence, notes=result.notes, ) async def review_code(self, prompt: str) -> ReviewArtifact: from pydantic import BaseModel, Field class _Finding(BaseModel): severity: str category: str file: str | None = None line: int | None = None message: str fixable: bool = True class _Out(BaseModel): passed: bool cycle: int findings: list[_Finding] = Field(default_factory=list) summary: str static_analysis_status: str = "passed" notes: str | None = None llm = self._chat().with_structured_output(_Out) result: _Out = await llm.ainvoke(prompt) jira_key = _extract(prompt, r"Jira key:\s*(\S+)", "UNKNOWN") return ReviewArtifact( jira_key=jira_key, passed=result.passed, cycle=result.cycle, findings=[ ReviewFinding( severity=(f.severity or "medium").strip().lower(), category=f.category, file=f.file, line=f.line, message=f.message, fixable=f.fixable, ) for f in result.findings ], summary=result.summary, static_analysis_status=result.static_analysis_status, notes=result.notes, ) async def fix_code(self, prompt: str) -> Dict[str, str]: from pydantic import BaseModel, Field class _FilePatch(BaseModel): path: str content: str class _Out(BaseModel): files: list[_FilePatch] = Field(default_factory=list) llm = self._chat().with_structured_output(_Out) result: _Out = await llm.ainvoke(prompt) return {f.path: f.content for f in result.files} async def generate_qa_report(self, prompt: str) -> QAReportArtifact: from pydantic import BaseModel, Field class _Scenario(BaseModel): name: str type: str description: str acceptance_criterion: str | None = None class _Out(BaseModel): passed: bool test_matrix: list[_Scenario] = Field(default_factory=list) apex_tests_run: int = 0 apex_tests_passed: int = 0 apex_failures: list[str] = Field(default_factory=list) manual_scenarios: list[str] = Field(default_factory=list) summary: str notes: str | None = None llm = self._chat().with_structured_output(_Out) result: _Out = await llm.ainvoke(prompt) jira_key = _extract(prompt, r"Jira key:\s*(\S+)", "UNKNOWN") return QAReportArtifact( jira_key=jira_key, passed=result.passed, test_matrix=[ TestScenario( name=s.name, type=(s.type or "manual").strip().lower(), # type: ignore[arg-type] description=s.description, acceptance_criterion=s.acceptance_criterion, ) for s in result.test_matrix ], apex_tests_run=result.apex_tests_run, apex_tests_passed=result.apex_tests_passed, apex_failures=result.apex_failures, manual_scenarios=result.manual_scenarios, summary=result.summary, notes=result.notes, ) def get_llm_client(settings: Optional[Settings] = None) -> LLMClient: settings = settings or get_settings() if settings.llm_provider == "openai": return OpenAILLMClient(settings) return MockLLMClient() def _extract(text: str, pattern: str, default: str) -> str: m = re.search(pattern, text) return m.group(1) if m else default _STOPWORDS = { "the", "a", "an", "and", "or", "of", "on", "to", "for", "with", "in", "add", "new", "test", "agent", "nova", } def _significant_tokens(text: str) -> set[str]: raw = re.findall(r"[a-z0-9_]{3,}", (text or "").lower()) tokens: set[str] = set() for t in raw: if t not in _STOPWORDS: tokens.add(t) for part in t.split("_"): if len(part) >= 3 and part not in _STOPWORDS: tokens.add(part) return tokens def _section(text: str, header: str) -> str: idx = text.find(header) if idx < 0: return "" rest = text[idx + len(header) :] # Stop at the next top-level "Title:" line (same style as our prompt sections). parts = re.split( r"\n(?:Description|Acceptance criteria|Comments|Attachments|Confluence hits|Repository hits|v1 allowlist|Out of scope from analyst|Notes|Confidence from analyst):", rest, maxsplit=1, ) return parts[0].strip() def _refs(prompt: str) -> list[str]: refs = [] if "Confluence hits:" in prompt: block = _section(prompt, "Confluence hits:") for line in block.splitlines(): if line.strip().startswith("-"): refs.append(line.strip("- ").strip()) if "Repository hits:" in prompt: block = _section(prompt, "Repository hits:") for line in block.splitlines(): if line.strip().startswith("-"): refs.append(line.strip("- ").strip()) return refs def _parse_file_blocks(block: str) -> Dict[str, str]: """Parse '### path\\n```\\ncontent\\n```' sections from prompt blocks.""" files: Dict[str, str] = {} if not block.strip(): return files parts = re.split(r"^###\s+(.+)$", block, flags=re.MULTILINE) i = 1 while i < len(parts) - 1: path = parts[i].strip() body = parts[i + 1] m = re.search(r"```(?:\w+)?\n(.*?)```", body, re.DOTALL) if m: files[path] = m.group(1).rstrip("\n") + "\n" i += 2 return files