architect_cli.py

"""CLI entry point for the Architect feature.

Runs the Architect agent against a requirements source (URL or file) and
generates skeleton Page Objects + a Pytest script under ``pages/<app>/``
and ``tests/<app>/``. Independent of pytest and the executor graph.

Example:
    python architect_cli.py --app hr_portal --source requirements.md
    python architect_cli.py --app hr_portal --source https://wiki/story/42 --dry-run
"""

import argparse
import ast
import asyncio
import re
import sys
from pathlib import Path

from langchain_core.messages import AIMessage, HumanMessage, ToolMessage

from agents.architect import (
    build_api_service_planner_graph,
    build_api_service_writer_graph,
    build_api_test_writer_graph,
    build_ui_pom_planner_graph,
    build_ui_pom_writer_graph,
    build_ui_test_writer_graph,
)
from agents.architect_briefing import build_existing_state_briefing
from config import ModelFactory, get_settings
from tools.architect_reviewer import (
    Origin,
    ReviewTarget,
    Severity,
    format_text,
    review,
)
from tools.architect_tools import (
    _read_source,
    _write_code_file_impl,
    create_architect_tools,
)
from utils.logger import get_logger, setup_logging

log = get_logger("architect_cli")

_APP_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9_]*$")

# Coverage-gap heuristics. These are best-effort and never block the run —
# their job is to surface "the source has more ACs than tests written" as a
# warning the operator sees alongside the run summary, not to enforce 1:1
# coverage. False positives and false negatives are acceptable; the goal is
# to make silently-dropped ACs impossible to miss during the review pass.
_AC_NUMBERED_PATTERN = re.compile(
    r"\b(\d+)\.\s+(.+?)(?=\s+\d+\.\s+|$)",
    re.DOTALL,
)
_HTML_TAG_PATTERN = re.compile(r"<[^>]+>")
_HTML_ENTITIES = {
    "&amp;": "&", "&quot;": '"', "&nbsp;": " ",
    "&apos;": "'", "&#39;": "'", "&lt;": "<", "&gt;": ">",
}
# Matches an "AcceptanceCriteria" JSON key with an optional dotted namespace
# prefix, so we capture the field value across ALM systems:
#   - bare:           "AcceptanceCriteria"
#   - Azure DevOps:   "Microsoft.VSTS.Common.AcceptanceCriteria"
#   - other prefixed: "Custom.Module.AcceptanceCriteria"
# Case-insensitive so camelCase variants (acceptanceCriteria) also match.
# When this pattern misses, _extract_acs falls back to scanning the whole
# source — works but loses precision because numbered items in the work-item
# Description (e.g. "User Story 1:") and the JSON tail can swallow the last
# AC's body, dropping it via the body-length filter.
_AC_FIELD_KEY_PATTERN = re.compile(
    r'"(?:\w+\.)*AcceptanceCriteria"\s*:\s*"((?:[^"\\]|\\.)*)"',
    re.IGNORECASE,
)
_TEST_FILE_PATH_PATTERN = re.compile(r"(?:^|/)tests/[^/]+/test_\w+\.py$")
_INITIAL_STATE_TASK_PATTERN = re.compile(
    r"_initial_state\(\s*[\"']((?:[^\"'\\]|\\.)+)[\"']"
)
# Captures a POM file write — ``pages/<app>/<module>.py`` — and exposes the
# ``<module>`` name (e.g. ``daily_log_page``) so we can match against POM
# imports in test files. Excludes ``__init__.py`` since those don't define
# Page Object classes.
_POM_FILE_PATH_PATTERN = re.compile(
    r"(?:^|/)pages/[^/]+/(?!__init__\.py$)(\w+)\.py$"
)
# Service-Object equivalent of _POM_FILE_PATH_PATTERN — used by API-mode
# coverage analysis.
_SERVICE_FILE_PATH_PATTERN = re.compile(
    r"(?:^|/)services/[^/]+/(?!__init__\.py$)(\w+)\.py$"
)
# Matches ``from pages.<app>.<module> import <ClassName>`` lines in test
# files. Used to spot tests that reference POMs the architect never wrote
# — those tests fail at import time, so surfacing the gap up front saves
# the operator a confusing ImportError on the first pytest run.
_POM_IMPORT_PATTERN = re.compile(
    r"^\s*from\s+pages\.(\w+)\.(\w+)\s+import\s+(\w+)",
    re.MULTILINE,
)
# Service-Object equivalent of _POM_IMPORT_PATTERN.
_SERVICE_IMPORT_PATTERN = re.compile(
    r"^\s*from\s+services\.(\w+)\.(\w+)\s+import\s+(\w+)",
    re.MULTILINE,
)

# Some lightweight chat models (gemini-2.5-flash-lite in particular) emit
# an empty terminating AIMessage after the seeded read_source result and
# never start writing code — the multi-step instructions buried at the top
# of the human turn aren't enough to push them past one round-trip.
# Re-issuing the write instructions as a fresh user turn nudges them past
# that, but it costs ~150 tokens of context, so we only do it for models
# whose id matches a known-weak hint. Stronger models (qwen2.5:32b,
# gemini-2.5-flash, claude, etc.) handle the original prompt on their own
# and don't pay the token tax. Add to the tuple if you find another model
# that needs the nudge.
_WEAK_AGENTIC_MODEL_HINTS = ("flash-lite",)


def _needs_followup_nudge(model_id: str) -> bool:
    return any(h in model_id.lower() for h in _WEAK_AGENTIC_MODEL_HINTS)


def _strip_html(text: str) -> str:
    """Strip HTML tags and decode common entities — good enough for ALM
    payloads where rich-text fields are HTML embedded in JSON strings."""
    no_tags = _HTML_TAG_PATTERN.sub(" ", text)
    for entity, replacement in _HTML_ENTITIES.items():
        no_tags = no_tags.replace(entity, replacement)
    return no_tags


