Plan Parser

Plan file parsing for task generation.

Parse implementation plan files into structured task definitions.

When an agent completes a task that results in an implementation plan (written to .claude/plan.md or a similar file in the workspace), this module reads the plan file, parses its markdown content, and extracts individual steps that can be turned into follow-up tasks.

Classes

PlanQualityReport dataclass

PlanQualityReport(is_design_doc: bool, is_implementation_plan: bool, quality_score: float, total_sections: int, actionable_sections: int, filtered_sections: int, actionable_ratio: float, warnings: list[str] = list(), recommendation: str = '')

Quality assessment of a parsed plan document.

PlanStep dataclass

PlanStep(title: str, description: str, priority_hint: int = 0, raw_title: str = '')

A single actionable step extracted from an implementation plan.

ParsedPlan dataclass

ParsedPlan(source_file: str, steps: list[PlanStep] = list(), raw_content: str = '')

The result of parsing a plan file.

Functions

find_plan_file

find_plan_file(workspace: str, patterns: list[str] | None = None) -> str | None

Search for a plan file in the workspace directory.

Checks each candidate pattern in order. Patterns that contain glob characters (* or ?) are expanded; the most-recently-modified match is returned so that freshly written plans take priority.

For plain (non-glob) patterns the file is checked directly.

If no pattern matches, falls back to a deep scan that looks for recently-modified markdown files with plan-like structure indicators (e.g. headings containing "Phase 1:", "Step 2:", etc.).

Returns the first match, or None if no plan file is found.

Source code in src/plan_parser.py

def find_plan_file(workspace: str, patterns: list[str] | None = None) -> str | None:
    """Search for a plan file in the workspace directory.

    Checks each candidate pattern in order.  Patterns that contain glob
    characters (``*`` or ``?``) are expanded; the most-recently-modified
    match is returned so that freshly written plans take priority.

    For plain (non-glob) patterns the file is checked directly.

    If no pattern matches, falls back to a deep scan that looks for
    recently-modified markdown files with plan-like structure indicators
    (e.g. headings containing "Phase 1:", "Step 2:", etc.).

    Returns the first match, or ``None`` if no plan file is found.
    """
    candidates = patterns or DEFAULT_PLAN_FILE_PATTERNS
    for pattern in candidates:
        full_pattern = os.path.join(workspace, pattern)
        if any(c in pattern for c in ("*", "?")):
            # Glob pattern — expand and pick the newest match
            matches = [p for p in glob.glob(full_pattern) if os.path.isfile(p)]
            if matches:
                # Return the most-recently-modified file so the latest plan
                # wins when multiple plan files exist (e.g. date-prefixed).
                matches.sort(key=lambda p: os.path.getmtime(p), reverse=True)
                return matches[0]
        else:
            if os.path.isfile(full_pattern):
                return full_pattern

    # Fallback: deep scan for recently-modified markdown files with plan indicators
    result = _deep_scan_for_plan(workspace)
    if result:
        print(
            f"Plan file discovery: found plan via deep scan at {result} "
            f"(not in standard patterns: {candidates})"
        )
    return result

read_plan_file

read_plan_file(path: str) -> str

Read the raw contents of a plan file.

Source code in src/plan_parser.py

def read_plan_file(path: str) -> str:
    """Read the raw contents of a plan file."""
    with open(path, "r", encoding="utf-8") as f:
        return f.read()

parse_plan

parse_plan(content: str, source_file: str = '', max_steps: int = 5) -> ParsedPlan

Parse plan markdown content into structured steps.

Supports several common plan formats:

1. Implementation-section plans — Large design documents with a dedicated ## Implementation Plan or similar section containing ### sub-headings for each phase. Only the sub-headings within the implementation section become steps.

2. Heading-based plans — Each ## Section or ### Section becomes a step, with the text underneath as the description.

3. Numbered-list plans — Top-level numbered items (1. ..., 2. ...) become individual steps with the item text and any indented sub-items as the description.

4. Mixed — If both heading sections and numbered lists are present, heading sections take priority.

Parameters:

Name	Type	Description	Default
content	str	Raw markdown content of the plan file.	required
source_file	str	Path to the plan file (for traceability).	''
max_steps	int	Maximum number of steps to return (truncated if exceeded).	5

Returns a ParsedPlan containing the ordered list of steps.

Source code in src/plan_parser.py

