orchestrata_ultimate.planning.plan_parser

Reference › orchestrata_ultimate › planning › plan_parser

plan_parser parses the project's execution-plan markdown into structured steps (SPEC-107 FR-001), so that the plan-context provider can show the current, upcoming and completed steps without re-reading the markdown on every volk call. Its importers are kernel.dispatcher, session.controller._goals and the package facade; tests are in tests/planning/test_plan_parser.py.

parse_plan_text splits the text at phase headings (_split_into_phases), finds step headings of the form Step N: title within each phase (_parse_phase_steps), and reads the labelled fields in each step body (_extract_step_body_fields, _parse_actions): goal, file, actions, verify, expected and the fallback. Each step becomes a frozen PlanStep, with to_dict for serialisation. parse_plan_file reads a file; the path is resolved by _resolve_plan_path from an explicit argument, then the environment variable ORCHESTRATA_PLAN_PATH, then plans/execution-plan.md. find_step looks a step up by number.

For constitutional Article XII (plan fidelity, RUL-015), parse_plan_status extracts a plan's own bold status marker in either of the two conventions the project's plans use. is_plan_active treats a plan as active unless the status carries a terminal word such as done, complete, superseded, closed, archived or cancelled. load_plan_governance_facts combines these into a PlanGovernanceFacts record (plan id, active flag, status, steps) and raises FileNotFoundError when the resolved plan file does not exist.

At a glance

Class diagram

Dependency diagram

API

Classes

PlanStep (dataclass)

Decorators: dataclass(frozen=True) · Line 22

One structured step from an execution plan.

PlanGovernanceFacts (dataclass)

Decorators: traces(spec='007', story='US-002', reqs=['REQ-007-US-002-001']), dataclass(frozen=True) · Line 321

Ground facts for constitutional Article XII (plan fidelity, RUL-015), derived from a REAL, parsed execution plan -- never fabricated.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/planning/plan_parser.py

"""Parse the project's execution plan markdown into structured steps.

Spec: SPEC-107 FR-001. The agent needs a compact, machine-readable view of the
active plan so that the `plan_context` provider can render the current step,
upcoming steps, and recently completed steps without re-parsing the markdown on
every volk call.
"""

from __future__ import annotations

import os
import re
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces


@dataclass(frozen=True)
class PlanStep:
    """One structured step from an execution plan.

    Attributes:
        step_num: Global step number as shown in the plan (e.g. 1, 2, 42).
        title: Short step title from the heading.
        phase: Name of the phase that contains the step, if known.
        goal: The human-readable goal of the step.
        file: Target file path, if the step names one.
        actions: Ordered action items from the step body.
        verify: Verification command or procedure.
        expected: Expected result / output.
        fallback: Guidance for what to do if verification fails.
    """

    step_num: int
    title: str
    phase: str = ""
    goal: str = ""
    file: str = ""
    actions: tuple[str, ...] = ()
    verify: str = ""
    expected: str = ""
    fallback: str = ""

    def to_dict(self) -> dict[str, Any]:
        """Return a plain dict representation for serialization."""
        return {
            "step_num": self.step_num,
            "title": self.title,
            "phase": self.phase,
            "goal": self.goal,
            "file": self.file,
            "actions": list(self.actions),
            "verify": self.verify,
            "expected": self.expected,
            "fallback": self.fallback,
        }


# Phase headings: "## Phase 0: Foundation (Steps 1–38)"
_PHASE_RE = re.compile(r"^##\s+(.+?)$", re.MULTILINE)
# Step headings: "### Step 1: Fix wrong import path in logging_config.py"
_STEP_RE = re.compile(r"^###\s+Step\s+(\d+)\s*:\s*(.+?)$", re.MULTILINE)
# Field tags used inside a step body.
_FIELD_TAGS = {
    "goal": "**Goal:**",
    "file": "**File:**",
    "actions": "**Actions:**",
    "verify": "**Verify:**",
    "expect": "**Expect:**",
    "expected": "**Expected:**",
    "fallback": "**If it fails:**",
}


@traces(spec="107", story="FR-001", reqs=["REQ-107-060"])
def _split_into_phases(text: str) -> list[tuple[str, str]]:
    """Split plan text into (phase_name, phase_body) chunks."""
    matches = list(_PHASE_RE.finditer(text))
    if not matches:
        return [("", text)]

    chunks: list[tuple[str, str]] = []
    for i, match in enumerate(matches):
        phase_name = match.group(1).strip()
        start = match.end()
        end = matches[i + 1].start() if i + 1 < len(matches) else len(text)
        chunks.append((phase_name, text[start:end]))
    return chunks