def _extract_acs(source: str) -> list[str]:
    """Best-effort extraction of numbered acceptance criteria from a source.

    Focuses on a JSON ``AcceptanceCriteria`` field when one is present —
    matches the bare key, the Azure DevOps ``Microsoft.VSTS.Common.``
    prefix, and any other dotted namespace (case-insensitive) — so we
    avoid numbered items in the work-item description that aren't ACs.
    Falls back to scanning the whole source when no key matches. Returns
    formatted strings like ``"1. Multi-Input Support: ..."`` or an empty
    list when no AC pattern is detectable.
    """
    if not source or source.startswith("ERROR"):
        return []

    field_match = _AC_FIELD_KEY_PATTERN.search(source)
    target = (field_match.group(1).replace('\\"', '"').replace("\\\\", "\\")
              if field_match else source)
    target = _strip_html(target)
    target = re.sub(r"\s+", " ", target).strip()

    acs: list[str] = []
    for m in _AC_NUMBERED_PATTERN.finditer(target):
        num = m.group(1)
        body = m.group(2).strip().rstrip(".").strip()
        if 8 <= len(body) <= 300:
            acs.append(f"{num}. {body}")
    return acs


def _iter_write_calls(messages: list) -> list[tuple[str, str]]:
    """Collect ``(filepath, content)`` for every ``write_code_file`` tool
    call in the AIMessages. Filepaths are normalised to forward slashes
    so downstream callers don't have to repeat the conversion."""
    out: list[tuple[str, str]] = []
    for m in messages:
        if type(m).__name__ != "AIMessage":
            continue
        for tc in (getattr(m, "tool_calls", None) or []):
            if isinstance(tc, dict):
                name = tc.get("name", "")
                args = tc.get("args", {}) or {}
            else:
                name = getattr(tc, "name", "")
                args = getattr(tc, "args", {}) or {}
            if name != "write_code_file":
                continue
            filepath = args.get("filepath", "").replace("\\", "/")
            content = args.get("content", "")
            out.append((filepath, content))
    return out


def _extract_test_tasks(messages: list) -> list[str]:
    """Pull task strings from each generated test file.

    Keeps only ``write_code_file`` calls targeting ``tests/<app>/test_*.py``
    and extracts the string passed to ``<app>_initial_state(...)``. The
    pattern matches both UI (``<app>_initial_state``) and API
    (``<app>_api_initial_state``) factory names since both end with
    ``_initial_state(``. Returns one entry per test function.

    NOTE: only matches inline string literals (``_initial_state("text")``).
    When the model assigns the task to a variable first \
    (``task = "..."; ... _initial_state(task)``), the regex returns
    nothing for that file — which is why the coverage count uses
    ``_extract_test_files_written`` instead of ``len(_extract_test_tasks(...))``.
    """
    tasks: list[str] = []
    for filepath, content in _iter_write_calls(messages):
        if not _TEST_FILE_PATH_PATTERN.search(filepath):
            continue
        for match in _INITIAL_STATE_TASK_PATTERN.finditer(content):
            tasks.append(match.group(1))
    return tasks


def _extract_test_files_written(messages: list) -> set[str]:
    """Return the set of ``tests/<app>/test_*.py`` paths the architect wrote.

    Used as the authoritative ``tests_written`` count for the coverage
    summary — robust to whether the test embeds the task as an inline
    string literal or assigns it to a variable first.
    """
    files: set[str] = set()
    for filepath, _content in _iter_write_calls(messages):
        if _TEST_FILE_PATH_PATTERN.search(filepath):
            files.add(filepath)
    return files


def _extract_pom_imports(messages: list) -> list[tuple[str, str, str]]:
    """Walk test-file writes and pull every ``from pages.<app>.<module>
    import <ClassName>`` line. Returns one ``(app, module, class_name)``
    tuple per import — a test file that imports two POMs yields two entries.
    """
    imports: list[tuple[str, str, str]] = []
    for filepath, content in _iter_write_calls(messages):
        if not _TEST_FILE_PATH_PATTERN.search(filepath):
            continue
        for match in _POM_IMPORT_PATTERN.finditer(content):
            imports.append((match.group(1), match.group(2), match.group(3)))
    return imports


def _extract_service_imports(messages: list) -> list[tuple[str, str, str]]:
    """Service-Object equivalent of ``_extract_pom_imports``.

    Returns ``(app, module, class_name)`` tuples for every
    ``from services.<app>.<module> import <Class>`` line found in test files.
    """
    imports: list[tuple[str, str, str]] = []
    for filepath, content in _iter_write_calls(messages):
        if not _TEST_FILE_PATH_PATTERN.search(filepath):
            continue
        for match in _SERVICE_IMPORT_PATTERN.finditer(content):
            imports.append((match.group(1), match.group(2), match.group(3)))
    return imports


def _tools_by_name(all_tools: list, names: set[str]) -> list:
    """Filter a tool list by tool name.

    The architect graphs share a single ``create_architect_tools()`` list
    but each phase wants only a subset (planner: no ``write_code_file``;
    writer: no ``propose_module_list``). Filtering here keeps tool
    creation centralised while letting each phase enforce its own
    behavioural envelope at construction time.
    """
    return [t for t in all_tools if getattr(t, "name", None) in names]


def _extract_module_list_from_planner(messages: list) -> list[dict]:
    """Return the planner's proposed module list, as a list of dicts with
    keys ``module``, ``class_name``, and ``description``.

    Walks the planner's messages for an AIMessage that called
    ``propose_module_list`` and returns the ``modules`` argument verbatim
    (after light type-coercion to dict). Empty list when the planner
    never called the tool, called it with no modules, or the args were
    malformed — Phase 1B treats an empty list as "skip the writer loop"
    and the test phase falls back to whatever exists on disk.

    Last-call-wins if the planner calls the tool more than once (which
    the prompt forbids but defensive handling is cheap).
    """
    last: list[dict] = []
    for m in messages:
        if type(m).__name__ != "AIMessage":
            continue
        for tc in (getattr(m, "tool_calls", None) or []):
            if isinstance(tc, dict):
                name = tc.get("name", "")
                args = tc.get("args", {}) or {}
            else:
                name = getattr(tc, "name", "")
                args = getattr(tc, "args", {}) or {}
            if name != "propose_module_list":
                continue
            modules = args.get("modules") or []
            if not isinstance(modules, list):
                continue
            cleaned: list[dict] = []
            for entry in modules:
                if isinstance(entry, dict):
                    cleaned.append(entry)
            last = cleaned
    return last