def parse_plan(
    content: str, source_file: str = "", max_steps: int = 5,
) -> ParsedPlan:
    """Parse plan markdown content into structured steps.

    Supports several common plan formats:

    1. **Implementation-section plans** — Large design documents with a
       dedicated ``## Implementation Plan`` or similar section containing
       ``###`` sub-headings for each phase. Only the sub-headings within
       the implementation section become steps.

    2. **Heading-based plans** — Each ``## Section`` or ``### Section``
       becomes a step, with the text underneath as the description.

    3. **Numbered-list plans** — Top-level numbered items (``1. ...``,
       ``2. ...``) become individual steps with the item text and any
       indented sub-items as the description.

    4. **Mixed** — If both heading sections and numbered lists are present,
       heading sections take priority.

    Args:
        content: Raw markdown content of the plan file.
        source_file: Path to the plan file (for traceability).
        max_steps: Maximum number of steps to return (truncated if exceeded).

    Returns a ``ParsedPlan`` containing the ordered list of steps.
    """
    result = ParsedPlan(source_file=source_file, raw_content=content)

    if not content.strip():
        return result

    # Try implementation-section parsing first (for design documents)
    steps = _parse_implementation_section(content)
    if steps:
        result.steps = steps[:max_steps]
        return result

    # Try heading-based parsing
    steps = _parse_heading_sections(content)
    if steps:
        # Quality check: if too many non-actionable steps were extracted,
        # try to filter to only step/phase headings
        quality = _score_parse_quality(steps)
        if quality < 0.4 and len(steps) > 5:
            # Re-filter: keep only headings that look like implementation phases
            filtered = [
                s for s in steps
                if STEP_HEADING_PATTERN.match(s.title)
                or _is_likely_actionable(s.title)
            ]
            if filtered:
                steps = filtered
                print(
                    f"Plan parser: quality score {quality:.2f} too low, "
                    f"filtered {len(steps)} actionable steps from headings"
                )
        result.steps = steps[:max_steps]
        return result

    # Fall back to numbered-list parsing
    steps = _parse_numbered_list(content)
    if steps:
        result.steps = steps[:max_steps]
        return result

    # Final fallback: treat the entire content as a single step
    # (only if there's meaningful content beyond a title)
    title = _extract_title(content)
    body = _remove_title(content).strip()
    if body:
        result.steps = [PlanStep(title=title or "Implementation Plan", description=body)]

    return result

build_task_description

build_task_description(step: PlanStep, parent_task: object | None = None, plan_context: str = '') -> str

Build a self-contained task description from a plan step.

The description is enriched with context from the parent task and the overall plan so that the executing agent has all the information it needs without access to external context.

When the step description contains multiple sub-steps (e.g. from consolidation or a well-structured phase), a high-level outline is extracted and presented prominently so the agent can see all the work in the phase at a glance.

Parameters:

Name	Type	Description	Default
step	PlanStep	The parsed plan step.	required
parent_task	object | None	The parent task object (must have .title and .description).	None
plan_context	str	Additional context from the plan preamble.	''

Returns:

Type	Description
str	A self-contained description string.

Source code in src/plan_parser.py

def build_task_description(
    step: PlanStep,
    parent_task: object | None = None,
    plan_context: str = "",
) -> str:
    """Build a self-contained task description from a plan step.

    The description is enriched with context from the parent task and
    the overall plan so that the executing agent has all the information
    it needs without access to external context.

    When the step description contains multiple sub-steps (e.g. from
    consolidation or a well-structured phase), a high-level outline
    is extracted and presented prominently so the agent can see all
    the work in the phase at a glance.

    Args:
        step: The parsed plan step.
        parent_task: The parent task object (must have .title and .description).
        plan_context: Additional context from the plan preamble.

    Returns:
        A self-contained description string.
    """
    parts: list[str] = []

    parts.append(f"**{step.title}**\n")

    if plan_context:
        parts.append(f"## Background Context\n{plan_context}\n")

    if parent_task and hasattr(parent_task, "title"):
        parts.append(
            f"This task is part of the implementation plan from: "
            f"**{parent_task.title}**\n"
        )

    # Extract a high-level outline of steps if the description contains
    # sub-step structure (### headings or numbered items).
    outline = _extract_steps_outline(step.description)
    if outline:
        parts.append(f"## High-Level Steps\n{outline}\n")

    parts.append(f"## Task Details\n{step.description}")

    return "\n".join(parts)

validate_plan_quality

validate_plan_quality(content: str) -> PlanQualityReport

Assess the quality of a plan document for task splitting.

Analyzes the markdown content to determine whether it is an actionable implementation plan or a design/reference document, and scores it accordingly.

Source code in src/plan_parser.py