@traces(spec="107", story="FR-001", reqs=["REQ-107-060"])
def _extract_step_body_fields(body: str) -> dict[str, str]:
    """Parse the labeled fields inside a single step body.

    The execution plan uses bold labels such as ``**Goal:**`` to delimit
    sections. This function captures the text between two such labels, or
    between a label and the end of the body / next ``---`` separator.
    """
    # Build an ordered list of (field_key, start_pos) for known tags.
    positions: list[tuple[str, int]] = [
        (key, match.start())
        for key, tag in _FIELD_TAGS.items()
        for match in re.finditer(re.escape(tag), body)
    ]

    # Also stop capture at the next horizontal rule.
    rule_matches = list(re.finditer(r"^---\s*$", body, re.MULTILINE))
    rule_positions = [m.start() for m in rule_matches]

    positions.sort(key=lambda item: item[1])

    fields: dict[str, str] = {}
    for idx, (key, start) in enumerate(positions):
        # Value starts after the tag itself.
        value_start = start + len(_FIELD_TAGS[key])
        # Value ends at the next label, the next rule, or end of body.
        end_candidates = [p for p in rule_positions if p > start]
        if idx + 1 < len(positions):
            end_candidates.append(positions[idx + 1][1])
        value_end = min(end_candidates) if end_candidates else len(body)
        value = body[value_start:value_end].strip()
        # Deduplicate: prefer the first occurrence of a canonical key.
        canonical = "expected" if key == "expect" else key
        if canonical not in fields or not fields[canonical]:
            fields[canonical] = value
    return fields


@traces(spec="107", story="FR-001", reqs=["REQ-107-060"])
def _parse_actions(actions_block: str) -> tuple[str, ...]:
    """Split an actions markdown block into individual action strings.

    Actions are usually a numbered list (``1. Open file...``) or, less often, a
    bulleted list. We keep each item's full text but collapse internal newlines
    to a single space for compactness.
    """
    items: list[str] = []
    for line in actions_block.splitlines():
        stripped = line.strip()
        if not stripped:
            continue
        # Numbered or bullet item.
        if re.match(r"^(\d+\.|[-*])\s+", stripped):
            items.append(re.sub(r"^(\d+\.|[-*])\s+", "", stripped))
        elif items:
            # Continuation line for the previous item.
            items[-1] = (items[-1] + " " + stripped).strip()
    return tuple(items)


@traces(spec="107", story="FR-001", reqs=["REQ-107-060"])
def _parse_phase_steps(phase_name: str, phase_body: str) -> list[PlanStep]:
    """Extract all steps from a single phase body."""
    steps: list[PlanStep] = []
    step_matches = list(_STEP_RE.finditer(phase_body))
    if not step_matches:
        return steps

    for i, match in enumerate(step_matches):
        step_num = int(match.group(1))
        title = match.group(2).strip()
        body_start = match.end()
        body_end = (
            step_matches[i + 1].start()
            if i + 1 < len(step_matches)
            else len(phase_body)
        )
        body = phase_body[body_start:body_end]
        fields = _extract_step_body_fields(body)

        actions_block = fields.get("actions", "")
        actions = _parse_actions(actions_block)

        steps.append(
            PlanStep(
                step_num=step_num,
                title=title,
                phase=phase_name,
                goal=fields.get("goal", ""),
                file=fields.get("file", ""),
                actions=actions,
                verify=fields.get("verify", ""),
                expected=fields.get("expected", ""),
                fallback=fields.get("fallback", ""),
            )
        )

    return steps


@traces(spec="107", reqs=["REQ-107-003", "REQ-107-060"])
def parse_plan_text(text: str) -> list[PlanStep]:
    """Parse execution-plan markdown text into ordered :class:`PlanStep` objects.

    Args:
        text: Full markdown content of a plan.

    Returns:
        Steps ordered by ``step_num``. Steps that cannot be parsed are skipped.
    """
    text = text.replace("\r\n", "\n")
    steps: list[PlanStep] = [
        step
        for phase_name, phase_body in _split_into_phases(text)
        for step in _parse_phase_steps(phase_name, phase_body)
    ]

    steps.sort(key=lambda s: s.step_num)
    return steps


def _resolve_plan_path(
    plan_path: str | Path | None,
    env: Mapping[str, str] | None,
) -> Path:
    """Resolve the plan path shared by ``parse_plan_file`` and
    ``load_plan_governance_facts``: an explicit path wins, otherwise
    ``ORCHESTRATA_PLAN_PATH``, otherwise ``plans/execution-plan.md``."""
    resolved_env = env if env is not None else os.environ
    if plan_path is None:
        plan_path = resolved_env.get("ORCHESTRATA_PLAN_PATH", "plans/execution-plan.md")
    return Path(plan_path)


@traces(spec="107", reqs=["REQ-107-003", "REQ-107-060"])
def parse_plan_file(
    plan_path: str | Path | None = None,
    *,
    env: Mapping[str, str] | None = None,
) -> list[PlanStep]:
    """Parse the plan file at ``plan_path``.

    Args:
        plan_path: Path to the plan markdown. If None, resolves via the
            ``ORCHESTRATA_PLAN_PATH`` environment variable, defaulting to
            ``plans/execution-plan.md``.
        env: Optional environment mapping for testing; falls back to
            ``os.environ``.

    Returns:
        Ordered list of parsed steps.

    Raises:
        FileNotFoundError: When the resolved plan file does not exist.
    """
    path = _resolve_plan_path(plan_path, env)
    text = path.read_text(encoding="utf-8")
    return parse_plan_text(text)