def _extract_pom_modules_written(messages: list) -> set[str]:
    """Return the set of POM module names the architect actually wrote.

    A write to ``pages/fit_track_pro/daily_log_page.py`` adds
    ``"daily_log_page"`` to the result. Used to spot test imports that
    reference POMs the model silently dropped.
    """
    modules: set[str] = set()
    for filepath, _content in _iter_write_calls(messages):
        match = _POM_FILE_PATH_PATTERN.search(filepath)
        if match:
            modules.add(match.group(1))
    return modules


def _extract_service_modules_written(messages: list) -> set[str]:
    """Service-Object equivalent of ``_extract_pom_modules_written``."""
    modules: set[str] = set()
    for filepath, _content in _iter_write_calls(messages):
        match = _SERVICE_FILE_PATH_PATTERN.search(filepath)
        if match:
            modules.add(match.group(1))
    return modules


def _iter_write_call_results(messages: list) -> list[tuple[str, str, str]]:
    """Pair every ``write_code_file`` tool call with its result ToolMessage.

    Returns ``(filepath, content, result_str)`` per write attempt. The
    correlation key is ``tool_call_id`` — LangGraph attaches it on both the
    AIMessage tool_call and the corresponding ToolMessage, so a single
    walk of the message list pairs them deterministically.
    """
    pending: dict[str, tuple[str, str]] = {}
    paired: list[tuple[str, str, str]] = []
    for m in messages:
        cls = type(m).__name__
        if cls == "AIMessage":
            for tc in (getattr(m, "tool_calls", None) or []):
                if isinstance(tc, dict):
                    name = tc.get("name", "")
                    args = tc.get("args", {}) or {}
                    tc_id = tc.get("id", "")
                else:
                    name = getattr(tc, "name", "")
                    args = getattr(tc, "args", {}) or {}
                    tc_id = getattr(tc, "id", "")
                if name != "write_code_file" or not tc_id:
                    continue
                filepath = args.get("filepath", "").replace("\\", "/")
                content = args.get("content", "")
                pending[tc_id] = (filepath, content)
        elif cls == "ToolMessage":
            tc_id = getattr(m, "tool_call_id", "")
            result = str(getattr(m, "content", ""))
            if tc_id in pending:
                filepath, content = pending.pop(tc_id)
                paired.append((filepath, content, result))
    return paired


def _read_target_from_disk(filepath: str) -> str | None:
    """Read a file under cwd for the PRESERVED branch; None on failure."""
    try:
        return (Path.cwd() / filepath).read_text(encoding="utf-8")
    except OSError as exc:
        log.info("reviewer_skipped_unreadable", path=filepath, error=str(exc))
        return None


def _collect_review_targets(messages: list) -> dict[str, ReviewTarget]:
    """Map every ``write_code_file`` outcome to a ReviewTarget per the plan
    §7 table.

    ``SUCCESS`` / ``DRY-RUN``  → Origin.NEW with the LLM's content.
    ``NO-OP``                   → previous entry stays (idempotent re-emit).
    ``ERROR: ... already exists`` → Origin.PRESERVED with on-disk content
                                    (file kept; LLM's attempt was refused).
    Any other ``ERROR``         → skipped + ``reviewer_skipped_unwritable``.

    Last-write-wins per path: identical SUCCESS/DRY-RUN writes overwrite
    the prior content (model self-correction churn).
    """
    targets: dict[str, ReviewTarget] = {}
    for filepath, content, result in _iter_write_call_results(messages):
        if not filepath:
            continue
        if result.startswith("SUCCESS") or result.startswith("DRY-RUN"):
            targets[filepath] = ReviewTarget(content=content, origin=Origin.NEW)
        elif result.startswith("NO-OP"):
            continue
        elif result.startswith("ERROR") and "already exists" in result:
            disk = _read_target_from_disk(filepath)
            if disk is not None:
                targets[filepath] = ReviewTarget(content=disk, origin=Origin.PRESERVED)
        elif result.startswith("ERROR"):
            log.info("reviewer_skipped_unwritable",
                     path=filepath, reason=result[:200])
    return targets


def _add_conftest_to_targets(
    targets: dict[str, ReviewTarget], app: str, conftest_result: str,
) -> None:
    """Merge the orchestrator-written conftest into the targets dict.

    The conftest write is performed by ``_write_conftest_for_app``, not by
    an LLM tool call, so it never appears in ``messages`` and the standard
    collector misses it. Mirror the same outcome → origin mapping here.
    """
    conftest_path = f"tests/{app}/conftest.py"
    rendered = _CONFTEST_TEMPLATE.replace("__APP__", app)
    if conftest_result.startswith("SUCCESS") or conftest_result.startswith("DRY-RUN"):
        targets[conftest_path] = ReviewTarget(content=rendered, origin=Origin.NEW)
    elif conftest_result.startswith("PRESERVED"):
        disk = _read_target_from_disk(conftest_path)
        if disk is not None:
            targets[conftest_path] = ReviewTarget(content=disk, origin=Origin.PRESERVED)


def _read_source_content_from_messages(messages: list) -> str:
    """Return the result text of the read_source ToolMessage, or ``""``."""
    for m in messages:
        if (type(m).__name__ == "ToolMessage"
                and getattr(m, "name", "") == "read_source"):
            return str(getattr(m, "content", ""))
    return ""


