a0_software_orchestrator/helpers/handover_protocol.py

"""Handover protocol helpers for specialist subagents (Phase 1 - read-only spike).

Implements the 3-stage context transfer pattern documented in
/usr/workdir/handover-spike.md (sections 4 and 5).

This module is a HELPER (not a Tool). It exposes pure functions that the
HandoverToSpecialist tool and future code can call. It is intentionally
LLM-free in Phase 1: the orchestrator is expected to pre-fill the LLM
summary before invoking the tool. The actual LLM summarisation belongs in
a later phase.

Protection rules (enforced here):
- never compact mid-subagent
- never skip snapshot (callers MUST call snapshot_state first)
- never overwrite resume/conversation_summary with empty content
- never use chat history as project memory (briefing files are
  curated LLM summaries, not raw chat dumps)
"""
from __future__ import annotations

import os
import re
from datetime import datetime, timezone
from typing import Optional

from usr.plugins.a0_software_orchestrator.helpers.context_budget import (
    estimate_tokens,
    should_warn,
)


# Known specialist subagent profile directories. The orchestrator will only
# TODO: Centralize this allow-list so call_subordinate enforces it too.
# Currently call_subordinate can invoke ANY profile, bypassing this check.
# accept handovers to profiles in this allow-list. Update this list when new
# profiles are added to /a0/usr/plugins/a0_software_orchestrator/agents/.
KNOWN_SPECIALISTS = {
    "a0_software_orchestrator",
    "codebase_explorer",
    "deploy_agent",
    "implementation_engineer",
    "quality_reviewer",
    "release_auditor",
    "requirements_analyst",
    "runtime_devops_engineer",
    "security_data_engineer",
    "solution_architect",
    "test_debug_engineer",
    "hacker",
    "ui_design_specialist",
}

# Hard caps (token budget). All numbers are conservative defaults that match
# section 4 of /usr/workdir/handover-spike.md.
DEFAULT_MAX_CONTEXT_TOKENS = 3000
DEFAULT_RECENT_TURNS = 5
HARD_CAP_HISTORY_TOKENS = 50_000
MAX_HANDOVER_DEPTH = 3


# Filename safety: keep this strict so we never escape the project root via a
# malicious topic name. We allow letters, digits, dashes, underscores and dots.
_SAFE_SLUG = re.compile(r"[^a-zA-Z0-9._-]+")


def _safe_slug(value: str) -> str:
    """Sanitise a free-form string into a filename-safe slug."""
    cleaned = _SAFE_SLUG.sub("-", value.strip())
    cleaned = cleaned.strip("-")
    return cleaned or "handover"


def _today_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%d")


def _timestamp_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def list_known_specialists() -> list[str]:
    """Return the sorted list of accepted specialist profile names."""
    return sorted(KNOWN_SPECIALISTS)


def is_known_specialist(profile: str) -> bool:
    """True if `profile` is in the handover allow-list."""
    return profile in KNOWN_SPECIALISTS


def build_briefing_path(project_root: str, specialist: str, topic: str) -> str:
    """Return the absolute path where a briefing file should be written.

    Layout: <project_root>/.a0proj/handover/<specialist>-<topic>-<date>.md
    The directory is created on demand by append_briefing().
    """
    if not is_known_specialist(specialist):
        raise ValueError(f"Unknown specialist profile: {specialist!r}")
    slug = _safe_slug(topic)
    filename = f"{_safe_slug(specialist)}-{slug}-{_today_iso()}.md"
    return os.path.join(project_root, ".a0proj", "handover", filename)


def render_briefing_section(
    *,
    specialist: str,
    topic: str,
    summary: str,
    decisions: list[str],
    artifacts: list[str],
    recent_turns: list[str],
    return_condition: str,
    handover_reason: str,
) -> str:
    """Render the markdown content for a new briefing-file section.

    This is the canonical layout. The orchestrator (or the tool) appends the
    result to a per-specialist briefing file. Sections are append-only and
    timestamped, in line with the 'never overwrite' protection rule.
    """
    lines = [
        f"## {specialist} – {topic} ({_timestamp_iso()})",
        "",
        f"**Handover reason:** {handover_reason.strip() or 'n/a'}",
        "",
        "**Summary (LLM-curated, max 300 tokens):**",
        "",
        summary.strip() or "_(no summary provided)_",
        "",
        "**Recent turns (verbatim, last N):**",
        "",
    ]
    if recent_turns:
        for turn in recent_turns:
            lines.append(f"> {turn.strip()}")
            lines.append("")
    else:
        lines.append("_(no recent turns provided)_")
        lines.append("")
    lines.append("**Decisions so far:**")
    lines.append("")
    if decisions:
        for d in decisions:
            lines.append(f"- {d}")
    else:
        lines.append("- _(none yet)_")
    lines.append("")
    lines.append("**Artifacts (paths/references):**")
    lines.append("")
    if artifacts:
        for a in artifacts:
            lines.append(f"- `{a}`")
    else:
        lines.append("- _(none yet)_")
    lines.append("")
    lines.append(f"**Return condition:** {return_condition.strip() or 'user_request'}")
    lines.append("")
    lines.append("---")
    lines.append("")
    return "\n".join(lines)


