283 lines
9.3 KiB
Python
283 lines
9.3 KiB
Python
|
"""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",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 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}]"
|