# ── Module summary extraction (split-architecture infrastructure) ──
#
# When the split architect runs the POM/Service agent first and then a
# per-AC test-writer agent in a Python loop, the test writer needs to
# know which classes the previous pass produced and what methods they
# expose. AST-parsing the written files is the cleanest way to extract
# that — class names, ``@tool_method``-decorated method signatures, and
# docstrings — without re-running the LLM. The extracted summary is
# injected into the test-writer prompt so it knows which classes to
# import and which methods are available for assertions.
#
# The same logic handles UI POMs (under ``pages/``) and API Service
# Objects (under ``services/``) because both use the ``@tool_method``
# decorator pattern from ``BasePage`` / ``BaseService``.


def _has_tool_method_decorator(func_node: ast.AST) -> bool:
    """True if the function carries an ``@tool_method`` decorator.

    Handles both bare ``@tool_method`` and any future call form
    ``@tool_method(...)``. Other decorators (``@property``,
    ``@staticmethod``, etc.) return False so they're filtered from the
    test-writer briefing — only methods the executor's agent can call
    as tools belong in the summary.
    """
    decorators = getattr(func_node, "decorator_list", [])
    for dec in decorators:
        if isinstance(dec, ast.Name) and dec.id == "tool_method":
            return True
        if (isinstance(dec, ast.Call)
                and isinstance(dec.func, ast.Name)
                and dec.func.id == "tool_method"):
            return True
    return False


def _format_method_params(args_node: ast.arguments) -> list[dict]:
    """Return ``[{name, type}]`` for non-self positional/keyword args.

    Defaults and kwonly args are out of scope for v1 — POM/Service tool
    methods rarely use them, and adding them would make the prompt
    briefing noisier for marginal benefit. Add support if a real-world
    signature ever needs it.
    """
    params: list[dict] = []
    for arg in args_node.args:
        if arg.arg == "self":
            continue
        params.append({
            "name": arg.arg,
            "type": ast.unparse(arg.annotation) if arg.annotation else None,
        })
    return params


def _extract_module_summary(messages: list, kind: str = "pom") -> list[dict]:
    """Return structured summaries of every POM/Service class written.

    Walks ``write_code_file`` tool calls targeting ``pages/<app>/<module>.py``
    (when ``kind="pom"``) or ``services/<app>/<module>.py`` (when
    ``kind="service"``), AST-parses each one, and extracts every class
    with its ``@tool_method``-decorated methods. The returned summaries
    feed the per-AC test-writer agent's prompt so it knows which classes
    exist, where to import them from, and what verifiable methods they
    expose.

    Args:
        messages: LangGraph message list from the POM/Service agent run.
        kind: ``"pom"`` for UI Page Objects, ``"service"`` for API
            Service Objects. Determines which file paths are scanned.

    Returns:
        List of dicts shaped::

            {
                "module": "login_page",
                "module_path": "pages.adams_golf_club.login_page",
                "class_name": "LoginPage",
                "docstring": "Page Object for member login.",
                "methods": [
                    {
                        "name": "navigate",
                        "params": [],
                        "returns": "str",
                        "docstring": "Navigate to the login page.",
                    },
                    ...
                ],
            }

        Multiple classes in the same file produce multiple entries.
        Files that fail to AST-parse (shouldn't happen — the architect's
        write tool already validates syntax) are silently skipped to
        avoid breaking the briefing for the rest.
    """
    if kind == "pom":
        path_pattern = _POM_FILE_PATH_PATTERN
        path_prefix = "pages"
    elif kind == "service":
        path_pattern = _SERVICE_FILE_PATH_PATTERN
        path_prefix = "services"
    else:
        raise ValueError(f"Unknown kind: {kind!r} (expected 'pom' or 'service')")

    # ``seen`` dedups by ``(module_path, class_name)``. When the LLM
    # re-emits the same write multiple times (the retry-on-timeout failure
    # mode we observed in practice), the same class ends up parsed
    # multiple times — once per duplicate write. The dedup is keyed on
    # the path so two distinct classes with the same name in different
    # modules still produce separate entries, and the LAST write wins
    # if the model emitted differing variants of the same class (later
    # versions are presumed more correct than earlier ones).
    seen: dict[tuple[str, str], dict] = {}
    for filepath, content in _iter_write_calls(messages):
        match = path_pattern.search(filepath)
        if not match:
            continue
        module_name = match.group(1)

        # Derive ``app`` from the path: pages/adams_golf_club/login_page.py
        # → adams_golf_club. ``filepath`` is already normalised to forward
        # slashes by ``_iter_write_calls``.
        path_parts = filepath.split("/")
        try:
            root_idx = path_parts.index(path_prefix)
        except ValueError:
            continue
        if root_idx + 2 >= len(path_parts):
            continue
        app = path_parts[root_idx + 1]
        module_path = f"{path_prefix}.{app}.{module_name}"

        try:
            tree = ast.parse(content)
        except SyntaxError:
            continue

        for node in ast.walk(tree):
            if not isinstance(node, ast.ClassDef):
                continue
            class_summary = {
                "module": module_name,
                "module_path": module_path,
                "class_name": node.name,
                "docstring": ast.get_docstring(node) or "",
                "methods": [],
            }
            for item in node.body:
                if not isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    continue
                if not _has_tool_method_decorator(item):
                    continue
                returns = (
                    ast.unparse(item.returns) if item.returns else None
                )
                class_summary["methods"].append({
                    "name": item.name,
                    "params": _format_method_params(item.args),
                    "returns": returns,
                    "docstring": ast.get_docstring(item) or "",
                })
            seen[(module_path, node.name)] = class_summary
    return list(seen.values())


