290 lines
11 KiB
Python
290 lines
11 KiB
Python
"""Generate context files from discovery answers.
|
|
|
|
Uses the mapping defined in context/discovery-context-mapping.md to transform
|
|
discovery-log.md entries into the appropriate context files.
|
|
|
|
Only processes High/Medium confidence answers. Low-confidence answers are
|
|
skipped and must be handled by `rdb ask-more` first.
|
|
|
|
Preserves existing content in context files - never overwrites.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from .discovery import read_discovery_answers
|
|
|
|
|
|
# Mapping from question ID to list of (file_path, section_name, format_type)
|
|
# format_type: "body" = fill under heading with body text
|
|
# "table" = append row to table-based section
|
|
# "append" = append as a new section at end of file
|
|
CONTEXT_MAP = {
|
|
"Q-001": [ # What problem are we solving?
|
|
("context/product-brief.md", "## Problem", "body"),
|
|
("context/project-brief.md", "## Problem Statement", "body"),
|
|
],
|
|
"Q-002": [ # Who is the user?
|
|
("context/product-brief.md", "## Users", "body"),
|
|
("context/development-context.md", "## IDEs and Editors", "body"),
|
|
],
|
|
"Q-003": [ # What does success look like?
|
|
("context/product-brief.md", "## Success Criteria", "body"),
|
|
("context/project-brief.md", "## Success Metrics", "body"),
|
|
],
|
|
"Q-004": [ # What is the minimum useful version?
|
|
("context/product-brief.md", "## Minimum Useful Version", "body"),
|
|
("context/project-brief.md", "## Key Features (MVP)", "body"),
|
|
],
|
|
"Q-005": [ # What data do we need?
|
|
("context/architecture.md", "## Data Flow", "body"),
|
|
("context/development-context.md", "## Dependencies", "body"),
|
|
],
|
|
"Q-006": [ # What systems must it connect to?
|
|
("context/architecture.md", "## External Integrations", "body"),
|
|
("context/infrastructure-context.md", None, "append"),
|
|
],
|
|
"Q-007": [ # What are the risks?
|
|
("context/risks.md", "# Risks", "table"),
|
|
],
|
|
"Q-008": [ # What must not happen?
|
|
("context/assumptions.md", "# Assumptions", "table"),
|
|
],
|
|
"Q-009": [ # How will we test it?
|
|
("context/development-context.md", "## Build & Test", "body"),
|
|
("context/project-brief.md", "## Timeline & Milestones", "body"),
|
|
],
|
|
"Q-010": [ # How will it be deployed?
|
|
("context/architecture.md", "## Deployment Architecture", "body"),
|
|
("context/infrastructure-context.md", None, "append"),
|
|
],
|
|
"Q-011": [ # Who owns and directs the project, and what are its organisational goals?
|
|
("context/company-context.md", "## Mission", "body"),
|
|
("context/project-brief.md", "## Target Audience", "body"),
|
|
],
|
|
"Q-012": [ # What is the technology stack, repository structure, and architecture approach?
|
|
("context/development-context.md", "## Tech Stack", "body"),
|
|
("context/development-context.md", "## Coding Standards", "body"),
|
|
("context/architecture.md", "## Overview", "body"),
|
|
("context/architecture.md", "## Core Components", "body"),
|
|
],
|
|
"Q-013": [ # Where is it hosted, which environments exist, and what are the security requirements?
|
|
("context/infrastructure-context.md", "## Hosting", "body"),
|
|
("context/infrastructure-context.md", "## Environments", "body"),
|
|
("context/infrastructure-context.md", "## Monitoring & Alerting", "body"),
|
|
("context/infrastructure-context.md", "## Security", "body"),
|
|
],
|
|
"Q-014": [ # What agent and developer workflow tools, constraints, and guidelines apply?
|
|
("context/agent-guidelines.md", "## Purpose", "body"),
|
|
("context/agent-guidelines.md", "## Preferred Tools", "body"),
|
|
("context/repository-context.md", "## Purpose", "body"),
|
|
("context/repository-context.md", "## Contributing", "body"),
|
|
],
|
|
}
|
|
|
|
|
|
def generate_context_files(
|
|
root: Path, min_confidence: str = "Medium"
|
|
) -> dict[str, list[str]]:
|
|
"""Generate context files from discovery answers.
|
|
|
|
Args:
|
|
root: Project root path.
|
|
min_confidence: Minimum confidence to process (High, Medium).
|
|
|
|
Returns:
|
|
Dict with keys: 'generated', 'skipped' each a list of strings.
|
|
"""
|
|
valid_confidences = {"High", "Medium"}
|
|
answers = read_discovery_answers(root)
|
|
|
|
if not answers:
|
|
return {"generated": [], "skipped": ["No discovery answers found in discovery-log.md"]}
|
|
|
|
filtered = [a for a in answers if a["confidence"] in valid_confidences]
|
|
skipped = [a for a in answers if a["confidence"] not in valid_confidences]
|
|
|
|
generated_files: set[str] = set()
|
|
|
|
for answer in filtered:
|
|
qid = answer["id"]
|
|
rules = CONTEXT_MAP.get(qid)
|
|
if not rules:
|
|
skipped.append(f"{qid} (no mapping)")
|
|
continue
|
|
|
|
for file_path, section_name, fmt in rules:
|
|
_write_section(
|
|
root=root,
|
|
file_path=file_path,
|
|
section_name=section_name,
|
|
format_type=fmt,
|
|
answer=answer,
|
|
rules=rules,
|
|
)
|
|
generated_files.add(file_path)
|
|
|
|
skipped_ids = [a["id"] for a in skipped]
|
|
if skipped_ids:
|
|
skipped = ["Low-confidence or flagged answers (skipped): " + ", ".join(skipped_ids)]
|
|
|
|
return {"generated": sorted(generated_files), "skipped": skipped}
|
|
|
|
|
|
def _write_section(root, file_path, section_name, format_type, answer, rules):
|
|
"""Dispatch to the appropriate write strategy."""
|
|
target = root / file_path
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
if format_type == "table":
|
|
_write_table_row(target, section_name, answer)
|
|
elif format_type == "append":
|
|
_append_new_section(target, answer, rules)
|
|
else:
|
|
# body text type
|
|
existing = target.read_text(encoding="utf-8") if target.exists() else ""
|
|
lines = existing.splitlines() if existing else []
|
|
|
|
section_idx = _find_section(lines, section_name) if section_name else -1
|
|
|
|
if section_idx >= 0:
|
|
_fill_or_append_body(lines, target, section_idx, answer)
|
|
else:
|
|
_create_new_section(target, section_name, answer, format_type)
|
|
|
|
|
|
def _find_section(lines, section_header):
|
|
"""Find the index of a section header line in the markdown."""
|
|
for i, line in enumerate(lines):
|
|
if line.strip() == section_header:
|
|
return i
|
|
return -1
|
|
|
|
|
|
def _fill_or_append_body(lines, target, section_idx, answer):
|
|
"""Fill or append body text under an existing section header.
|
|
|
|
Preserves the rest of the document - never drops subsequent sections.
|
|
"""
|
|
start = section_idx + 1
|
|
|
|
# Find another section header to know the boundary
|
|
end = len(lines)
|
|
for i in range(start, len(lines)):
|
|
if lines[i].startswith("## "):
|
|
end = i
|
|
break
|
|
|
|
content_lines = [l.strip() for l in lines[start:end] if l.strip()]
|
|
|
|
# Check if section is empty or only has TBD placeholder
|
|
is_tbd = (
|
|
len(content_lines) == 1 and content_lines[0].lower().startswith("tbd")
|
|
) or len(content_lines) == 0
|
|
|
|
answer_text = _format_answer_text(answer)
|
|
|
|
if is_tbd:
|
|
# Replace TBD with discovery answer, keep rest of document
|
|
new_lines = (
|
|
lines[:section_idx + 1]
|
|
+ [answer_text, ""]
|
|
+ lines[end:]
|
|
)
|
|
else:
|
|
# Append under existing content, after the last non-empty line in this section
|
|
insert_at = start
|
|
for i in range(start, end):
|
|
if lines[i].strip():
|
|
insert_at = i + 1
|
|
new_lines = (
|
|
lines[:insert_at]
|
|
+ [answer_text, ""]
|
|
+ lines[insert_at:]
|
|
)
|
|
|
|
target.write_text("\n".join(new_lines), encoding="utf-8")
|
|
|
|
|
|
def _write_table_row(target, section_header, answer):
|
|
"""Append a row to a table-based context file."""
|
|
existing = target.read_text(encoding="utf-8") if target.exists() else ""
|
|
|
|
q_num = answer["id"].split("-")[1]
|
|
all_questions = [
|
|
"What problem are we solving?",
|
|
"Who is the user?",
|
|
"What does success look like?",
|
|
"What is the minimum useful version?",
|
|
"What data do we need?",
|
|
"What systems must it connect to?",
|
|
"What are the risks?",
|
|
"What must not happen?",
|
|
"How will we test it?",
|
|
"How will it be deployed?",
|
|
"Who owns and directs the project, and what are its organisational goals?",
|
|
"What is the technology stack, repository structure, and architecture approach?",
|
|
"Where is it hosted, which environments exist, and what are the security requirements?",
|
|
"What agent and developer workflow tools, constraints, and guidelines apply?",
|
|
]
|
|
question_text = all_questions[int(q_num) - 1] if int(q_num) <= len(all_questions) else answer["question"]
|
|
|
|
fname = target.name
|
|
if fname == "risks.md":
|
|
risk_id = f"RISK-{q_num}"
|
|
clean = answer["answer"].replace("\\|", "|").strip()
|
|
row = f"| {risk_id} | {question_text}: {clean} | Medium | Monitor and review | Open |\n"
|
|
elif fname == "assumptions.md":
|
|
assump_id = f"ASSUMPTION-{q_num}"
|
|
clean = answer["answer"].replace("\\|", "|").strip()
|
|
row = f"| {assump_id} | {question_text}: {clean} | Medium | Yes |\n"
|
|
else:
|
|
row = f"| {answer['id']} | {question_text} | {answer['answer']} | Open |\n"
|
|
|
|
target.write_text(existing + "\n" + row, encoding="utf-8")
|
|
|
|
|
|
def _append_new_section(target, answer, rules):
|
|
"""Append a new section to the end of an existing file."""
|
|
existing = target.read_text(encoding="utf-8") if target.exists() else ""
|
|
|
|
display_name = None
|
|
for _, sec, _ in rules:
|
|
if sec and sec.startswith("## "):
|
|
display_name = sec
|
|
break
|
|
if not display_name:
|
|
q_num = answer["id"].split("-")[1]
|
|
display_name = f"## Q-{q_num} Discovery Answer"
|
|
|
|
section_text = (
|
|
f"\n{display_name}\n\n"
|
|
f"**Discovery Question:** {answer['question']}\n\n"
|
|
f"**Answer:** {answer['answer'].strip()}\n\n"
|
|
f"**Confidence:** {answer['confidence']}\n"
|
|
)
|
|
target.write_text(existing + section_text, encoding="utf-8")
|
|
|
|
|
|
def _create_new_section(target, header, answer, fmt):
|
|
"""Create a new section in a context file when it doesn't exist yet."""
|
|
existing = target.read_text(encoding="utf-8") if target.exists() else ""
|
|
|
|
if not header:
|
|
q_num = answer["id"].split("-")[1]
|
|
header = f"## Q-{q_num} Discovery Answer"
|
|
|
|
section_text = (
|
|
f"\n{header}\n\n"
|
|
f"**Discovery Question:** {answer['question']}\n\n"
|
|
f"**Answer:** {answer['answer'].strip()}\n\n"
|
|
f"**Confidence:** {answer['confidence']}\n"
|
|
)
|
|
target.write_text(existing + section_text, encoding="utf-8")
|
|
|
|
|
|
def _format_answer_text(answer):
|
|
"""Format a discovery answer into readable body text."""
|
|
return answer["answer"].replace("\\|", "|").strip()
|