def append_briefing(path: str, section_markdown: str) -> int:
    """Append `section_markdown` to the briefing file at `path`.

    Creates the parent directory if it does not exist. Returns the new file
    size in bytes. Raises OSError on filesystem errors.
    """
    if not section_markdown or not section_markdown.strip():
        # Protection rule: never overwrite/append with empty content.
        raise ValueError("refusing to append empty briefing content")
    parent = os.path.dirname(path)
    if parent:
        os.makedirs(parent, exist_ok=True)
    # Append-only: open in append mode, never in write/truncate mode.
    with open(path, "a", encoding="utf-8") as fh:
        fh.write(section_markdown)
        if not section_markdown.endswith("\n"):
            fh.write("\n")
    return os.path.getsize(path)


def estimate_handover_tokens(
    *,
    summary: str,
    recent_turns: list[str],
    decisions: list[str],
    artifacts: list[str],
) -> int:
    """Rough token estimate of the stage-1 handover package (no system prompt)."""
    text_parts = [summary or ""]
    text_parts.extend(recent_turns or [])
    text_parts.extend(decisions or [])
    text_parts.extend(artifacts or [])
    blob = "\n".join(text_parts)
    return estimate_tokens(blob)


def validate_history_request(max_tokens: int) -> int:
    """Validate a stage-2 history request against the hard cap.

    Returns the effective token budget (clamped to HARD_CAP_HISTORY_TOKENS).
    Raises ValueError if the caller asks for zero or negative tokens.
    """
    if max_tokens <= 0:
        raise ValueError("max_tokens must be positive")
    return min(max_tokens, HARD_CAP_HISTORY_TOKENS)


def build_handover_package(
    *,
    specialist: str,
    topic: str,
    summary: str,
    recent_turns: list[str],
    decisions: list[str],
    artifacts: list[str],
    return_condition: str,
    handover_reason: str,
    project_root: str,
    max_context_tokens: int = DEFAULT_MAX_CONTEXT_TOKENS,
) -> dict:
    """Build a stage-1 handover package as a plain dict.

    The dict contains both the metadata the subagent will receive AND the
    resolved briefing-file path. The caller (tool) is responsible for writing
    the briefing file via append_briefing() before invoking the subagent.

    Raises ValueError for unknown specialists, oversized payloads, or
    empty summary / topic (the latter is a soft warning, not a hard error,
    so the caller can decide).
    """
    if not is_known_specialist(specialist):
        raise ValueError(
            f"Unknown specialist profile: {specialist!r}. "
            f"Known specialists: {', '.join(list_known_specialists())}"
        )
    if not topic or not topic.strip():
        raise ValueError("topic must be a non-empty string")

    tokens = estimate_handover_tokens(
        summary=summary,
        recent_turns=recent_turns,
        decisions=decisions,
        artifacts=artifacts,
    )
    warn_oversize = False
    if tokens > max_context_tokens:
        # We do NOT raise here: the orchestrator may decide to truncate or
        # accept the cost. We do flag it.
        warn_oversize = True

    briefing_path = build_briefing_path(project_root, specialist, topic)
    package = {
        "specialist": specialist,
        "topic": topic,
        "summary": summary.strip(),
        "recent_turns": list(recent_turns or []),
        "decisions": list(decisions or []),
        "artifacts": list(artifacts or []),
        "return_condition": (return_condition or "user_request").strip(),
        "handover_reason": (handover_reason or "").strip(),
        "briefing_file": briefing_path,
        "estimated_tokens": tokens,
        "max_context_tokens": max_context_tokens,
        "oversize_warning": warn_oversize,
        "_meta": {
            "created_at": _timestamp_iso(),
            "phase": "phase-1-readonly",
        },
    }
    return package


def persona_marker(specialist: str) -> str:
    """Return a short, UI-friendly persona marker for `specialist`.

    The marker is intentionally compact so it can be prefixed to chat
    messages without inflating the visible output. Phase 3 may add a richer
    UI representation (icon, color, etc.).
    """
    safe = specialist.replace("_", " ").strip()
    return f"[{safe}]"