# ── Programmatic conftest write (split-architecture infrastructure) ──
#
# The conftest factory is fully templated — only the function name varies
# by app. Asking the LLM to produce it wastes a tool call and risks the
# whole "POM phase ends without writing conftest" failure mode we hit in
# practice. We render the file from a Python template here and write it
# from the orchestrator, removing one step from the POM agent's job and
# closing that failure class entirely.
#
# Variability handling: if a user has a custom conftest from a prior run
# or hand-written extras, the existing-file check in ``_write_code_file_impl``
# preserves it (the auto-write returns ERROR, which the helper translates
# to PRESERVED for the orchestrator's purposes). ``--force`` overrides
# and overwrites — same semantics as every other architect-generated file.

_CONFTEST_TEMPLATE = '''from agents.state import AgentState


def __APP___initial_state(task: str) -> AgentState:
    """Return the initial AgentState for __APP__ tests."""
    return {
        "task": task,
        "messages": [],
        "steps_taken": [],
        "dom_snapshots": [],
        "error_history": [],
        "screenshots": [],
        "current_phase": "init",
        "final_result": None,
        "iteration_count": 0,
        "routing_decision": "pom_tools",
        "healing_occurred": False,
        "test_context": {},
    }
'''


def _write_conftest_for_app(app: str, force: bool, dry_run: bool) -> str:
    """Render and write the standard conftest factory file for ``app``.

    Returns a status string mirroring ``_write_code_file_impl``'s convention:
      - ``SUCCESS``: wrote a new file (or overwrote with ``force``).
      - ``DRY-RUN``: would have written; content was echoed to stdout.
      - ``PRESERVED``: file already exists and ``--force`` was not set —
        translates the underlying ``ERROR: ... already exists`` into a
        friendlier signal because preserving a custom conftest is the
        intended behaviour, not an error.
      - ``ERROR``: something genuinely went wrong (path traversal,
        write permission, etc.). The orchestrator aborts in this case.
    """
    target_path = f"tests/{app}/conftest.py"
    content = _CONFTEST_TEMPLATE.replace("__APP__", app)

    # ``written_pairs=None`` — the conftest helper is called once per
    # architect run from the orchestrator, not from the agent's tool
    # loop, so the per-run dedup set isn't useful here.
    result = _write_code_file_impl(
        target_path, content,
        force=force, dry_run=dry_run,
        written_pairs=None,
    )

    # Existing-file-without-force is the documented "preserve customisation"
    # path, not a failure. Re-frame it for the orchestrator + operator.
    if result.startswith("ERROR") and "already exists" in result:
        return (f"PRESERVED: {target_path} already exists; keeping the "
                f"existing content (re-run with --force to overwrite "
                f"with the standard template).")
    return result


def _format_module_summary_for_prompt(summaries: list[dict]) -> str:
    """Render extracted summaries as a prompt-ready briefing block.

    Format optimised for LLM consumption: one block per class with the
    import path, purpose docstring, and method signatures. The test-writer
    agent reads this to compose imports and pick verification methods.

    An empty input returns ``"(no classes available)"`` so the test-writer
    prompt always contains *something* under the briefing heading — useful
    for debugging if the POM agent silently produced nothing.
    """
    if not summaries:
        return "(no classes available)"

    blocks: list[str] = []
    for summary in summaries:
        lines = [
            f"### {summary['class_name']}",
            f"Import: from {summary['module_path']} import {summary['class_name']}",
        ]
        if summary["docstring"]:
            lines.append(f"Purpose: {summary['docstring']}")
        if not summary["methods"]:
            lines.append("(no @tool_method methods)")
        else:
            lines.append("Methods:")
            for method in summary["methods"]:
                params_str = ", ".join(
                    (f"{p['name']}: {p['type']}" if p["type"] else p["name"])
                    for p in method["params"]
                )
                returns_str = f" -> {method['returns']}" if method["returns"] else ""
                lines.append(f"  - {method['name']}({params_str}){returns_str}")
                if method["docstring"]:
                    lines.append(f"    {method['docstring']}")
        blocks.append("\n".join(lines))
    return "\n\n".join(blocks)


async def _run(source: str, app: str, force: bool, dry_run: bool,
               recursion_limit: int, mode: str = "ui") -> None:
    """Top-level entry — wires the shared setup (settings, LLM, tools)
    and invokes the split multi-agent architect.

    The single-pass architecture was removed once the split topology
    (POM/Service-Object agent + per-AC test-writer loop) reached parity
    with it in both modes and surpassed it on test-iteration reliability.
    """
    settings = get_settings()
    llm = ModelFactory(settings).get_architect_chat_model()
    tools = create_architect_tools(force=force, dry_run=dry_run)

    log.info("architect_run_start", source=source, app=app, mode=mode,
             force=force, dry_run=dry_run)

    await _run_split_architecture(
        source=source, app=app, mode=mode, llm=llm, tools=tools,
        recursion_limit=recursion_limit,
        force=force, dry_run=dry_run,
    )