def find_step(steps: list[PlanStep], step_num: int) -> PlanStep | None:
    """Return the step with the given number, or None."""
    for step in steps:
        if step.step_num == step_num:
            return step
    return None


# Bold "status" markers used across this project's plans/*.md documents in
# two observed conventions: the value inside the bold span
# (``**STATUS: DONE**``, plans/execution-plan.md's own convention) or only
# the label bolded (``**Status:** Complete``). Neither is fabricated for
# this purpose -- both are the project's own pre-existing convention.
_STATUS_INLINE_RE = re.compile(r"\*\*\s*status\s*:\s*([^*\n]+?)\s*\*\*", re.IGNORECASE)
_STATUS_LABEL_RE = re.compile(
    r"\*\*\s*status\s*:?\s*\*\*\s*:?\s*([^\n]+)", re.IGNORECASE
)

# Status text (case-insensitive substring) that means a plan no longer
# governs -- it has been closed out, not merely undocumented. Used by
# constitutional Article XII (plan fidelity, RUL-015) to decide whether
# active_plan/1 should be asserted at all.
_TERMINAL_PLAN_STATUSES = (
    "DONE",
    "COMPLETE",
    "COMPLETED",
    "SUPERSEDED",
    "CLOSED",
    "ARCHIVED",
    "CANCELLED",
    "CANCELED",
)


@traces(spec="107", story="FR-001", reqs=["REQ-107-060"])
def parse_plan_status(text: str) -> str | None:
    """Extract a plan document's own ``**Status:** ...`` marker, if any.

    Returns None when neither convention is present -- most plan documents
    in this project never carry this marker at all, which is a fact about
    the document, not a parsing failure.
    """
    match = _STATUS_INLINE_RE.search(text) or _STATUS_LABEL_RE.search(text)
    if not match:
        return None
    return match.group(1).strip()


@traces(spec="107", story="FR-001", reqs=["REQ-107-060"])
def is_plan_active(status: str | None) -> bool:
    """Whether a plan carrying this parsed ``status`` currently governs.

    A plan is presumed ACTIVE (in force) unless it carries an explicit
    terminal marker (DONE/COMPLETE/SUPERSEDED/CLOSED/ARCHIVED/CANCELLED) --
    most plan documents never bother to write "STATUS: ACTIVE", they just
    ARE the current plan, so the absence of a marker must not read as
    inactive. ``plans/execution-plan.md`` itself is the concrete
    counter-example this exists for: it is marked ``STATUS: DONE``.
    """
    if status is None:
        return True
    upper = status.upper()
    return not any(terminal in upper for terminal in _TERMINAL_PLAN_STATUSES)


@traces(spec="007", story="US-002", reqs=["REQ-007-US-002-001"])
@dataclass(frozen=True)
class PlanGovernanceFacts:
    """Ground facts for constitutional Article XII (plan fidelity, RUL-015),
    derived from a REAL, parsed execution plan -- never fabricated. Python
    asserts these as ground facts; constitutional.pl's ``plan_fidelity/2``
    only reasons over them (ADR-0005; see prolog/kg_relevance.pl's header
    for the same doctrine stated for a different consumer).

    Attributes:
        plan_id: Stable identifier for this plan (the resolved plan path,
            as a string) -- the PlanId argument to active_plan/1 and
            plan_step/2.
        active: Whether the plan currently governs (see ``is_plan_active``).
            When False, no active_plan/1 fact should be asserted at all --
            Article XII's vacuous-truth clause then applies honestly.
        status: The raw parsed status text, if any (for logging).
        steps: The parsed steps, in the same structure SPEC-107 context
            injection (``plan_context.py``) already consumes.
    """

    plan_id: str
    active: bool
    status: str | None
    steps: tuple[PlanStep, ...]


@traces(spec="007", story="US-002", reqs=["REQ-007-US-002-001"])
def load_plan_governance_facts(
    plan_path: str | Path | None = None,
    *,
    env: Mapping[str, str] | None = None,
) -> PlanGovernanceFacts:
    """Resolve, parse, and evaluate the execution plan for constitutional
    Article XII (plan fidelity, RUL-015).

    Args:
        plan_path: Same resolution as ``parse_plan_file``.
        env: Optional environment mapping for testing; falls back to
            ``os.environ``.

    Returns:
        PlanGovernanceFacts for the resolved plan.

    Raises:
        FileNotFoundError: When the resolved plan file does not exist.
    """
    path = _resolve_plan_path(plan_path, env)
    text = path.read_text(encoding="utf-8")
    status = parse_plan_status(text)
    return PlanGovernanceFacts(
        plan_id=str(path),
        active=is_plan_active(status),
        status=status,
        steps=tuple(parse_plan_text(text)),
    )