validate_handoff.py

#!/usr/bin/env python3
"""
Location: pact-plugin/hooks/validate_handoff.py
Summary: SubagentStop hook that validates PACT agent/teammate handoff format.
Used by: Claude Code hooks.json SubagentStop hook (fires for both background
         Task agents and Agent Teams teammates)

Validates that PACT agents complete with proper handoff information
(produced, decisions, next steps) in their transcript text.

Note: Task protocol compliance (status, metadata) is NOT validated here.
Task state may still be in flux at SubagentStop time (agents self-manage
status under Agent Teams, and the team-lead may process output after this hook
fires), so Task state cannot be reliably checked here.

CANONICAL STRUCTURED HANDOFF (do NOT relocate this hook): this hook
validates the PROSE form of a HANDOFF in the agent transcript and is a
legacy, IN-PROCESS-only convenience. The STRUCTURED 6-field handoff lives in
`metadata.handoff`, and its PRESENCE is handled lead-side at acceptance-commit
by `_emit_lead_side_agent_handoff` in task_lifecycle_gate.py — its
emit-eligibility short-circuits on an absent handoff — and therefore fires in
BOTH teammate modes (in-process AND separate-process). (A completion-time
advisory branch that once emitted `handoff_missing` / `handoff_schema_invalid`
there was permanently dormant under the bare-owner convention and has been
retired.) This SubagentStop prose check does NOT fire for a separate-process
(e.g. tmux/iTerm2) teammate — such a teammate fires its OWN Stop/SessionEnd,
never a SubagentStop in the lead's process — so the prose nudge is simply
absent there. That absence is intentional and acceptable: the lead-side
presence handling above already covers both modes. Do NOT "restore" this prose
check onto a teammate end-of-life surface believing validation was lost — it
was not.

Input: JSON from stdin with `last_assistant_message` (preferred, SDK v2.1.47+),
       `transcript` (fallback), and `agent_type` (the role-class gate field, #812)
Output: JSON with `systemMessage` if handoff format is incomplete
"""

from __future__ import annotations

import json
import sys
import re

from shared.error_output import hook_error_json

# Suppress false "hook error" display in Claude Code UI on bare exit paths
_SUPPRESS_OUTPUT = json.dumps({"suppressOutput": True})


# Lossless fields — information that would be lost when agent context ends.
# When a structured HANDOFF section is present, these must appear as subsections.
LOSSLESS_FIELDS = {
    "produced": {
        "patterns": [
            r"(?:^|\n)\s*\d*\.?\s*produced\s*:",
        ],
        "description": "Produced",
    },
    "key_decisions": {
        "patterns": [
            r"(?:^|\n)\s*\d*\.?\s*key\s+decisions?\s*:",
        ],
        "description": "Key decisions",
    },
}

# Signal-type completions (e.g., pact-auditor) use audit_summary, not HANDOFF format
SIGNAL_COMPLETION_PATTERNS = [
    r"AUDIT\s+SIGNAL",
    r"audit_summary",
    r"completion_type.+signal",
]


# Required handoff elements with their patterns and descriptions
HANDOFF_ELEMENTS = {
    "what_produced": {
        "patterns": [
            r"(?:produced|created|generated|output|implemented|wrote|built|delivered)",
            r"(?:file|document|component|module|function|class|api|endpoint|schema)",
            r"(?:completed|finished|done with)",
        ],
        "description": "what was produced",
    },
    "key_decisions": {
        "patterns": [
            r"(?:decision|chose|selected|opted|rationale|reason|because)",
            r"(?:trade-?off|alternative|approach|strategy|pattern)",
            r"(?:decided to|went with|picked)",
        ],
        "description": "key decisions",
    },
    "next_steps": {
        "patterns": [
            r"(?:next|needs|requires|depends|should|must|recommend)",
            r"(?:follow-?up|remaining|todo|to-?do|action item)",
            r"(?:test engineer|tester|reviewer|next agent|next phase)",
        ],
        "description": "next steps/needs",
    },
}


def is_signal_completion(transcript: str) -> bool:
    """
    Check if transcript represents a signal-type completion (e.g., pact-auditor).

    Signal-type completions use audit_summary in task metadata rather than
    the standard HANDOFF format, so lossless field validation does not apply.

    Args:
        transcript: The agent's complete output/transcript

    Returns:
        True if this is a signal-type completion
    """
    for pattern in SIGNAL_COMPLETION_PATTERNS:
        if re.search(pattern, transcript, re.IGNORECASE):
            return True
    return False


def check_lossless_fields(transcript: str) -> list:
    """
    Check if a structured HANDOFF section contains the lossless fields.

    Lossless fields are information that would be lost when the agent's
    context window ends: what was produced and what decisions were made.

    Args:
        transcript: The agent's complete output/transcript

    Returns:
        List of missing lossless field descriptions (empty if all present)
    """
    missing_lossless = []
    transcript_lower = transcript.lower()

    for field_key, field_info in LOSSLESS_FIELDS.items():
        found = False
        for pattern in field_info["patterns"]:
            if re.search(pattern, transcript_lower):
                found = True
                break
        if not found:
            missing_lossless.append(field_info["description"])

    return missing_lossless