async def _run_split_architecture(
    source: str, app: str, mode: str, llm, tools,
    recursion_limit: int,
    force: bool = False, dry_run: bool = False,
) -> None:
    """Multi-agent architect — POM agent writes Page Objects, the
    orchestrator writes the templated conftest, then a per-AC test-writer
    agent is invoked once per acceptance criterion in a Python loop. Each
    LLM session has a smaller, focused prompt and only sees the work
    directly relevant to its job.

    Pass 1 (POM agent):
        - Reads the source via the seeded ``read_source`` ToolMessage.
        - Writes Page Objects only — the conftest is no longer the
          agent's responsibility; the orchestrator handles it
          deterministically below.
        - Does NOT write tests.

    Conftest auto-write (between pass 1 and pass 2):
        - The conftest factory is fully templated (only the function
          name varies), so we render it from a Python constant rather
          than burn a tool call on it. ``--force`` governs overwrite of
          a user-customised file, same as every other generated file.
          ``force`` and ``dry_run`` are threaded here from ``_run`` so
          the helper can match the rest of the architect's flag
          semantics.

    Pass 2 (per-AC test writer, looped programmatically):
        - The POM/Service agent's written output is AST-parsed into a
          class + method briefing.
        - The source's acceptance criteria are extracted.
        - For each AC, a fresh test-writer graph is invoked with a
          focused user message containing the AC text and the briefing.
        - Each invocation writes ONE test file.

    The Python ``for`` loop replaces the LLM-side iteration that was
    collapsing on local models. The test-writer agent never has to
    remember "I'm on AC 3 of 5" because each call is independent.

    Mode dispatch: ``mode="ui"`` uses the Page-Object pair of graph
    builders; ``mode="api"`` uses the Service-Object pair. Everything
    else (conftest auto-write, source seeding, AC extraction,
    aggregated diagnostics) is mode-agnostic — the auto-written
    conftest's factory name pattern (``<app>_initial_state``) is the
    same for both modes, matching the convention in
    ``tests/sample_api/conftest.py``.
    """
    if mode == "ui":
        build_phase1_planner_graph = build_ui_pom_planner_graph
        build_phase1_writer_graph = build_ui_pom_writer_graph
        build_phase2_graph = build_ui_test_writer_graph
        phase1_kind = "pom"
        phase1_root = "pages"
        phase1_artifact_label = "Page Object"
        phase1_briefing_label = "PAGE OBJECT"
    elif mode == "api":
        build_phase1_planner_graph = build_api_service_planner_graph
        build_phase1_writer_graph = build_api_service_writer_graph
        build_phase2_graph = build_api_test_writer_graph
        phase1_kind = "service"
        phase1_root = "services"
        phase1_artifact_label = "Service Object"
        phase1_briefing_label = "SERVICE OBJECT"
    else:
        raise ValueError(f"Unknown architect mode: {mode!r}")

    # Per-phase tool scoping. The planner can read source and propose a
    # list, but MUST NOT have write_code_file (otherwise it can smuggle
    # writes into the planning step and re-collapse the loop we just
    # split apart). The writer can read and write but MUST NOT have
    # propose_module_list (its job is to write one specific module the
    # planner already chose).
    planner_tools = _tools_by_name(tools, {"read_source", "propose_module_list"})
    writer_tools = _tools_by_name(tools, {"read_source", "write_code_file"})

    # ── Source pre-fetch + state-aware briefing ───────────────────
    # Read the source first so (a) the briefing can score exemplars
    # against the work item's DB/auth signals, and (b) an unreadable
    # source aborts the run before any other setup work.
    log.info("architect_seed_read_source", source=source)
    source_content = await _read_source(source)
    if source_content.startswith("ERROR"):
        log.error("architect_run_aborted_source_unreadable",
                  source=source, error=source_content)
        return

    # Two briefings: Phase 1 (POM/Service writer) doesn't need to see a
    # test exemplar — its job is the page/service surface, not test shape.
    # Dropping the exemplar trims ~2.5KB of constant context and removes
    # a stylistic distraction. Phase 2 keeps the exemplar because copying
    # the canonical test shape (try/finally, dual-close, assertion-place-
    # ment) is exactly what it's there for.
    phase1_state_briefing = build_existing_state_briefing(
        app=app, mode=mode, source_content=source_content,
        include_exemplar=False,
    )
    phase2_state_briefing = build_existing_state_briefing(
        app=app, mode=mode, source_content=source_content,
        include_exemplar=True,
    )

    # ── Phase 1A: Planner ─────────────────────────────────────────
    # The planner reads the source and proposes the list of POMs / Service
    # Objects the work item needs via the ``propose_module_list`` tool.
    # One small call, bounded output (a structured list, not file
    # content) — runs comfortably within the per-call wall-clock budget
    # even on local 27–32B models.
    log.info("architect_split_pom_phase_start", app=app, mode=mode)
    planner_graph = build_phase1_planner_graph(llm, planner_tools, app=app)

    planner_initial_message = HumanMessage(
        content=(
            (f"{phase1_state_briefing}\n\n" if phase1_state_briefing else "")
            + f"Source: {source}\n"
            f"App namespace: {app}\n\n"
            f"Do the following:\n"
            f"1. Call ``read_source`` with the source above (its content "
            f"is already attached below).\n"
            f"2. Identify every {phase1_artifact_label} the work item "
            f"needs (see LOGIN PRECONDITION DETECTION / GROUPING RULE in "
            f"your system prompt).\n"
            f"3. Call ``propose_module_list`` ONCE with the full list. "
            f"Do NOT write any code — a separate WRITER agent scaffolds "
            f"each module from your list.\n"
            f"4. Reply with a one-paragraph summary."
        )
    )

    seed_tool_call_id = "seed_read_source_1"
    planner_seeded_messages = [
        planner_initial_message,
        AIMessage(
            content="",
            tool_calls=[{
                "name": "read_source",
                "args": {"path_or_url": source},
                "id": seed_tool_call_id,
            }],
        ),
        ToolMessage(
            content=source_content,
            name="read_source",
            tool_call_id=seed_tool_call_id,
        ),
    ]

    try:
        planner_state = await planner_graph.ainvoke(
            {"messages": planner_seeded_messages},
            config={"recursion_limit": recursion_limit},
        )
        planner_messages = planner_state.get("messages", [])
    except Exception as exc:
        log.warning("architect_split_planner_phase_failed", error=str(exc))
        planner_messages = list(planner_seeded_messages)

    module_specs = _extract_module_list_from_planner(planner_messages)
    log.info("architect_split_planner_phase_done",
             modules_proposed=len(module_specs),
             modules=[s.get("module", "") for s in module_specs])

    if not module_specs:
        log.warning(
            "architect_split_planner_empty",
            hint=("Planner did not call ``propose_module_list`` or "
                  "called it with an empty list — the per-module writer "
                  "loop will be skipped. The test-writer phase will run "
                  "with an empty class catalog; tests may end up wiring "
                  "imports that don't resolve."),
        )

    # ── Phase 1B: Per-module writer loop ──────────────────────────
    # Mirrors Phase 2's structure: Python loop, fresh graph per
    # invocation, bounded per-call input (briefing + one module spec)
    # and output (one ``write_code_file`` call). A slow generation now
    # fails in isolation; the loop carries on with the remaining
    # modules instead of losing the whole phase.
    pom_messages: list = list(planner_messages)
    for idx, spec in enumerate(module_specs, start=1):
        module = (spec.get("module") or "").strip()
        class_name = (spec.get("class_name") or "").strip()
        description = (spec.get("description") or "").strip()
        if not (module and class_name):
            log.warning("architect_split_writer_spec_skipped",
                        spec_index=idx, reason="missing module or class_name",
                        spec=spec)
            continue
        log.info("architect_split_writer_phase_start",
                 spec_index=idx, module=module, class_name=class_name)
        writer_graph = build_phase1_writer_graph(llm, writer_tools, app=app)
        writer_user_message = HumanMessage(
            content=(
                (f"{phase1_state_briefing}\n\n" if phase1_state_briefing else "")
                + f"TARGET MODULE ({idx} of {len(module_specs)} — write "
                f"THIS ONE ONLY):\n"
                f"- module: {module}\n"
                f"- class_name: {class_name}\n"
                f"- description: {description}\n\n"
                f"Now call ``write_code_file`` ONCE with the complete "
                f"{phase1_artifact_label} at "
                f"``{phase1_root}/{app}/{module}.py`` defining class "
                f"``{class_name}``. The class MUST cover every method "
                f"the description names — the downstream test writer "
                f"cannot invent methods that don't exist."
            )
        )
        try:
            writer_state = await writer_graph.ainvoke(
                {"messages": [writer_user_message]},
                config={"recursion_limit": recursion_limit},
            )
        except Exception as exc:
            log.warning("architect_split_writer_phase_failed",
                        spec_index=idx, module=module, error=str(exc))
            continue
        writer_messages = writer_state.get("messages", [])
        writer_writes = [
            getattr(m, "content", "")
            for m in writer_messages
            if (type(m).__name__ == "ToolMessage"
                and getattr(m, "name", "") == "write_code_file")
        ]
        log.info("architect_split_writer_phase_done",
                 spec_index=idx, module=module,
                 write_results=writer_writes)
        pom_messages.extend(writer_messages)

    pom_writes = [
        getattr(m, "content", "")
        for m in pom_messages
        if (type(m).__name__ == "ToolMessage"
            and getattr(m, "name", "") == "write_code_file")
    ]
    log.info("architect_split_pom_phase_done",
             tool_messages=len([
                 m for m in pom_messages
                 if type(m).__name__ == "ToolMessage"
             ]),
             write_results=pom_writes)

    # ── Programmatic conftest write ─────────────────────────────
    # The conftest factory is fully templated — only the function name
    # varies by app — so writing it from the orchestrator is strictly
    # more reliable than asking the LLM to do it (which historically
    # got skipped when the POM agent collapsed early or exhausted its
    # iteration budget on duplicate-write loops). ``--force`` governs
    # whether a user-customised conftest is preserved or overwritten,
    # matching the semantics of every other architect-generated file.
    conftest_result = _write_conftest_for_app(
        app=app, force=force, dry_run=dry_run,
    )
    log.info("architect_split_conftest_auto_written",
             app=app, result=conftest_result)
    if conftest_result.startswith("ERROR"):
        log.error(
            "architect_split_conftest_write_failed",
            app=app, result=conftest_result,
            hint=("The deterministic conftest write failed. The per-AC "
                  "tests import from this file, so the test-writer "
                  "phase is aborted to avoid generating tests that "
                  "would fail at collection time. Investigate the "
                  "underlying error (path traversal, write permission, "
                  "or a SyntaxError in the templated content) and "
                  "rerun."),
        )
        return

    # ── Briefing extraction (AST-based) ───────────────────────────
    # ``phase1_kind`` selects ``pages/`` (UI POMs) vs ``services/`` (API
    # Service Objects); the extractor handles both via the same AST logic.
    summaries = _extract_module_summary(pom_messages, kind=phase1_kind)
    briefing = _format_module_summary_for_prompt(summaries)
    log.info("architect_split_briefing_built",
             kind=phase1_kind,
             classes=[s["class_name"] for s in summaries],
             total_methods=sum(len(s["methods"]) for s in summaries))

    # ── AC extraction ─────────────────────────────────────────────
    acs = _extract_acs(source_content)
    if not acs:
        log.warning(
            "architect_split_no_acs_extracted",
            hint=("Source did not yield any extractable acceptance "
                  "criteria — the per-AC test-writer loop will not run. "
                  "Check the source format: the AC extractor expects "
                  "an AcceptanceCriteria field with a numbered list."),
        )

    # ── Pass 2: per-AC test writer loop ───────────────────────────
    # Each iteration builds a fresh graph and invokes it with a focused
    # user message. The agent's view is bounded to a single AC — no
    # iteration logic lives in the LLM. We accumulate every invocation's
    # messages so the post-run diagnostics see the complete picture.
    all_messages: list = list(pom_messages)
    for idx, ac in enumerate(acs, start=1):
        log.info("architect_split_test_phase_start", ac_index=idx, ac=ac[:200])
        test_graph = build_phase2_graph(llm, tools, app=app)
        per_ac_user_message = HumanMessage(
            content=(
                (f"{phase2_state_briefing}\n\n" if phase2_state_briefing else "")
                + f"Acceptance Criterion {idx} of {len(acs)} (write the test "
                f"for THIS ONE ONLY):\n\n"
                f"{ac}\n\n"
                f"AVAILABLE {phase1_briefing_label} CLASSES (use ONLY "
                f"these — do not invent or import others):\n\n"
                f"{briefing}\n\n"
                f"A conftest factory ``{app}_initial_state(task)`` is "
                f"already available at ``tests/{app}/conftest.py``.\n\n"
                f"Now call ``write_code_file`` ONCE with the complete "
                f"test file at ``tests/{app}/test_<your_chosen_name>.py``. "
                f"Pick a file name that reflects this specific AC — "
                f"do NOT reuse generic names that might collide with "
                f"other tests."
            )
        )
        try:
            test_state = await test_graph.ainvoke(
                {"messages": [per_ac_user_message]},
                config={"recursion_limit": recursion_limit},
            )
        except Exception as exc:
            log.warning("architect_split_test_phase_failed",
                        ac_index=idx, error=str(exc))
            continue
        per_ac_messages = test_state.get("messages", [])
        per_ac_writes = [
            getattr(m, "content", "")
            for m in per_ac_messages
            if (type(m).__name__ == "ToolMessage"
                and getattr(m, "name", "") == "write_code_file")
        ]
        log.info("architect_split_test_phase_done",
                 ac_index=idx,
                 write_results=per_ac_writes)
        all_messages.extend(per_ac_messages)

    # ── Aggregated diagnostics ────────────────────────────────────
    # Same checks as the single-pass run, but driven by the union of
    # all messages across the POM phase and every test-writer invocation.
    tool_messages = [
        m for m in all_messages if type(m).__name__ == "ToolMessage"
    ]
    writes = [
        getattr(m, "content", "") for m in tool_messages
        if getattr(m, "name", "") == "write_code_file"
    ]
    log.info("architect_run_done",
             tool_messages=len(tool_messages),
             write_results=writes)

    test_files = _extract_test_files_written(all_messages)
    if acs:
        log.info("architect_coverage_summary",
                 acs_in_source=len(acs),
                 tests_written=len(test_files),
                 acs=acs)
        if len(acs) > len(test_files):
            log.warning(
                "architect_coverage_gap",
                acs_in_source=len(acs),
                tests_written=len(test_files),
                hint=("source contains more ACs than tests written — "
                      "check the per-AC test-writer logs for the "
                      "specific AC indices that did not produce a file."),
            )

    if mode == "ui":
        imports_in_tests = _extract_pom_imports(all_messages)
        modules_written_set = _extract_pom_modules_written(all_messages)
    else:
        imports_in_tests = _extract_service_imports(all_messages)
        modules_written_set = _extract_service_modules_written(all_messages)
    if imports_in_tests:
        imported_modules = sorted({mod for _ns, mod, _cls in imports_in_tests})
        # A module is "missing" only if it was neither written in this run
        # NOR present on disk under the phase-1 root. Pre-existing scaffolding
        # (e.g. login_page.py that the new feature traverses but doesn't
        # change) is correctly *not* re-emitted by the architect, so warning
        # on it would be a false alarm — the test imports resolve fine at
        # collection time.
        phase1_dir = Path.cwd() / phase1_root / app
        missing_modules = [
            m for m in imported_modules
            if m not in modules_written_set
            and not (phase1_dir / f"{m}.py").exists()
        ]
        log.info("architect_module_summary",
                 kind=phase1_kind,
                 modules_imported=imported_modules,
                 modules_written=sorted(modules_written_set))
        if missing_modules:
            example_imports = sorted({
                f"{phase1_root}.{ns}.{mod} import {cls}"
                for ns, mod, cls in imports_in_tests
                if mod in missing_modules
            })
            log.warning(
                "architect_module_gap",
                kind=phase1_kind,
                missing_modules=missing_modules,
                example_imports=example_imports,
                hint=(f"test files import {phase1_artifact_label} modules "
                      f"that were never written by the phase-1 agent and "
                      f"are not present on disk — these tests will fail at "
                      f"import time. Either rerun the architect or "
                      f"manually scaffold the missing modules under "
                      f"{phase1_root}/<app>/."),
            )

    # ── Reviewer pass ─────────────────────────────────────────────
    # Advisory only — findings print to stdout and log to architect.log,
    # but never fail the run. The operator decides what to act on.
    targets = _collect_review_targets(all_messages)
    _add_conftest_to_targets(targets, app, conftest_result)
    findings = review(targets, source_content=source_content, app=app)
    error_count = sum(1 for f in findings if f.severity == Severity.ERROR)
    warning_count = sum(1 for f in findings if f.severity == Severity.WARNING)
    log.info("architect_review_done",
             total=len(findings),
             errors=error_count,
             warnings=warning_count,
             targets_reviewed=len(targets))
    print(format_text(findings, targets), file=sys.stdout)


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Generate POM + test skeletons from a requirements source",
    )
    parser.add_argument("--source", required=True,
                        help="URL or local file path containing the requirements")
    parser.add_argument("--app", required=True,
                        help="App namespace (snake_case). Files land under "
                             "pages/<app>/ and tests/<app>/.")
    parser.add_argument("--force", action="store_true",
                        help="Overwrite existing files instead of refusing")
    parser.add_argument("--dry-run", action="store_true",
                        help="Print intended writes without touching disk")
    parser.add_argument("--recursion-limit", type=int, default=12,
                        help="Max tool-call loop iterations (default: 12)")
    parser.add_argument("--mode", choices=("ui", "api"), default="ui",
                        help="Scaffold UI Page Objects ('ui', default) or "
                             "API Service Objects ('api')")
    args = parser.parse_args()

    if not _APP_NAME_PATTERN.fullmatch(args.app):
        parser.error("--app must be a lowercase snake_case identifier "
                     "(e.g. 'hr_portal')")

    # Persist this run's structured events to a dedicated audit log. The
    # default ``mode="w"`` truncates at session start, so each file holds
    # exactly one run — same lifecycle as the executor's execution.log,
    # but lives outside results/logs/ so pytest never wipes it.
    settings = get_settings()
    setup_logging(log_file=settings.architect_log_file)

    asyncio.run(_run(
        source=args.source,
        app=args.app,
        force=args.force,
        dry_run=args.dry_run,
        recursion_limit=args.recursion_limit,
        mode=args.mode,
    ))


if __name__ == "__main__":
    main()