def validate_plan_quality(content: str) -> PlanQualityReport:
    """Assess the quality of a plan document for task splitting.

    Analyzes the markdown content to determine whether it is an actionable
    implementation plan or a design/reference document, and scores it
    accordingly.
    """
    if not content.strip():
        return PlanQualityReport(
            is_design_doc=False,
            is_implementation_plan=False,
            quality_score=0.0,
            total_sections=0,
            actionable_sections=0,
            filtered_sections=0,
            actionable_ratio=0.0,
            warnings=[],
            recommendation="Empty document — nothing to split.",
        )

    # Extract all ## and ### headings (skip those inside code fences)
    heading_pattern = re.compile(r"^#{2,3}\s+(.+)$", re.MULTILINE)
    fence_ranges = _fenced_code_ranges(content)
    headings = [
        m.group(1).strip() for m in heading_pattern.finditer(content)
        if not _is_inside_code_fence(m.start(), fence_ranges)
    ]
    total_sections = len(headings)

    if total_sections == 0:
        return PlanQualityReport(
            is_design_doc=False,
            is_implementation_plan=False,
            quality_score=0.0,
            total_sections=0,
            actionable_sections=0,
            filtered_sections=0,
            actionable_ratio=0.0,
            warnings=[],
            recommendation="No headings found — cannot split into tasks.",
        )

    # Classify each heading
    actionable_count = 0
    design_indicators = 0
    for heading in headings:
        clean = _clean_step_title(heading)
        clean_lower = clean.lower()

        if clean_lower in NON_ACTIONABLE_HEADINGS:
            design_indicators += 1
            continue

        if STEP_HEADING_PATTERN.match(clean) or _is_likely_actionable(clean):
            actionable_count += 1
        else:
            # Check for informational keywords
            words = set(clean_lower.split())
            if words & _INFORMATIONAL_KEYWORDS:
                design_indicators += 1

    filtered = total_sections - actionable_count
    actionable_ratio = round(actionable_count / total_sections, 2) if total_sections else 0.0

    # Determine document type
    is_design_doc = design_indicators > actionable_count and actionable_count <= 2
    is_implementation_plan = actionable_count >= 2 or (
        actionable_count >= 1 and actionable_ratio >= 0.3
    )

    # Quality score: proportion of actionable sections, boosted by step patterns
    quality_score = actionable_ratio
    if any(STEP_HEADING_PATTERN.match(_clean_step_title(h)) for h in headings):
        quality_score = min(1.0, quality_score + 0.2)

    # Warnings
    warnings: list[str] = []
    if is_design_doc:
        warnings.append(
            "Document appears to be a design/reference document rather than "
            "an implementation plan."
        )
    if actionable_count > 15:
        warnings.append(
            f"High step count ({actionable_count}) — consider consolidating "
            f"related steps."
        )

    # Recommendation
    if is_design_doc:
        recommendation = (
            "This looks like a design document. Consider extracting only the "
            "implementation sections into a separate plan."
        )
    elif is_implementation_plan and quality_score >= 0.5:
        recommendation = "Good implementation plan — suitable for automatic task splitting."
    elif is_implementation_plan:
        recommendation = (
            "Implementation plan detected but quality is moderate. "
            "Review generated tasks for relevance."
        )
    else:
        recommendation = "Unable to identify clear implementation steps."

    return PlanQualityReport(
        is_design_doc=is_design_doc,
        is_implementation_plan=is_implementation_plan,
        quality_score=quality_score,
        total_sections=total_sections,
        actionable_sections=actionable_count,
        filtered_sections=filtered,
        actionable_ratio=actionable_ratio,
        warnings=warnings,
        recommendation=recommendation,
    )

parse_and_generate_steps

parse_and_generate_steps(content: str, *, max_steps: int = 5, enforce_quality: bool = False, min_quality_score: float = 0.3) -> tuple[list[dict], PlanQualityReport]

Parse a plan document and return steps as dicts with a quality report.

This is the main orchestrator integration point. It combines parse_plan() with validate_plan_quality() and applies an additional post-filter to remove any remaining non-actionable steps.

Parameters:

Name	Type	Description	Default
content	str	Raw markdown content.	required
max_steps	int	Maximum number of steps to return.	5
enforce_quality	bool	If True, return no steps when quality is below threshold.	False
min_quality_score	float	Minimum quality score (used when enforce_quality=True).	0.3

Returns:

Type	Description
list[dict]	A tuple of (steps, quality_report) where each step is a dict with
PlanQualityReport	title and description keys.

Source code in src/plan_parser.py

def parse_and_generate_steps(
    content: str,
    *,
    max_steps: int = 5,
    enforce_quality: bool = False,
    min_quality_score: float = 0.3,
) -> tuple[list[dict], PlanQualityReport]:
    """Parse a plan document and return steps as dicts with a quality report.

    This is the main orchestrator integration point.  It combines
    ``parse_plan()`` with ``validate_plan_quality()`` and applies an
    additional post-filter to remove any remaining non-actionable steps.

    Args:
        content: Raw markdown content.
        max_steps: Maximum number of steps to return.
        enforce_quality: If True, return no steps when quality is below threshold.
        min_quality_score: Minimum quality score (used when enforce_quality=True).

    Returns:
        A tuple of (steps, quality_report) where each step is a dict with
        ``title`` and ``description`` keys.
    """
    quality = validate_plan_quality(content)

    if enforce_quality and quality.quality_score < min_quality_score:
        return [], quality

    parsed = parse_plan(content, max_steps=max_steps)

    # Post-filter: remove any remaining non-actionable steps
    filtered_steps: list[dict] = []
    for step in parsed.steps:
        title_lower = step.title.lower()
        if title_lower in NON_ACTIONABLE_HEADINGS:
            continue
        # Use raw_title (original heading) when available for fidelity
        title = step.raw_title if step.raw_title else step.title
        filtered_steps.append({
            "title": title,
            "description": step.description,
        })

    # Enforce safety cap
    filtered_steps = filtered_steps[:max_steps]

    return filtered_steps, quality