def validate_handoff(transcript: str) -> tuple:
    """
    Check if transcript contains proper handoff elements.

    Args:
        transcript: The agent's complete output/transcript

    Returns:
        Tuple of (is_valid, missing_elements, lossless_warnings)
        - is_valid: True if handoff passes validation
        - missing_elements: list of missing element descriptions
        - lossless_warnings: list of missing lossless field names (structured path only)
    """
    missing = []

    # First, check for explicit handoff section (indicates structured handoff)
    has_handoff_section = bool(re.search(
        r"(?:##?\s*)?(?:handoff|hand-off|hand off|summary|output|deliverables)[\s:]*\n",
        transcript,
        re.IGNORECASE
    ))

    # If there's an explicit handoff section, validate lossless fields
    if has_handoff_section:
        # Signal-type completions skip lossless validation
        if is_signal_completion(transcript):
            return True, [], []

        lossless_missing = check_lossless_fields(transcript)
        return True, [], lossless_missing

    # Otherwise, check for implicit handoff elements
    transcript_lower = transcript.lower()

    for element_key, element_info in HANDOFF_ELEMENTS.items():
        found = False
        for pattern in element_info["patterns"]:
            if re.search(pattern, transcript_lower):
                found = True
                break

        if not found:
            missing.append(element_info["description"])

    # Consider valid if at least 2 out of 3 elements are present
    # (some agents may not have explicit decisions if straightforward)
    is_valid = len(missing) <= 1

    return is_valid, missing, []


def is_pact_agent(agent_identifier: str) -> bool:
    """
    Check if the agent is a PACT framework agent (role-class gate).

    Args:
        agent_identifier: The agent's role identifier — under #812 this is the
            harness-set ``agent_type`` (e.g. ``"pact-preparer"``). The ``pact-``
            prefix family below matches ``agent_type`` values directly, so the
            same prefix-check answers the role-class question against the field
            that is actually present at SubagentStop.

    Returns:
        True if this is a PACT agent that should be validated
    """
    if not agent_identifier:
        return False

    pact_prefixes = ["pact-", "PACT-", "pact_", "PACT_"]
    return any(agent_identifier.startswith(prefix) for prefix in pact_prefixes)


def main():
    """
    Main entry point for the SubagentStop hook.

    Reads agent/teammate transcript from stdin and validates handoff format
    (prose) for PACT agents. Fires for both background Task agents and
    Agent Teams teammates. Outputs warning messages if validation fails.
    """
    try:
        # Read input from stdin
        try:
            input_data = json.load(sys.stdin)
        except json.JSONDecodeError:
            # No input or invalid JSON - can't validate
            print(_SUPPRESS_OUTPUT)
            sys.exit(0)

        # Prefer last_assistant_message (SDK v2.1.47+), fall back to transcript
        transcript = input_data.get("last_assistant_message", "") or input_data.get("transcript", "")
        # #812 role-class gate: key on the harness-set ``agent_type``, NOT
        # ``agent_id``. agent_id is ABSENT under the separate-process teammate
        # model (v4.4.0), so the prior agent_id-keyed check was DORMANT for all
        # teammates — this hook fires on SubagentStop (teammate-only) and the
        # teammate's agent_type (e.g. "pact-preparer") matches the pact- prefix.
        # The lead's "PACT:"-prefixed agent_type never reaches this hook
        # (SubagentStop is teammate-only). Fail-safe: advisory systemMessage, no
        # DENY — re-enabling it can only ADD a missing-HANDOFF warning.
        agent_type = input_data.get("agent_type", "")

        # Only validate PACT agents
        if not is_pact_agent(agent_type):
            print(_SUPPRESS_OUTPUT)
            sys.exit(0)

        warnings = []

        # Skip transcript validation if very short (likely an error case)
        if len(transcript) >= 100:
            is_valid, missing, lossless_missing = validate_handoff(transcript)

            if not is_valid and missing:
                warnings.append(
                    f"PACT Handoff Warning: Agent '{agent_type}' completed without "
                    f"proper handoff. Missing: {', '.join(missing)}. "
                    "Consider including: what was produced, key decisions, and next steps."
                )

            if lossless_missing:
                warnings.append(
                    f"PACT Lossless Field Warning: Agent '{agent_type}' HANDOFF "
                    f"section is missing: {', '.join(lossless_missing)}. "
                    "These fields preserve information that would otherwise be lost."
                )

        # Output warnings if any
        if warnings:
            output = {
                "systemMessage": " | ".join(warnings)
            }
            print(json.dumps(output))
        else:
            print(_SUPPRESS_OUTPUT)

        sys.exit(0)

    except Exception as e:
        # Don't block on errors - just warn
        print(f"Hook warning (validate_handoff): {e}", file=sys.stderr)
        print(hook_error_json("validate_handoff", e))
        sys.exit(0)


if __name__ == "__main__":
    main()