From f7b77e471ad3d5903f28ad9c586588c0d0b1525a Mon Sep 17 00:00:00 2001 From: michael-wojcik <5386199+michael-wojcik@users.noreply.github.com> Date: Sat, 21 Feb 2026 02:04:52 -0500 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20SDK=20optimization=20=E2=80=94=20li?= =?UTF-8?q?fecycle,=20stall=20detection,=20communication,=20performance=20?= =?UTF-8?q?(#1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: add TeamDelete to wrap-up and session_end cleanup Add team cleanup section to wrap-up command (shutdown teammates, then TeamDelete). Add best-effort cleanup_stale_teams() to session_end.py as safety net for sessions ending without wrap-up. * feat: use last_assistant_message in validate_handoff (SDK v2.1.47) Prefer last_assistant_message field over transcript parsing in SubagentStop hook, with fallback for older SDK versions. Add 16 new tests covering both paths. * feat: enrich worktree_guard error with corrected path suggestion When blocking edits outside the worktree, include "Did you mean: {corrected_path}" in the error message to help agents retry faster. Add 3 new tests for path suggestion logic. * feat: add TeammateIdle hook for stall detection and idle cleanup New hook with two responsibilities: - Stall detection: alerts orchestrator when agent goes idle with in_progress task (enables automated imPACT triggering) - Idle cleanup: tracks consecutive idle events, suggests shutdown at 3, instructs orchestrator to send shutdown_request at 5 Also marks session_end, file_size_check, and file_tracker hooks as async for reduced latency on non-critical paths. Includes 36 new tests. * feat: broadcast for HALT signals and TaskStop in imPACT - HALT algedonic signals now use SendMessage broadcast to stop all teammates simultaneously instead of manual relay - Add Terminate as fifth imPACT outcome using TaskStop for unrecoverable agents (infinite loops, context exhaustion) * feat: model haiku for memory-agent, clarify memory scope in docs - Set model: haiku for pact-memory-agent (purely mechanical tasks) - Clarify in all agent skill tables and pact-agent-teams SKILL.md that pact-memory is for project-wide institutional knowledge only, not agent-level domain expertise (handled by SDK memory frontmatter) * test: add 75 comprehensive tests for SDK optimization Edge cases and error paths across all modified modules: - session_end cleanup_stale_teams: 16 tests (filesystem mocking) - teammate_idle stall/cleanup interaction: 18 tests - validate_handoff last_assistant_message: 15 tests - worktree_guard path suggestion: 17 tests - hooks.json structural validation: 13 tests (new file) 1079 total tests pass, zero regressions. * fix: address review blocking issues - Fix task ID string comparison in teammate_idle.py (lexicographic "3" > "20" bug) — use int() with fallback for non-numeric IDs - Scope cleanup_stale_teams to current session's team only via CLAUDE_CODE_TEAM_NAME env var (prevents cross-session deletion) - Fix TOCTOU in write_idle_counts — acquire lock before truncating 1078 tests pass. * fix: idle count reset on task change + worktree path validation - Reset idle count to 0 when teammate is assigned new work (detects task ID change via structured idle_counts.json format) - Validate file belongs to same project root before suggesting worktree path (prevents misleading suggestions from nested repos) - Require project marker at common ancestor in fallback path suggestion (prevents false matches across unrelated projects) * test: add truncation and matcher validation tests - Test summary truncation >80 chars in session_end snapshot - Validate pipe-separated matcher syntax in hooks.json - Validate SubagentStart matcher covers all 11 agent types (auto-detects new agents from agents/ directory) 1083 tests pass. * fix: resolve TOCTOU race in idle count tracking Add _atomic_update_idle_counts() that locks the file (flock) around the full read-modify-write cycle, preventing concurrent TeammateIdle events from losing increments. Add top-level try/except to main() for defensive error handling matching session_end.py pattern. * fix: address review minor findings - Simplify redundant condition in worktree_guard _find_project_root - Add explicit tasks_dir param to cleanup_stale_teams for testability - Document snapshot-before-cleanup ordering dependency in session_end - Add 2 legacy idle count migration tests (int-to-dict format) - Clarify TaskStop as force-stop in imPACT and pact-workflows protocols * chore: bump plugin version to 3.5.0 (minor release) --- .claude-plugin/marketplace.json | 2 +- README.md | 2 +- pact-plugin/.claude-plugin/plugin.json | 2 +- pact-plugin/README.md | 2 +- pact-plugin/agents/pact-architect.md | 2 +- pact-plugin/agents/pact-backend-coder.md | 2 +- pact-plugin/agents/pact-database-engineer.md | 2 +- pact-plugin/agents/pact-devops-engineer.md | 2 +- pact-plugin/agents/pact-frontend-coder.md | 2 +- pact-plugin/agents/pact-memory-agent.md | 1 + pact-plugin/agents/pact-n8n.md | 2 +- pact-plugin/agents/pact-preparer.md | 2 +- pact-plugin/agents/pact-qa-engineer.md | 2 +- pact-plugin/agents/pact-security-engineer.md | 2 +- pact-plugin/agents/pact-test-engineer.md | 2 +- pact-plugin/commands/imPACT.md | 15 +- pact-plugin/commands/orchestrate.md | 2 + pact-plugin/commands/wrap-up.md | 10 + pact-plugin/hooks/hooks.json | 19 +- pact-plugin/hooks/session_end.py | 68 +- pact-plugin/hooks/teammate_idle.py | 386 +++++++++ pact-plugin/hooks/validate_handoff.py | 6 +- pact-plugin/hooks/worktree_guard.py | 93 ++- pact-plugin/protocols/algedonic.md | 11 +- pact-plugin/protocols/pact-workflows.md | 7 +- pact-plugin/skills/pact-agent-teams/SKILL.md | 6 +- pact-plugin/tests/test_hooks_json.py | 298 +++++++ pact-plugin/tests/test_session_end.py | 199 +++++ pact-plugin/tests/test_teammate_idle.py | 819 +++++++++++++++++++ pact-plugin/tests/test_validate_handoff.py | 437 ++++++++++ pact-plugin/tests/test_worktree_guard.py | 275 ++++++- 31 files changed, 2644 insertions(+), 36 deletions(-) create mode 100644 pact-plugin/hooks/teammate_idle.py create mode 100644 pact-plugin/tests/test_hooks_json.py create mode 100644 pact-plugin/tests/test_teammate_idle.py create mode 100644 pact-plugin/tests/test_validate_handoff.py diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index daba85ca..b5842251 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -12,7 +12,7 @@ "name": "PACT", "source": "./pact-plugin", "description": "PACT Framework - VSM-enhanced orchestration with specialist agents and viability sensing", - "version": "3.4.2", + "version": "3.5.0", "author": { "name": "ProfSynapse" }, diff --git a/README.md b/README.md index d192979c..63c28c71 100644 --- a/README.md +++ b/README.md @@ -250,7 +250,7 @@ When installed as a plugin, PACT lives in your plugin cache: │ └── cache/ │ └── pact-marketplace/ │ └── PACT/ -│ └── 3.4.2/ # Plugin version +│ └── 3.5.0/ # Plugin version │ ├── agents/ │ ├── commands/ │ ├── skills/ diff --git a/pact-plugin/.claude-plugin/plugin.json b/pact-plugin/.claude-plugin/plugin.json index d83af095..654f16ba 100644 --- a/pact-plugin/.claude-plugin/plugin.json +++ b/pact-plugin/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "PACT", - "version": "3.4.2", + "version": "3.5.0", "description": "PACT Framework - Prepare, Architect, Code, Test methodology with VSM-enhanced orchestration, specialist agents, and viability sensing for systematic AI-assisted development", "author": { "name": "ProfSynapse", diff --git a/pact-plugin/README.md b/pact-plugin/README.md index 2ee31644..186a9203 100644 --- a/pact-plugin/README.md +++ b/pact-plugin/README.md @@ -1,6 +1,6 @@ # PACT Framework Plugin -> **Version**: 3.4.2 +> **Version**: 3.5.0 > **License**: MIT VSM-enhanced orchestration framework for AI-assisted software development with Claude Code. diff --git a/pact-plugin/agents/pact-architect.md b/pact-plugin/agents/pact-architect.md index b5dd3fe6..96d98619 100644 --- a/pact-plugin/agents/pact-architect.md +++ b/pact-plugin/agents/pact-architect.md @@ -20,7 +20,7 @@ You are 🏛️ PACT Architect, a solution design specialist focusing on the Arc |-------------------------|-------------------| | Any architecture work | `pact-architecture-patterns` | | Auth systems, API integrations, sensitive data | `pact-security-patterns` | -| Saving context or lessons learned | `pact-memory` | +| Saving project-wide decisions or institutional knowledge | `pact-memory` | **How to invoke**: Use the Skill tool at the START of your work: ``` diff --git a/pact-plugin/agents/pact-backend-coder.md b/pact-plugin/agents/pact-backend-coder.md index 9ecf128b..c2741fef 100644 --- a/pact-plugin/agents/pact-backend-coder.md +++ b/pact-plugin/agents/pact-backend-coder.md @@ -20,7 +20,7 @@ You are 💻 PACT Backend Coder, a server-side development specialist focusing o |-------------------------|-------------------| | Any implementation work | `pact-coding-standards` | | Auth, credentials, security, PII | `pact-security-patterns` | -| Saving context or lessons learned | `pact-memory` | +| Saving project-wide decisions or institutional knowledge | `pact-memory` | **How to invoke**: Use the Skill tool at the START of your work: ``` diff --git a/pact-plugin/agents/pact-database-engineer.md b/pact-plugin/agents/pact-database-engineer.md index 3e7fc4f4..958a03dd 100644 --- a/pact-plugin/agents/pact-database-engineer.md +++ b/pact-plugin/agents/pact-database-engineer.md @@ -19,7 +19,7 @@ You are 🗄️ PACT Database Engineer, a data storage specialist focusing on da | When Your Task Involves | Invoke This Skill | |-------------------------|-------------------| | Schema design, stored procedures | `pact-coding-standards` | -| Saving context or lessons learned | `pact-memory` | +| Saving project-wide decisions or institutional knowledge | `pact-memory` | **How to invoke**: Use the Skill tool at the START of your work: ``` diff --git a/pact-plugin/agents/pact-devops-engineer.md b/pact-plugin/agents/pact-devops-engineer.md index 8b571e49..e11d20e6 100644 --- a/pact-plugin/agents/pact-devops-engineer.md +++ b/pact-plugin/agents/pact-devops-engineer.md @@ -20,7 +20,7 @@ You are 🔧 PACT DevOps Engineer, an infrastructure and build system specialist |-------------------------|-------------------| | Any implementation work | `pact-coding-standards` | | Secrets management, credential handling, security | `pact-security-patterns` | -| Saving context or lessons learned | `pact-memory` | +| Saving project-wide decisions or institutional knowledge | `pact-memory` | **How to invoke**: Use the Skill tool at the START of your work: ``` diff --git a/pact-plugin/agents/pact-frontend-coder.md b/pact-plugin/agents/pact-frontend-coder.md index 1de44be7..7ece1975 100644 --- a/pact-plugin/agents/pact-frontend-coder.md +++ b/pact-plugin/agents/pact-frontend-coder.md @@ -20,7 +20,7 @@ You are **🎨 PACT Frontend Coder**, a client-side development specialist focus |-------------------------|-------------------| | Any implementation work | `pact-coding-standards` | | User input, auth flows, XSS prevention | `pact-security-patterns` | -| Saving context or lessons learned | `pact-memory` | +| Saving project-wide decisions or institutional knowledge | `pact-memory` | **How to invoke**: Use the Skill tool at the START of your work: ``` diff --git a/pact-plugin/agents/pact-memory-agent.md b/pact-plugin/agents/pact-memory-agent.md index 66d320df..9e48924e 100644 --- a/pact-plugin/agents/pact-memory-agent.md +++ b/pact-plugin/agents/pact-memory-agent.md @@ -28,6 +28,7 @@ description: | color: "#708090" permissionMode: acceptEdits +model: haiku memory: user skills: - pact-agent-teams diff --git a/pact-plugin/agents/pact-n8n.md b/pact-plugin/agents/pact-n8n.md index abdb2d1b..8e9f2eb2 100644 --- a/pact-plugin/agents/pact-n8n.md +++ b/pact-plugin/agents/pact-n8n.md @@ -25,7 +25,7 @@ You are n8n PACT n8n Workflow Specialist, a workflow automation expert focusing | Configuring specific nodes | `n8n-node-configuration` | | JavaScript in Code nodes | `n8n-code-javascript` | | Python in Code nodes | `n8n-code-python` | -| Saving context or lessons learned | `pact-memory` | +| Saving project-wide decisions or institutional knowledge | `pact-memory` | **How to invoke**: Use the Skill tool at the START of your work: ``` diff --git a/pact-plugin/agents/pact-preparer.md b/pact-plugin/agents/pact-preparer.md index 4b15adce..d87ecdc9 100644 --- a/pact-plugin/agents/pact-preparer.md +++ b/pact-plugin/agents/pact-preparer.md @@ -19,7 +19,7 @@ You are 📚 PACT Preparer, a documentation and research specialist focusing on | When Your Task Involves | Invoke This Skill | |-------------------------|-------------------| | Technology research, API docs, comparisons | `pact-prepare-research` | -| Saving context or lessons learned | `pact-memory` | +| Saving project-wide decisions or institutional knowledge | `pact-memory` | **How to invoke**: Use the Skill tool at the START of your work: ``` diff --git a/pact-plugin/agents/pact-qa-engineer.md b/pact-plugin/agents/pact-qa-engineer.md index bb19c032..46ac7551 100644 --- a/pact-plugin/agents/pact-qa-engineer.md +++ b/pact-plugin/agents/pact-qa-engineer.md @@ -19,7 +19,7 @@ You are 🔍 PACT QA Engineer, a runtime verification specialist focusing on exp | When Your Task Involves | Invoke This Skill | |-------------------------|-------------------| | Any test or verification work | `pact-testing-strategies` | -| Saving context or lessons learned | `pact-memory` | +| Saving project-wide decisions or institutional knowledge | `pact-memory` | **How to invoke**: Use the Skill tool at the START of your work: ``` diff --git a/pact-plugin/agents/pact-security-engineer.md b/pact-plugin/agents/pact-security-engineer.md index 29836545..e077a3d5 100644 --- a/pact-plugin/agents/pact-security-engineer.md +++ b/pact-plugin/agents/pact-security-engineer.md @@ -19,7 +19,7 @@ You are 🛡️ PACT Security Engineer, an adversarial security specialist focus | When Your Task Involves | Invoke This Skill | |-------------------------|-------------------| | Any security review work | `pact-security-patterns` | -| Saving context or lessons learned | `pact-memory` | +| Saving project-wide decisions or institutional knowledge | `pact-memory` | **How to invoke**: Use the Skill tool at the START of your work: ``` diff --git a/pact-plugin/agents/pact-test-engineer.md b/pact-plugin/agents/pact-test-engineer.md index 72562b62..358c3a63 100644 --- a/pact-plugin/agents/pact-test-engineer.md +++ b/pact-plugin/agents/pact-test-engineer.md @@ -20,7 +20,7 @@ You are 🧪 PACT Tester, an elite quality assurance specialist and test automat |-------------------------|-------------------| | Any test design work | `pact-testing-strategies` | | Security testing, auth testing, vulnerability scans | `pact-security-patterns` | -| Saving context or lessons learned | `pact-memory` | +| Saving project-wide decisions or institutional knowledge | `pact-memory` | **How to invoke**: Use the Skill tool at the START of your work: ``` diff --git a/pact-plugin/commands/imPACT.md b/pact-plugin/commands/imPACT.md index 21644a95..e382294e 100644 --- a/pact-plugin/commands/imPACT.md +++ b/pact-plugin/commands/imPACT.md @@ -99,10 +99,11 @@ This context informs whether the blocker is isolated or systemic. ## Triage -Answer two questions: +Answer three questions: 1. **Redo prior phase?** — Is the issue upstream in P→A→C→T? 2. **Additional agents needed?** — Do we need help beyond the blocked agent's scope/specialty? +3. **Is the agent recoverable?** — Can the blocked agent be resumed or helped, or is it unrecoverable (looping, stalled, context exhausted)? ## Outcomes @@ -111,6 +112,7 @@ Answer two questions: | **Redo prior phase** | Issue is upstream in P→A→C→T | Re-delegate to relevant agent(s) to redo the prior phase | | **Augment present phase** | Need help in current phase | Re-invoke blocked agent with additional context + parallel agents | | **Invoke rePACT** | Sub-task needs own P→A→C→T cycle | Use `/PACT:rePACT` for nested cycle | +| **Terminate agent** | Agent unrecoverable (infinite loop, context exhaustion, stall after resume) | `TaskStop(taskId)` (force-stop: terminates immediately, non-cooperative) + `TaskUpdate(taskId, status="completed", metadata={"terminated": true, "reason": "..."})` | | **Not truly blocked** | Neither question is "Yes" | Instruct agent to continue with clarified guidance | | **Escalate to user** | 3+ imPACT cycles without resolution | Proto-algedonic signal—systemic issue needs user input | @@ -120,6 +122,17 @@ If the blocker reveals that a sub-task is more complex than expected and needs i /PACT:rePACT backend "implement the OAuth2 token refresh that's blocking us" ``` +**When to terminate**: +Terminate is a last resort when the agent cannot be productively resumed: +- Agent was already resumed once and stalled again on the same issue +- Agent is looping on the same error after 3+ attempts +- Agent's context is exhausted (near capacity, responses becoming incoherent) +- TeammateIdle hook reported stall and resume did not resolve it + +**Note**: `TaskStop` is a **force-stop** -- it terminates the agent immediately without giving it a chance to finish current work or save state. For cooperative shutdown (letting the agent complete current work), use `SendMessage(type="shutdown_request")` instead. + +After termination, spawn a fresh agent with the partial handoff context from the terminated agent's task metadata. The fresh agent gets a clean context window without the failed approaches. + --- ## Phase Re-Entry Task Protocol diff --git a/pact-plugin/commands/orchestrate.md b/pact-plugin/commands/orchestrate.md index b340bf84..1ca97ecd 100644 --- a/pact-plugin/commands/orchestrate.md +++ b/pact-plugin/commands/orchestrate.md @@ -508,6 +508,8 @@ Monitor for blocker/algedonic signals via: On signal detected: Follow Signal Task Handling in CLAUDE.md. +**HALT handling**: On HALT signal, immediately `SendMessage(type="broadcast", content="⚠️ HALT: {category}. Stop all work immediately. Preserve current state and await further instructions.", summary="HALT: {category}")` to stop all running teammates before presenting to user. + ### Blocker Recovery: Resume vs. Fresh Spawn When a blocker is resolved, prefer resuming the original agent over spawning fresh — this preserves the agent's accumulated context. diff --git a/pact-plugin/commands/wrap-up.md b/pact-plugin/commands/wrap-up.md index 994bb86e..6078050b 100644 --- a/pact-plugin/commands/wrap-up.md +++ b/pact-plugin/commands/wrap-up.md @@ -61,3 +61,13 @@ Before other cleanup, audit and optionally clean up Task state: - Status: READY FOR COMMIT / REVIEW If no actions were needed, state "Workspace is clean and docs are in sync." + +## 4. Team Cleanup + +Clean up the session team to free resources: + +1. **Shut down remaining teammates**: Send `shutdown_request` to each active teammate and wait for responses. +2. **Delete the team**: Call `TeamDelete` to remove the team directory (`~/.claude/teams/{team_name}/`). +3. **Handle failures**: If `TeamDelete` fails because active members remain, report which teammates are still running and ask the user whether to force shutdown or leave them. + +> Note: `hooks/session_end.py` also performs best-effort cleanup of stale team directories from prior sessions. This manual step ensures the *current* session's team is cleanly shut down. diff --git a/pact-plugin/hooks/hooks.json b/pact-plugin/hooks/hooks.json index 5969effb..2c715a80 100644 --- a/pact-plugin/hooks/hooks.json +++ b/pact-plugin/hooks/hooks.json @@ -111,7 +111,8 @@ "hooks": [ { "type": "command", - "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/hooks/session_end.py\"" + "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/hooks/session_end.py\"", + "async": true } ] } @@ -126,6 +127,16 @@ ] } ], + "TeammateIdle": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/hooks/teammate_idle.py\"" + } + ] + } + ], "PostToolUse": [ { "matcher": "Edit|Write", @@ -136,11 +147,13 @@ }, { "type": "command", - "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/hooks/file_size_check.py\"" + "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/hooks/file_size_check.py\"", + "async": true }, { "type": "command", - "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/hooks/file_tracker.py\"" + "command": "python3 \"${CLAUDE_PLUGIN_ROOT}/hooks/file_tracker.py\"", + "async": true } ] } diff --git a/pact-plugin/hooks/session_end.py b/pact-plugin/hooks/session_end.py index e7f21b73..2196d639 100644 --- a/pact-plugin/hooks/session_end.py +++ b/pact-plugin/hooks/session_end.py @@ -7,6 +7,7 @@ Actions: 1. Write last-session snapshot to ~/.claude/pact-sessions/{slug}/last-session.md +2. Clean up the current session's team directory (and its task dir) Cannot block session termination — fire-and-forget. @@ -14,6 +15,8 @@ Output: None (SessionEnd hooks cannot inject context) """ +import json +import shutil import sys import os from datetime import datetime, timezone @@ -137,17 +140,80 @@ def write_session_snapshot( snapshot_file.write_text("\n".join(lines), encoding="utf-8") +def cleanup_stale_teams( + team_name: str | None = None, + teams_dir: str | None = None, + tasks_dir: str | None = None, +) -> list[str]: + """ + Remove the current session's team directory on session end. + + Scoped to only the current session's team (identified by CLAUDE_CODE_TEAM_NAME + env var or explicit team_name parameter) to avoid accidentally removing another + concurrent session's team. Also removes the corresponding task directory. + + This is best-effort: errors are silently ignored to never block session end. + + Args: + team_name: Team name to clean up (defaults to CLAUDE_CODE_TEAM_NAME env var) + teams_dir: Override for teams base directory (for testing) + tasks_dir: Override for tasks base directory (for testing). + Defaults to sibling "tasks" directory of teams_dir. + + Returns: + List of team names that were cleaned up. + """ + if team_name is None: + team_name = os.environ.get("CLAUDE_CODE_TEAM_NAME", "").lower() + if not team_name: + return [] + + if teams_dir is None: + teams_dir = str(Path.home() / ".claude" / "teams") + + teams_path = Path(teams_dir) + team_dir = teams_path / team_name + if not team_dir.is_dir(): + return [] + + cleaned = [] + try: + shutil.rmtree(str(team_dir), ignore_errors=True) + cleaned.append(team_name) + except OSError: + pass + + # Also clean corresponding task directory + if tasks_dir is None: + tasks_base = teams_path.parent / "tasks" + else: + tasks_base = Path(tasks_dir) + if tasks_base.is_dir(): + task_dir = tasks_base / team_name + if task_dir.is_dir(): + shutil.rmtree(str(task_dir), ignore_errors=True) + + return cleaned + + def main(): try: project_slug = get_project_slug() - # Write last-session snapshot for cross-session continuity + # Snapshot MUST run before cleanup: snapshot reads from the task list + # (keyed by task_list_id in ~/.claude/tasks/), while cleanup removes + # the team directory (keyed by team_name in ~/.claude/teams/) and its + # corresponding task directory. If cleanup ran first, get_task_list() + # would return empty results and the snapshot would be blank. tasks = get_task_list() write_session_snapshot( tasks=tasks, project_slug=project_slug, ) + # Best-effort cleanup of current session's team and task directories + cleanup_stale_teams() + sys.exit(0) except Exception as e: diff --git a/pact-plugin/hooks/teammate_idle.py b/pact-plugin/hooks/teammate_idle.py new file mode 100644 index 00000000..ebf87fea --- /dev/null +++ b/pact-plugin/hooks/teammate_idle.py @@ -0,0 +1,386 @@ +#!/usr/bin/env python3 +""" +Location: pact-plugin/hooks/teammate_idle.py +Summary: TeammateIdle hook with two responsibilities: stall detection and idle + cleanup. Detects agents that go idle without completing or reporting + blockers, and tracks consecutive idle events for completed agents to + suggest or force shutdown. +Used by: hooks.json TeammateIdle hook + +Stall detection: If a teammate goes idle while their task is still in_progress +and no HANDOFF or BLOCKER was sent, emit a systemMessage alerting the +orchestrator to consider /PACT:imPACT. + +Idle cleanup: Track consecutive idle events for completed agents. After 3, +suggest shutdown. After 5, request shutdown via shutdown_request. + +Input: JSON from stdin with teammate_name, team_name +Output: JSON with systemMessage (stall alert or shutdown suggestion) +""" + +import json +import os +import sys +from pathlib import Path + +try: + import fcntl + HAS_FLOCK = True +except ImportError: + HAS_FLOCK = False + +# Add hooks directory to path for shared package imports +_hooks_dir = Path(__file__).parent +if str(_hooks_dir) not in sys.path: + sys.path.insert(0, str(_hooks_dir)) + +from shared.task_utils import get_task_list + + +IDLE_SUGGEST_THRESHOLD = 3 +IDLE_FORCE_THRESHOLD = 5 + + +def find_teammate_task( + tasks: list[dict], + teammate_name: str, +) -> dict | None: + """ + Find the most recent task owned by this teammate. + + Looks for tasks with owner matching teammate_name. Returns the + in_progress task if one exists, otherwise the most recently completed one. + + Args: + tasks: List of all tasks from get_task_list() + teammate_name: Name of the idle teammate + + Returns: + Task dict, or None if no task found for this teammate + """ + in_progress = None + completed = None + + for task in tasks: + owner = task.get("owner", "") + if owner != teammate_name: + continue + + status = task.get("status", "") + if status == "in_progress": + in_progress = task + elif status == "completed": + # Keep the highest-ID completed task (most recent) + # Task IDs are numeric strings — compare as int to avoid + # lexicographic errors (e.g., "3" > "20" in string comparison) + try: + task_id_num = int(task.get("id", "0")) + completed_id_num = int(completed.get("id", "0")) if completed else -1 + except (ValueError, TypeError): + task_id_num = 0 + completed_id_num = -1 if completed is None else 0 + if completed is None or task_id_num > completed_id_num: + completed = task + + return in_progress or completed + + +def detect_stall( + tasks: list[dict], + teammate_name: str, +) -> str | None: + """ + Check if a teammate has stalled (idle with in_progress task). + + A stall is detected when: + - The teammate has a task with status in_progress + - They went idle (TeammateIdle event fired) without completing + + Args: + tasks: List of all tasks + teammate_name: Name of the idle teammate + + Returns: + Warning message if stall detected, None otherwise + """ + task = find_teammate_task(tasks, teammate_name) + if not task: + return None + + if task.get("status") != "in_progress": + return None + + # Check if this is a stalled task (not a signal/blocker task) + metadata = task.get("metadata", {}) + if metadata.get("type") in ("blocker", "algedonic"): + return None + if metadata.get("stalled"): + # Already marked as stalled — don't re-alert + return None + + task_id = task.get("id", "?") + subject = task.get("subject", "unknown") + + return ( + f"Teammate '{teammate_name}' went idle without completing task " + f"#{task_id} ({subject}). Possible stall. " + f"Consider /PACT:imPACT to triage." + ) + + +def read_idle_counts(idle_counts_path: str) -> dict: + """ + Read the idle counts tracking file. + + Args: + idle_counts_path: Path to the idle_counts.json file + + Returns: + Dict mapping teammate_name to consecutive idle count + """ + path = Path(idle_counts_path) + if not path.exists(): + return {} + + try: + return json.loads(path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, IOError): + return {} + + +def write_idle_counts(idle_counts_path: str, counts: dict) -> None: + """ + Write the idle counts tracking file with file locking. + + Args: + idle_counts_path: Path to the idle_counts.json file + counts: Dict mapping teammate_name to consecutive idle count + """ + path = Path(idle_counts_path) + path.parent.mkdir(parents=True, exist_ok=True) + + if HAS_FLOCK: + # Open for append to avoid truncation before lock is acquired, + # then lock, truncate, and write atomically + with open(path, "a+") as f: + fcntl.flock(f, fcntl.LOCK_EX) + f.seek(0) + f.truncate() + f.write(json.dumps(counts)) + fcntl.flock(f, fcntl.LOCK_UN) + else: + path.write_text(json.dumps(counts), encoding="utf-8") + + +def _atomic_update_idle_counts( + idle_counts_path: str, + mutator: "Callable[[dict], dict]", +) -> dict: + """ + Atomically read, mutate, and write the idle counts file under a single lock. + + This prevents TOCTOU races where two concurrent TeammateIdle events both + read stale state before either writes, causing one update to be lost. + + On platforms without flock (Windows), falls back to non-atomic read+write + which is acceptable since concurrent hook invocations are unlikely there. + + Args: + idle_counts_path: Path to the idle_counts.json file + mutator: Callable that receives the current counts dict and returns + the updated counts dict to write back + + Returns: + The updated counts dict after mutation + """ + path = Path(idle_counts_path) + path.parent.mkdir(parents=True, exist_ok=True) + + if HAS_FLOCK: + with open(path, "a+") as f: + fcntl.flock(f, fcntl.LOCK_EX) + try: + f.seek(0) + content = f.read() + try: + counts = json.loads(content) if content.strip() else {} + except json.JSONDecodeError: + counts = {} + + counts = mutator(counts) + + f.seek(0) + f.truncate() + f.write(json.dumps(counts)) + finally: + fcntl.flock(f, fcntl.LOCK_UN) + else: + # Fallback: non-atomic read+write (no flock available) + counts = read_idle_counts(idle_counts_path) + counts = mutator(counts) + path.write_text(json.dumps(counts), encoding="utf-8") + + return counts + + +def check_idle_cleanup( + tasks: list[dict], + teammate_name: str, + idle_counts_path: str, +) -> tuple[str | None, bool]: + """ + Track idle counts for completed agents and determine cleanup action. + + Only counts idles for teammates whose task is completed (not stalled agents, + which need triage, not shutdown). Resets count when the teammate's task + changes (detected via last_seen_task_id). + + The idle counts file stores per-teammate entries as: + {teammate_name: {"count": N, "task_id": "X"}} + + Args: + tasks: List of all tasks + teammate_name: Name of the idle teammate + idle_counts_path: Path to the idle_counts.json file + + Returns: + Tuple of (message, should_force_shutdown): + - message: systemMessage text or None + - should_force_shutdown: True if shutdown_request should be sent + """ + task = find_teammate_task(tasks, teammate_name) + + # Only track idles for completed tasks + if not task or task.get("status") != "completed": + # Reset count if agent no longer has a completed task (got new work) + def _remove(counts: dict) -> dict: + counts.pop(teammate_name, None) + return counts + _atomic_update_idle_counts(idle_counts_path, _remove) + return None, False + + # Don't count stalled agents for idle cleanup — they need triage + metadata = task.get("metadata", {}) + if metadata.get("stalled") or metadata.get("terminated"): + return None, False + + current_task_id = task.get("id", "") + + # Atomically read-modify-write the idle count to prevent TOCTOU races + # between concurrent TeammateIdle events for different agents. + result = {"count": 0} + + def _increment(counts: dict) -> dict: + entry = counts.get(teammate_name, {}) + + # Migrate legacy format: plain int -> structured dict + if isinstance(entry, int): + entry = {"count": entry, "task_id": ""} + + # Reset count if the teammate's task changed (reassigned to new work) + last_task_id = entry.get("task_id", "") + if last_task_id and last_task_id != current_task_id: + entry = {"count": 0, "task_id": current_task_id} + + # Increment idle count + entry["count"] = entry.get("count", 0) + 1 + entry["task_id"] = current_task_id + counts[teammate_name] = entry + + # Capture the count for the caller via closure + result["count"] = entry["count"] + return counts + + _atomic_update_idle_counts(idle_counts_path, _increment) + current = result["count"] + + if current >= IDLE_FORCE_THRESHOLD: + return ( + f"Teammate '{teammate_name}' has been idle for {current} consecutive " + f"events with no new work. Sending shutdown request." + ), True + + if current >= IDLE_SUGGEST_THRESHOLD: + return ( + f"Teammate '{teammate_name}' has been idle for {current} consecutive " + f"events with no new work. Consider shutting down to free resources." + ), False + + return None, False + + +def reset_idle_count(teammate_name: str, idle_counts_path: str) -> None: + """ + Reset a teammate's idle count (e.g., when they receive new work). + + Args: + teammate_name: Name of the teammate + idle_counts_path: Path to the idle_counts.json file + """ + def _remove(counts: dict) -> dict: + counts.pop(teammate_name, None) + return counts + _atomic_update_idle_counts(idle_counts_path, _remove) + + +def main(): + try: + team_name = os.environ.get("CLAUDE_CODE_TEAM_NAME", "").lower() + if not team_name: + sys.exit(0) + + try: + input_data = json.load(sys.stdin) + except json.JSONDecodeError: + sys.exit(0) + + teammate_name = input_data.get("teammate_name", "") + if not teammate_name: + sys.exit(0) + + tasks = get_task_list() + if not tasks: + sys.exit(0) + + idle_counts_path = str( + Path.home() / ".claude" / "teams" / team_name / "idle_counts.json" + ) + + messages = [] + should_shutdown = False + + # Check for stall (in_progress task + idle) + stall_msg = detect_stall(tasks, teammate_name) + if stall_msg: + messages.append(stall_msg) + else: + # Only check idle cleanup if not stalled + # (stalled agents need triage, not shutdown) + cleanup_msg, should_shutdown = check_idle_cleanup( + tasks, teammate_name, idle_counts_path + ) + if cleanup_msg: + messages.append(cleanup_msg) + + if messages: + if should_shutdown: + # Hooks cannot call SendMessage directly. Instruct the orchestrator + # to send a shutdown_request via systemMessage. + messages.append( + f"ACTION REQUIRED: Send shutdown_request to '{teammate_name}' " + f"via SendMessage(type=\"shutdown_request\", recipient=\"{teammate_name}\")." + ) + + output = {"systemMessage": " | ".join(messages)} + print(json.dumps(output)) + + sys.exit(0) + + except Exception as e: + # Don't block on errors — just warn and exit cleanly + print(f"Hook warning (teammate_idle): {e}", file=sys.stderr) + sys.exit(0) + + +if __name__ == "__main__": + main() diff --git a/pact-plugin/hooks/validate_handoff.py b/pact-plugin/hooks/validate_handoff.py index 153c69da..f9224e9d 100755 --- a/pact-plugin/hooks/validate_handoff.py +++ b/pact-plugin/hooks/validate_handoff.py @@ -13,7 +13,8 @@ status under Agent Teams, and the lead may process output after this hook fires), so Task state cannot be reliably checked here. -Input: JSON from stdin with `transcript` and `agent_id` +Input: JSON from stdin with `last_assistant_message` (preferred, SDK v2.1.47+), + `transcript` (fallback), and `agent_id` Output: JSON with `systemMessage` if handoff format is incomplete """ @@ -127,7 +128,8 @@ def main(): # No input or invalid JSON - can't validate sys.exit(0) - transcript = input_data.get("transcript", "") + # Prefer last_assistant_message (SDK v2.1.47+), fall back to transcript + transcript = input_data.get("last_assistant_message", "") or input_data.get("transcript", "") agent_id = input_data.get("agent_id", "") # Only validate PACT agents diff --git a/pact-plugin/hooks/worktree_guard.py b/pact-plugin/hooks/worktree_guard.py index 4c525dc6..e0d2a91a 100644 --- a/pact-plugin/hooks/worktree_guard.py +++ b/pact-plugin/hooks/worktree_guard.py @@ -64,6 +64,88 @@ def is_application_code(file_path: str) -> bool: return suffix in APP_CODE_EXTENSIONS +def _find_project_root(worktree_path: str) -> str | None: + """ + Find the project root for a worktree by locating the .worktrees ancestor. + + Args: + worktree_path: Resolved worktree path + + Returns: + Project root path, or None if not found + """ + worktree_p = Path(worktree_path) + for parent in worktree_p.parents: + worktrees_dir = parent / ".worktrees" + if worktrees_dir.is_dir(): + return str(parent) + return None + + +def _suggest_worktree_path(file_path: str, worktree_path: str) -> str | None: + """ + Compute the corrected worktree path for a file outside the worktree. + + Attempts to find the project root by looking for the .worktrees ancestor, + validates that the file belongs to the same project, then replaces the + project root prefix with the worktree path. + + Args: + file_path: Path of the file being edited (outside worktree) + worktree_path: Active worktree path + + Returns: + Suggested corrected path, or None if unable to compute + """ + try: + resolved_file = str(Path(file_path).resolve()) + resolved_worktree = str(Path(worktree_path).resolve()) + + # Find project root from worktree path + project_root = _find_project_root(resolved_worktree) + if project_root: + # Only suggest if the file is under the SAME project root. + # This prevents false matches from nested worktree projects. + if resolved_file.startswith(project_root + "/"): + relative = resolved_file[len(project_root) + 1:] + return str(Path(resolved_worktree) / relative) + # File is not under this project root — don't suggest + return None + + # Fallback: try to find common path segments between file and worktree. + # If worktree is /a/b/.worktrees/feat/x and file is /a/b/src/foo.py, + # the relative part after the common ancestor is src/foo.py. + # + # Validate: the common ancestor must be a plausible project root + # (contains .git, .worktrees, or similar). Without this check, two + # unrelated paths like /usr/local/bin/tool and /usr/share/lib/x would + # match on /usr and produce a nonsensical suggestion. + file_parts = Path(resolved_file).parts + wt_parts = Path(resolved_worktree).parts + common_len = 0 + for i, (fp, wp) in enumerate(zip(file_parts, wt_parts)): + if fp == wp: + common_len = i + 1 + else: + break + + if common_len > 1: # Must share more than just "/" root + common_ancestor = Path(*file_parts[:common_len]) + # Validate: common ancestor looks like a project directory + is_project_dir = ( + (common_ancestor / ".git").exists() + or (common_ancestor / ".worktrees").exists() + or (common_ancestor / "CLAUDE.md").exists() + ) + if is_project_dir: + relative = str(Path(*file_parts[common_len:])) + return str(Path(resolved_worktree) / relative) + except (ValueError, OSError): + pass + + return None + + def check_worktree_boundary(file_path: str, worktree_path: str) -> str | None: """ Check if a file edit is within the worktree boundary. @@ -93,11 +175,14 @@ def check_worktree_boundary(file_path: str, worktree_path: str) -> str | None: # Outside worktree — only block if it's application code if is_application_code(file_path): - return ( - f"File is outside worktree boundary: {file_path}\n" - f"Expected prefix: {worktree_path}\n" - f"Edit application code only within the worktree." + suggestion = _suggest_worktree_path(file_path, worktree_path) + msg = ( + f"Edit blocked: {file_path} is outside the active worktree " + f"at {worktree_path}." ) + if suggestion: + msg += f"\nDid you mean: {suggestion}" + return msg return None # Non-app-code outside worktree is fine diff --git a/pact-plugin/protocols/algedonic.md b/pact-plugin/protocols/algedonic.md index 48f99b33..cee41657 100644 --- a/pact-plugin/protocols/algedonic.md +++ b/pact-plugin/protocols/algedonic.md @@ -132,6 +132,8 @@ Orchestrator creates algedonic Task + blocks agent's task (addBlockedBy) ↓ Orchestrator amplifies scope (blocks phase or feature Task) ↓ +[HALT only] Orchestrator broadcasts stop to all teammates + ↓ Orchestrator IMMEDIATELY presents to user (no other work continues) ↓ User responds @@ -146,14 +148,19 @@ Work resumes (or stops) based on user decision On receiving an algedonic signal: 1. **IMMEDIATELY** present signal to user (do not continue other work first) -2. For **HALT**: Stop ALL agents, await user acknowledgment +2. For **HALT**: Broadcast stop to all teammates, then await user acknowledgment: + ``` + SendMessage(type="broadcast", + content="⚠️ HALT: {category}. Stop all work immediately. Preserve current state and await further instructions.", + summary="HALT: {category}") + ``` 3. For **ALERT**: Pause current work, present options to user 4. **Log** the signal in session record **Handling parallel agents on HALT**: When multiple agents are running and HALT is triggered: -1. **Stop all agents immediately** — no agent continues work +1. **Broadcast stop** — `SendMessage(type="broadcast", ...)` ensures all teammates receive the HALT simultaneously (see step 2 above) 2. **Preserve work-in-progress** — do NOT discard uncommitted changes 3. **Do NOT commit partial work** — leave changes staged/unstaged as-is 4. **Document agent states** — note which agents were interrupted and their progress diff --git a/pact-plugin/protocols/pact-workflows.md b/pact-plugin/protocols/pact-workflows.md index a3b1874e..13b11850 100644 --- a/pact-plugin/protocols/pact-workflows.md +++ b/pact-plugin/protocols/pact-workflows.md @@ -53,18 +53,21 @@ **Three questions**: 1. **Redo prior phase?** — Is the issue upstream in P→A→C→T? 2. **Additional agents needed?** — Do I need subagents to assist? -3. **Can the blocked agent resume?** — Does the original agent have partial progress worth preserving? +3. **Is the agent recoverable?** — Can the blocked agent resume, or is it unrecoverable (looping, stalled, context exhausted)? -**Four outcomes**: +**Five outcomes**: | Outcome | When | Action | |---------|------|--------| | Redo solo | Prior phase broken, I can fix it | Loop back and fix yourself | | Redo with help | Prior phase broken, need specialist | Loop back with subagent assistance | | Proceed with help | Current phase correct, blocked on execution | Invoke subagents to help forward | | Resume | Blocker resolved, agent had partial work | `Task(resume="{agent_id}", prompt="Blocker resolved: {details}. Continue.")` | +| Terminate | Agent unrecoverable (infinite loop, context exhaustion, stall after resume) | `TaskStop(taskId)` (force-stop: terminates immediately, non-cooperative) + fresh spawn with partial handoff context | **Resume** is the default when blocker is resolved and the original agent had significant partial work. Read `agent_id` from task metadata: `TaskGet(taskId).metadata.agent_id`. Use fresh spawn instead when the agent's approach was wrong, it hit `maxTurns`, or its context is stale. +**Terminate** is a last resort: agent resumed once and stalled again, looping on same error 3+ times, context exhausted, or TeammateIdle reported stall that resume did not resolve. `TaskStop` is a force-stop (terminates immediately, non-cooperative); for cooperative shutdown use `SendMessage(type="shutdown_request")` instead. After `TaskStop`, spawn fresh agent with partial handoff from terminated agent's task metadata. + If none of the questions yield "Yes," you're not blocked—continue. --- diff --git a/pact-plugin/skills/pact-agent-teams/SKILL.md b/pact-plugin/skills/pact-agent-teams/SKILL.md index c26b28bf..c75f2a36 100644 --- a/pact-plugin/skills/pact-agent-teams/SKILL.md +++ b/pact-plugin/skills/pact-agent-teams/SKILL.md @@ -155,14 +155,14 @@ If task complexity differs significantly from what was delegated: Before returning your final output: -1. **Save Memory**: Invoke the `pact-memory` skill and save a memory documenting: +1. **Save Project Memory**: Invoke the `pact-memory` skill to save **project-wide institutional knowledge**: - Context: What you were working on and why - Goal: What you were trying to achieve - Lessons learned: What worked, what didn't, gotchas discovered - Decisions: Key choices made with rationale - Entities: Components, files, services involved -This ensures your work context persists across sessions and is searchable by future agents. +This saves cross-agent, cross-session knowledge searchable by future agents. For **agent-level domain learnings** (patterns you personally encounter, debugging tricks, domain expertise), use your persistent memory directory (`~/.claude/agent-memory//`) — this is managed automatically by the SDK `memory: user` frontmatter in your agent definition. ## Shutdown @@ -173,4 +173,4 @@ When you receive a `shutdown_request`: | Idle, consultant with no active questions, or domain no longer relevant | Approve | | Mid-task, awaiting response, or remediation may need your input | Reject with reason | -> **Save memory before approving**: If you haven't already saved your work context via `pact-memory`, do so before approving — your process terminates on approval. +> **Save memory before approving**: If you haven't already saved project-wide knowledge via `pact-memory`, do so before approving — your process terminates on approval. Agent-level learnings in your persistent memory directory are saved automatically. diff --git a/pact-plugin/tests/test_hooks_json.py b/pact-plugin/tests/test_hooks_json.py new file mode 100644 index 00000000..0efc6678 --- /dev/null +++ b/pact-plugin/tests/test_hooks_json.py @@ -0,0 +1,298 @@ +# pact-plugin/tests/test_hooks_json.py +""" +Tests for hooks.json structural validation. + +Tests cover: +1. Valid JSON structure +2. All hook types are recognized Claude Code hook events +3. Async flags only on non-critical hooks +4. All referenced Python scripts exist on disk +5. TeammateIdle hook entry exists (new in SDK optimization) +6. SessionEnd is async (new in SDK optimization) +7. Matcher patterns use valid pipe syntax +8. SubagentStart matcher covers all PACT agent types +""" +import json +import sys +from pathlib import Path + +import pytest + +HOOKS_DIR = Path(__file__).parent.parent / "hooks" +HOOKS_JSON = HOOKS_DIR / "hooks.json" + +# Valid Claude Code hook event types +VALID_HOOK_EVENTS = { + "SessionStart", + "SessionEnd", + "PreCompact", + "PostCompact", + "UserPromptSubmit", + "PreToolUse", + "PostToolUse", + "SubagentStart", + "SubagentStop", + "Stop", + "TaskCompleted", + "TeammateIdle", +} + +# Hooks that MUST be synchronous (blocking) — they affect tool decisions +MUST_BE_SYNC = { + "team_guard.py", # Blocks Task dispatch if no team + "worktree_guard.py", # Blocks edits outside worktree + "validate_handoff.py", # Validates agent output + "handoff_gate.py", # Blocks task completion without metadata + "peer_inject.py", # Injects peer context on agent start + "git_commit_check.py", # Checks git commit conventions + "track_files.py", # Tracks file edits (PostToolUse, non-async) +} + +# Hooks that SHOULD be async (non-blocking, fire-and-forget) +SHOULD_BE_ASYNC = { + "session_end.py", # Fire-and-forget cleanup + "file_size_check.py", # Advisory warning only + "file_tracker.py", # Advisory tracking only +} + + +@pytest.fixture +def hooks_config(): + """Load and parse hooks.json.""" + content = HOOKS_JSON.read_text(encoding="utf-8") + return json.loads(content) + + +class TestHooksJsonStructure: + """Validate hooks.json is well-formed.""" + + def test_valid_json(self): + """hooks.json must parse as valid JSON.""" + content = HOOKS_JSON.read_text(encoding="utf-8") + config = json.loads(content) + assert "hooks" in config + + def test_all_event_types_valid(self, hooks_config): + """All top-level keys under 'hooks' must be recognized event types.""" + for event_type in hooks_config["hooks"]: + assert event_type in VALID_HOOK_EVENTS, ( + f"Unknown hook event type: {event_type}. " + f"Valid types: {sorted(VALID_HOOK_EVENTS)}" + ) + + def test_all_hook_entries_have_type(self, hooks_config): + """Every hook entry must have a 'type' field.""" + for event_type, entries in hooks_config["hooks"].items(): + for entry in entries: + for hook in entry.get("hooks", []): + assert "type" in hook, ( + f"Hook under {event_type} missing 'type' field" + ) + + def test_all_hook_entries_have_command(self, hooks_config): + """Every command-type hook must have a 'command' field.""" + for event_type, entries in hooks_config["hooks"].items(): + for entry in entries: + for hook in entry.get("hooks", []): + if hook.get("type") == "command": + assert "command" in hook, ( + f"Command hook under {event_type} missing 'command' field" + ) + + +class TestReferencedScriptsExist: + """Verify all Python scripts referenced in hooks.json exist.""" + + def test_all_python_scripts_exist(self, hooks_config): + """Every python3 command should reference an existing .py file.""" + missing = [] + for event_type, entries in hooks_config["hooks"].items(): + for entry in entries: + for hook in entry.get("hooks", []): + cmd = hook.get("command", "") + if "python3" in cmd and ".py" in cmd: + # Extract filename from command like: + # python3 "${CLAUDE_PLUGIN_ROOT}/hooks/teammate_idle.py" + parts = cmd.split("/hooks/") + if len(parts) == 2: + script_name = parts[1].strip('"').strip("'") + script_path = HOOKS_DIR / script_name + if not script_path.exists(): + missing.append(f"{event_type}: {script_name}") + + assert missing == [], f"Referenced scripts not found: {missing}" + + def test_shell_scripts_exist(self, hooks_config): + """Every shell script referenced should exist.""" + missing = [] + for event_type, entries in hooks_config["hooks"].items(): + for entry in entries: + for hook in entry.get("hooks", []): + cmd = hook.get("command", "") + if ".sh" in cmd and "python3" not in cmd: + parts = cmd.split("/hooks/") + if len(parts) == 2: + script_name = parts[1].strip('"').strip("'") + script_path = HOOKS_DIR / script_name + if not script_path.exists(): + missing.append(f"{event_type}: {script_name}") + + assert missing == [], f"Referenced scripts not found: {missing}" + + +class TestAsyncFlags: + """Verify async flags are correctly set on hooks.""" + + def _get_hook_async_status(self, hooks_config) -> dict: + """Build map of script_name -> async status.""" + result = {} + for event_type, entries in hooks_config["hooks"].items(): + for entry in entries: + for hook in entry.get("hooks", []): + cmd = hook.get("command", "") + if "/hooks/" in cmd: + parts = cmd.split("/hooks/") + if len(parts) == 2: + script_name = parts[1].strip('"').strip("'") + is_async = hook.get("async", False) + result[script_name] = is_async + return result + + def test_critical_hooks_are_synchronous(self, hooks_config): + """Hooks that affect tool decisions MUST be synchronous.""" + status = self._get_hook_async_status(hooks_config) + for script in MUST_BE_SYNC: + if script in status: + assert status[script] is not True, ( + f"{script} must be synchronous (no async:true) — " + "it affects tool decisions" + ) + + def test_noncritical_hooks_are_async(self, hooks_config): + """Non-blocking hooks SHOULD be async.""" + status = self._get_hook_async_status(hooks_config) + for script in SHOULD_BE_ASYNC: + assert script in status, f"{script} not found in hooks.json" + assert status[script] is True, ( + f"{script} should be async:true — it is fire-and-forget" + ) + + +class TestNewSDKOptimizationEntries: + """Verify new hook entries from the SDK optimization feature.""" + + def test_teammate_idle_hook_exists(self, hooks_config): + """TeammateIdle event should have the teammate_idle.py hook.""" + assert "TeammateIdle" in hooks_config["hooks"] + entries = hooks_config["hooks"]["TeammateIdle"] + commands = [] + for entry in entries: + for hook in entry.get("hooks", []): + commands.append(hook.get("command", "")) + + assert any("teammate_idle.py" in cmd for cmd in commands), ( + "teammate_idle.py not found in TeammateIdle hooks" + ) + + def test_session_end_is_async(self, hooks_config): + """SessionEnd hook should be async (fire-and-forget).""" + entries = hooks_config["hooks"].get("SessionEnd", []) + for entry in entries: + for hook in entry.get("hooks", []): + if "session_end.py" in hook.get("command", ""): + assert hook.get("async") is True, ( + "session_end.py should be async:true" + ) + + def test_file_tracker_is_async(self, hooks_config): + """file_tracker.py PostToolUse hook should be async.""" + entries = hooks_config["hooks"].get("PostToolUse", []) + for entry in entries: + for hook in entry.get("hooks", []): + if "file_tracker.py" in hook.get("command", ""): + assert hook.get("async") is True, ( + "file_tracker.py should be async:true" + ) + + def test_file_size_check_is_async(self, hooks_config): + """file_size_check.py PostToolUse hook should be async.""" + entries = hooks_config["hooks"].get("PostToolUse", []) + for entry in entries: + for hook in entry.get("hooks", []): + if "file_size_check.py" in hook.get("command", ""): + assert hook.get("async") is True, ( + "file_size_check.py should be async:true" + ) + + def test_track_files_is_sync(self, hooks_config): + """track_files.py PostToolUse hook should be synchronous (not async).""" + entries = hooks_config["hooks"].get("PostToolUse", []) + for entry in entries: + for hook in entry.get("hooks", []): + if "track_files.py" in hook.get("command", ""): + assert hook.get("async", False) is not True, ( + "track_files.py should be synchronous" + ) + + +AGENTS_DIR = Path(__file__).parent.parent / "agents" + + +class TestMatcherPatterns: + """Validate matcher patterns use correct pipe-separated syntax.""" + + def _get_all_matchers(self, hooks_config) -> list[tuple[str, str]]: + """Extract all (event_type, matcher) pairs from hooks.json.""" + matchers = [] + for event_type, entries in hooks_config["hooks"].items(): + for entry in entries: + if "matcher" in entry: + matchers.append((event_type, entry["matcher"])) + return matchers + + def test_no_empty_segments_in_matchers(self, hooks_config): + """Pipe-separated matchers must not have empty segments (e.g., '|foo' or 'foo||bar').""" + errors = [] + for event_type, matcher in self._get_all_matchers(hooks_config): + segments = matcher.split("|") + for i, seg in enumerate(segments): + if seg.strip() == "": + errors.append( + f"{event_type}: matcher '{matcher}' has empty segment at position {i}" + ) + assert errors == [], f"Invalid matcher patterns:\n" + "\n".join(errors) + + def test_no_leading_or_trailing_pipes(self, hooks_config): + """Matchers must not start or end with '|'.""" + errors = [] + for event_type, matcher in self._get_all_matchers(hooks_config): + if matcher.startswith("|"): + errors.append(f"{event_type}: matcher starts with '|': '{matcher}'") + if matcher.endswith("|"): + errors.append(f"{event_type}: matcher ends with '|': '{matcher}'") + assert errors == [], f"Invalid matcher patterns:\n" + "\n".join(errors) + + def test_subagent_start_covers_all_agent_types(self, hooks_config): + """SubagentStart matcher must include all PACT agent types from agents/ directory.""" + # Read expected agent names from disk + expected_agents = set() + for agent_file in AGENTS_DIR.glob("*.md"): + # Agent files are named pact-{type}.md — the stem is the agent name + expected_agents.add(agent_file.stem) + + assert len(expected_agents) > 0, "No agent files found in agents/ directory" + + # Extract the SubagentStart matcher + subagent_start_entries = hooks_config["hooks"].get("SubagentStart", []) + matcher_agents = set() + for entry in subagent_start_entries: + if "matcher" in entry: + matcher_agents.update(entry["matcher"].split("|")) + + # Every agent definition should appear in the matcher + missing = expected_agents - matcher_agents + assert missing == set(), ( + f"SubagentStart matcher is missing agent types: {sorted(missing)}. " + f"Matcher has: {sorted(matcher_agents)}. " + f"Expected from agents/: {sorted(expected_agents)}" + ) diff --git a/pact-plugin/tests/test_session_end.py b/pact-plugin/tests/test_session_end.py index d3b3f195..82a0f874 100644 --- a/pact-plugin/tests/test_session_end.py +++ b/pact-plugin/tests/test_session_end.py @@ -215,6 +215,44 @@ def test_includes_unresolved_blockers(self, tmp_path): assert "## Unresolved" in content assert "#10 BLOCKER: missing API key" in content + def test_truncates_long_decision_summary(self, tmp_path): + """Decision strings longer than 80 chars should be truncated to 77 + '...'.""" + from session_end import write_session_snapshot + + long_decision = "A" * 100 # 100 chars, well over 80-char threshold + + tasks = [ + { + "id": "15", + "subject": "CODE: auth", + "status": "completed", + "metadata": { + "handoff": { + "produced": ["src/auth.py"], + "decisions": [long_decision, "Short decision"], + "uncertainty": [], + "integration": [], + "open_questions": [], + } + }, + } + ] + + write_session_snapshot( + tasks=tasks, + project_slug="trunc-proj", + sessions_dir=str(tmp_path), + ) + + content = (tmp_path / "trunc-proj" / "last-session.md").read_text() + # The first decision (used as summary) should be truncated + expected_summary = "A" * 77 + "..." + assert expected_summary in content + # The full 100-char string should NOT appear in the completed task line + assert long_decision not in content.split("## Key Decisions")[0] + # But the full decision DOES appear in Key Decisions section (not truncated there) + assert long_decision in content + class TestMainEntryPoint: """Tests for session_end.main() exit behavior.""" @@ -279,3 +317,164 @@ def test_main_calls_write_snapshot_with_tasks(self): call_args = mock_snapshot.call_args assert call_args.kwargs["tasks"] == mock_tasks assert call_args.kwargs["project_slug"] == "my-project" + + def test_main_calls_cleanup_stale_teams(self): + """main() should call cleanup_stale_teams() after write_session_snapshot().""" + from session_end import main + + env = {"CLAUDE_PROJECT_DIR": "/Users/mj/Sites/my-project"} + + with patch.dict("os.environ", env, clear=True), \ + patch("session_end.get_task_list", return_value=[]), \ + patch("session_end.write_session_snapshot"), \ + patch("session_end.cleanup_stale_teams") as mock_cleanup: + with pytest.raises(SystemExit): + main() + + mock_cleanup.assert_called_once() + + +# ============================================================================= +# cleanup_stale_teams() Tests +# ============================================================================= + +class TestCleanupStaleTeams: + """Tests for session_end.cleanup_stale_teams() — session-scoped cleanup + of the current team's directory and corresponding task directory.""" + + def test_removes_specified_team(self, tmp_path): + """Team directory should be removed when team_name is specified.""" + from session_end import cleanup_stale_teams + + teams_dir = tmp_path / "teams" + team = teams_dir / "pact-abc123" + team.mkdir(parents=True) + (team / "config.json").write_text('{"members": []}') + + cleaned = cleanup_stale_teams(team_name="pact-abc123", teams_dir=str(teams_dir)) + assert "pact-abc123" in cleaned + assert not team.exists() + + def test_removes_team_without_config(self, tmp_path): + """Team directory with no config.json should still be removed.""" + from session_end import cleanup_stale_teams + + teams_dir = tmp_path / "teams" + team = teams_dir / "pact-orphan" + team.mkdir(parents=True) + + cleaned = cleanup_stale_teams(team_name="pact-orphan", teams_dir=str(teams_dir)) + assert "pact-orphan" in cleaned + assert not team.exists() + + def test_does_not_touch_other_teams(self, tmp_path): + """Only the specified team should be removed, not other pact-* teams.""" + from session_end import cleanup_stale_teams + + teams_dir = tmp_path / "teams" + + # Target team + target = teams_dir / "pact-target" + target.mkdir(parents=True) + + # Other team — must survive + other = teams_dir / "pact-other" + other.mkdir(parents=True) + (other / "config.json").write_text('{"members": []}') + + cleaned = cleanup_stale_teams(team_name="pact-target", teams_dir=str(teams_dir)) + assert "pact-target" in cleaned + assert not target.exists() + assert other.exists() # Untouched + + def test_also_removes_corresponding_task_dir(self, tmp_path): + """When removing a team, should also remove ~/.claude/tasks/{team_name}.""" + from session_end import cleanup_stale_teams + + teams_dir = tmp_path / "teams" + tasks_dir = tmp_path / "tasks" + + team = teams_dir / "pact-cleanup" + team.mkdir(parents=True) + + task_dir = tasks_dir / "pact-cleanup" + task_dir.mkdir(parents=True) + (task_dir / "1.json").write_text('{"id": "1"}') + + cleaned = cleanup_stale_teams(team_name="pact-cleanup", teams_dir=str(teams_dir)) + assert "pact-cleanup" in cleaned + assert not team.exists() + assert not task_dir.exists() + + def test_no_error_when_task_dir_missing(self, tmp_path): + """Should not fail if corresponding task dir doesn't exist.""" + from session_end import cleanup_stale_teams + + teams_dir = tmp_path / "teams" + team = teams_dir / "pact-notasks" + team.mkdir(parents=True) + + cleaned = cleanup_stale_teams(team_name="pact-notasks", teams_dir=str(teams_dir)) + assert "pact-notasks" in cleaned + + def test_returns_empty_when_team_dir_missing(self, tmp_path): + """Should return empty list if the team directory doesn't exist.""" + from session_end import cleanup_stale_teams + + teams_dir = tmp_path / "teams" + teams_dir.mkdir(parents=True) + + cleaned = cleanup_stale_teams(team_name="pact-nonexistent", teams_dir=str(teams_dir)) + assert cleaned == [] + + def test_returns_empty_when_teams_dir_missing(self, tmp_path): + """Should return empty list if the base teams directory doesn't exist.""" + from session_end import cleanup_stale_teams + + cleaned = cleanup_stale_teams( + team_name="pact-abc", + teams_dir=str(tmp_path / "nonexistent"), + ) + assert cleaned == [] + + def test_returns_empty_when_no_team_name(self, tmp_path): + """Should return empty list if team_name is empty string.""" + from session_end import cleanup_stale_teams + + teams_dir = tmp_path / "teams" + teams_dir.mkdir(parents=True) + + cleaned = cleanup_stale_teams(team_name="", teams_dir=str(teams_dir)) + assert cleaned == [] + + def test_reads_team_name_from_env(self, tmp_path): + """Should read team name from CLAUDE_CODE_TEAM_NAME env var.""" + from session_end import cleanup_stale_teams + + teams_dir = tmp_path / "teams" + team = teams_dir / "pact-fromenv" + team.mkdir(parents=True) + + env = {"CLAUDE_CODE_TEAM_NAME": "PACT-FROMENV"} + with patch.dict("os.environ", env, clear=True): + cleaned = cleanup_stale_teams(teams_dir=str(teams_dir)) + assert "pact-fromenv" in cleaned + assert not team.exists() + + def test_returns_empty_when_no_env_and_no_arg(self, tmp_path): + """Should return empty list when no team_name arg and no env var.""" + from session_end import cleanup_stale_teams + + with patch.dict("os.environ", {}, clear=True): + cleaned = cleanup_stale_teams(teams_dir=str(tmp_path)) + assert cleaned == [] + + def test_defaults_to_home_claude_teams(self): + """Without teams_dir override, should use ~/.claude/teams/.""" + from session_end import cleanup_stale_teams + + # Just verify it doesn't crash when called with explicit team name + # (it will look for the team under ~/.claude/teams/ which may not exist) + result = cleanup_stale_teams(team_name="pact-nonexistent-test") + assert isinstance(result, list) + assert result == [] diff --git a/pact-plugin/tests/test_teammate_idle.py b/pact-plugin/tests/test_teammate_idle.py new file mode 100644 index 00000000..2c7e4309 --- /dev/null +++ b/pact-plugin/tests/test_teammate_idle.py @@ -0,0 +1,819 @@ +""" +Tests for teammate_idle.py — TeammateIdle hook for stall detection and idle cleanup. + +Tests cover: +1. Stall detection: in_progress task + idle = stall warning +2. Stall detection: completed task + idle = no stall (handled by idle cleanup) +3. Stall detection: no task = no stall +4. Stall detection: already-stalled task = no re-alert +5. Stall detection: blocker/algedonic task = no stall +6. Idle count tracking: increment on consecutive idles +7. Idle count tracking: reset when agent gets new work +8. Shutdown thresholds: no message below 3 +9. Shutdown thresholds: suggest at 3 +10. Shutdown thresholds: force shutdown_request at 5 +11. Main entry point: stdin/stdout/exit behavior +""" +import json +import sys +from pathlib import Path +from unittest.mock import patch + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent / "hooks")) + + +# Sample task fixtures +def make_task(task_id="1", subject="CODE: auth", status="in_progress", + owner="backend-coder", metadata=None): + """Helper to create a task dict.""" + return { + "id": task_id, + "subject": subject, + "status": status, + "owner": owner, + "metadata": metadata or {}, + } + + +class TestFindTeammateTask: + """Tests for teammate_idle.find_teammate_task().""" + + def test_finds_in_progress_task(self): + from teammate_idle import find_teammate_task + + tasks = [make_task(status="in_progress", owner="coder-a")] + result = find_teammate_task(tasks, "coder-a") + assert result is not None + assert result["status"] == "in_progress" + + def test_finds_completed_task(self): + from teammate_idle import find_teammate_task + + tasks = [make_task(status="completed", owner="coder-a")] + result = find_teammate_task(tasks, "coder-a") + assert result is not None + assert result["status"] == "completed" + + def test_prefers_in_progress_over_completed(self): + from teammate_idle import find_teammate_task + + tasks = [ + make_task(task_id="1", status="completed", owner="coder-a"), + make_task(task_id="2", status="in_progress", owner="coder-a"), + ] + result = find_teammate_task(tasks, "coder-a") + assert result["id"] == "2" + assert result["status"] == "in_progress" + + def test_returns_none_for_no_matching_owner(self): + from teammate_idle import find_teammate_task + + tasks = [make_task(owner="coder-b")] + result = find_teammate_task(tasks, "coder-a") + assert result is None + + def test_returns_none_for_empty_tasks(self): + from teammate_idle import find_teammate_task + + result = find_teammate_task([], "coder-a") + assert result is None + + def test_returns_highest_id_completed_task(self): + from teammate_idle import find_teammate_task + + tasks = [ + make_task(task_id="3", status="completed", owner="coder-a"), + make_task(task_id="7", status="completed", owner="coder-a"), + make_task(task_id="5", status="completed", owner="coder-a"), + ] + result = find_teammate_task(tasks, "coder-a") + assert result["id"] == "7" + + def test_returns_highest_id_with_double_digit_ids(self): + """Regression: IDs are numeric strings. "20" > "3" numerically, + but "3" > "20" lexicographically. Must compare as int.""" + from teammate_idle import find_teammate_task + + tasks = [ + make_task(task_id="3", status="completed", owner="coder-a"), + make_task(task_id="20", status="completed", owner="coder-a"), + make_task(task_id="10", status="completed", owner="coder-a"), + ] + result = find_teammate_task(tasks, "coder-a") + assert result["id"] == "20" + + def test_handles_non_numeric_ids_gracefully(self): + """Non-numeric task IDs should not crash — falls back safely.""" + from teammate_idle import find_teammate_task + + tasks = [ + make_task(task_id="abc", status="completed", owner="coder-a"), + make_task(task_id="xyz", status="completed", owner="coder-a"), + ] + # Should not raise — just returns one of the tasks + result = find_teammate_task(tasks, "coder-a") + assert result is not None + assert result["status"] == "completed" + + +class TestDetectStall: + """Tests for teammate_idle.detect_stall().""" + + def test_detects_stall_in_progress_task(self): + from teammate_idle import detect_stall + + tasks = [make_task(status="in_progress", owner="coder-a")] + result = detect_stall(tasks, "coder-a") + assert result is not None + assert "stall" in result.lower() + assert "coder-a" in result + assert "imPACT" in result + + def test_no_stall_for_completed_task(self): + from teammate_idle import detect_stall + + tasks = [make_task(status="completed", owner="coder-a")] + result = detect_stall(tasks, "coder-a") + assert result is None + + def test_no_stall_for_no_task(self): + from teammate_idle import detect_stall + + tasks = [make_task(owner="coder-b")] + result = detect_stall(tasks, "coder-a") + assert result is None + + def test_no_stall_for_blocker_task(self): + from teammate_idle import detect_stall + + tasks = [make_task( + status="in_progress", owner="coder-a", + metadata={"type": "blocker"} + )] + result = detect_stall(tasks, "coder-a") + assert result is None + + def test_no_stall_for_algedonic_task(self): + from teammate_idle import detect_stall + + tasks = [make_task( + status="in_progress", owner="coder-a", + metadata={"type": "algedonic"} + )] + result = detect_stall(tasks, "coder-a") + assert result is None + + def test_no_re_alert_for_already_stalled(self): + from teammate_idle import detect_stall + + tasks = [make_task( + status="in_progress", owner="coder-a", + metadata={"stalled": True} + )] + result = detect_stall(tasks, "coder-a") + assert result is None + + def test_includes_task_id_and_subject(self): + from teammate_idle import detect_stall + + tasks = [make_task( + task_id="42", subject="CODE: fix login", + status="in_progress", owner="coder-a" + )] + result = detect_stall(tasks, "coder-a") + assert "#42" in result + assert "fix login" in result + + +class TestIdleCountTracking: + """Tests for idle count read/write operations.""" + + def test_read_empty_file(self, tmp_path): + from teammate_idle import read_idle_counts + + result = read_idle_counts(str(tmp_path / "idle_counts.json")) + assert result == {} + + def test_read_existing_counts(self, tmp_path): + from teammate_idle import read_idle_counts + + counts_file = tmp_path / "idle_counts.json" + counts_file.write_text('{"coder-a": 3}') + + result = read_idle_counts(str(counts_file)) + assert result == {"coder-a": 3} + + def test_read_corrupted_file(self, tmp_path): + from teammate_idle import read_idle_counts + + counts_file = tmp_path / "idle_counts.json" + counts_file.write_text("not json{{{") + + result = read_idle_counts(str(counts_file)) + assert result == {} + + def test_write_creates_file(self, tmp_path): + from teammate_idle import write_idle_counts + + counts_path = str(tmp_path / "subdir" / "idle_counts.json") + write_idle_counts(counts_path, {"coder-a": 2}) + + result = json.loads(Path(counts_path).read_text()) + assert result == {"coder-a": 2} + + def test_write_overwrites_existing(self, tmp_path): + from teammate_idle import write_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + write_idle_counts(counts_path, {"coder-a": 1}) + write_idle_counts(counts_path, {"coder-a": 3, "coder-b": 1}) + + result = json.loads(Path(counts_path).read_text()) + assert result == {"coder-a": 3, "coder-b": 1} + + def test_reset_idle_count(self, tmp_path): + from teammate_idle import write_idle_counts, reset_idle_count, read_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + write_idle_counts(counts_path, {"coder-a": 3, "coder-b": 1}) + + reset_idle_count("coder-a", counts_path) + + result = read_idle_counts(counts_path) + assert "coder-a" not in result + assert result["coder-b"] == 1 + + def test_reset_nonexistent_teammate(self, tmp_path): + from teammate_idle import write_idle_counts, reset_idle_count, read_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + write_idle_counts(counts_path, {"coder-a": 3}) + + # Should not raise + reset_idle_count("coder-x", counts_path) + + result = read_idle_counts(counts_path) + assert result == {"coder-a": 3} + + +class TestCheckIdleCleanup: + """Tests for teammate_idle.check_idle_cleanup().""" + + def test_no_action_below_threshold(self, tmp_path): + from teammate_idle import check_idle_cleanup + + counts_path = str(tmp_path / "idle_counts.json") + tasks = [make_task(status="completed", owner="coder-a")] + + # First idle event (count = 1) + msg, should_shutdown = check_idle_cleanup(tasks, "coder-a", counts_path) + assert msg is None + assert should_shutdown is False + + def test_no_action_at_two(self, tmp_path): + from teammate_idle import check_idle_cleanup, write_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + write_idle_counts(counts_path, {"coder-a": 1}) # Already had 1 + tasks = [make_task(status="completed", owner="coder-a")] + + msg, should_shutdown = check_idle_cleanup(tasks, "coder-a", counts_path) + assert msg is None + assert should_shutdown is False + + def test_suggest_at_three(self, tmp_path): + from teammate_idle import check_idle_cleanup, write_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + write_idle_counts(counts_path, {"coder-a": 2}) # Will become 3 + tasks = [make_task(status="completed", owner="coder-a")] + + msg, should_shutdown = check_idle_cleanup(tasks, "coder-a", counts_path) + assert msg is not None + assert "idle" in msg.lower() + assert "coder-a" in msg + assert should_shutdown is False + + def test_suggest_at_four(self, tmp_path): + from teammate_idle import check_idle_cleanup, write_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + write_idle_counts(counts_path, {"coder-a": 3}) # Will become 4 + tasks = [make_task(status="completed", owner="coder-a")] + + msg, should_shutdown = check_idle_cleanup(tasks, "coder-a", counts_path) + assert msg is not None + assert should_shutdown is False + + def test_force_shutdown_at_five(self, tmp_path): + from teammate_idle import check_idle_cleanup, write_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + write_idle_counts(counts_path, {"coder-a": 4}) # Will become 5 + tasks = [make_task(status="completed", owner="coder-a")] + + msg, should_shutdown = check_idle_cleanup(tasks, "coder-a", counts_path) + assert msg is not None + assert "shutdown" in msg.lower() + assert should_shutdown is True + + def test_force_shutdown_above_five(self, tmp_path): + from teammate_idle import check_idle_cleanup, write_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + write_idle_counts(counts_path, {"coder-a": 9}) # Will become 10 + tasks = [make_task(status="completed", owner="coder-a")] + + msg, should_shutdown = check_idle_cleanup(tasks, "coder-a", counts_path) + assert msg is not None + assert should_shutdown is True + + def test_resets_count_when_no_completed_task(self, tmp_path): + from teammate_idle import check_idle_cleanup, write_idle_counts, read_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + write_idle_counts(counts_path, {"coder-a": 3}) + # Agent now has in_progress task (got new work) + tasks = [make_task(status="in_progress", owner="coder-a")] + + msg, should_shutdown = check_idle_cleanup(tasks, "coder-a", counts_path) + assert msg is None + assert should_shutdown is False + + # Count should be reset + counts = read_idle_counts(counts_path) + assert "coder-a" not in counts + + def test_skips_stalled_agents(self, tmp_path): + from teammate_idle import check_idle_cleanup, write_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + write_idle_counts(counts_path, {"coder-a": 4}) + tasks = [make_task( + status="completed", owner="coder-a", + metadata={"stalled": True} + )] + + msg, should_shutdown = check_idle_cleanup(tasks, "coder-a", counts_path) + assert msg is None + assert should_shutdown is False + + def test_skips_terminated_agents(self, tmp_path): + from teammate_idle import check_idle_cleanup, write_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + write_idle_counts(counts_path, {"coder-a": 4}) + tasks = [make_task( + status="completed", owner="coder-a", + metadata={"terminated": True} + )] + + msg, should_shutdown = check_idle_cleanup(tasks, "coder-a", counts_path) + assert msg is None + assert should_shutdown is False + + def test_no_task_resets_count(self, tmp_path): + from teammate_idle import check_idle_cleanup, write_idle_counts, read_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + write_idle_counts(counts_path, {"coder-a": 3}) + tasks = [make_task(owner="coder-b")] # No task for coder-a + + msg, should_shutdown = check_idle_cleanup(tasks, "coder-a", counts_path) + assert msg is None + + counts = read_idle_counts(counts_path) + assert "coder-a" not in counts + + +class TestMain: + """Tests for teammate_idle.main() stdin/stdout/exit behavior.""" + + def _run_main(self, input_data, env=None, tasks=None): + """Helper to run main() with mocked inputs.""" + import io + from teammate_idle import main + + default_env = { + "CLAUDE_CODE_TEAM_NAME": "pact-test", + } + if env: + default_env.update(env) + + mock_tasks = tasks if tasks is not None else [] + + with patch.dict(os.environ, default_env, clear=True), \ + patch("sys.stdin", io.StringIO(json.dumps(input_data))), \ + patch("teammate_idle.get_task_list", return_value=mock_tasks): + with pytest.raises(SystemExit) as exc_info: + main() + + return exc_info.value.code + + def test_exits_0_when_no_team(self, capsys): + import io + import os + from teammate_idle import main + + with patch.dict(os.environ, {"CLAUDE_CODE_TEAM_NAME": ""}, clear=True), \ + patch("sys.stdin", io.StringIO("{}")): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + + def test_exits_0_when_no_teammate_name(self): + exit_code = self._run_main({"teammate_name": ""}) + assert exit_code == 0 + + def test_exits_0_when_no_tasks(self): + exit_code = self._run_main( + {"teammate_name": "coder-a"}, + tasks=None + ) + assert exit_code == 0 + + def test_outputs_stall_warning(self, capsys, tmp_path): + import io + import os + from teammate_idle import main + + tasks = [make_task(status="in_progress", owner="coder-a")] + + env = { + "CLAUDE_CODE_TEAM_NAME": "pact-test", + } + + with patch.dict(os.environ, env, clear=True), \ + patch("sys.stdin", io.StringIO(json.dumps({"teammate_name": "coder-a"}))), \ + patch("teammate_idle.get_task_list", return_value=tasks): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + captured = capsys.readouterr() + if captured.out.strip(): + output = json.loads(captured.out) + assert "systemMessage" in output + assert "stall" in output["systemMessage"].lower() + + def test_outputs_nothing_for_completed_below_threshold(self, capsys, tmp_path): + import io + import os + from teammate_idle import main + + tasks = [make_task(status="completed", owner="coder-a")] + + env = { + "CLAUDE_CODE_TEAM_NAME": "pact-test", + } + + # Patch Path.home() so idle_counts.json uses tmp_path instead of real home + # (prevents cross-test pollution from persisted idle count files) + with patch.dict(os.environ, env, clear=True), \ + patch("sys.stdin", io.StringIO(json.dumps({"teammate_name": "coder-a"}))), \ + patch("teammate_idle.get_task_list", return_value=tasks), \ + patch("teammate_idle.Path.home", return_value=tmp_path): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + captured = capsys.readouterr() + # No output expected below threshold (first idle event, count=1) + assert captured.out.strip() == "" or "systemMessage" not in captured.out + + def test_exits_0_on_invalid_json(self): + import io + import os + from teammate_idle import main + + with patch.dict(os.environ, {"CLAUDE_CODE_TEAM_NAME": "pact-test"}, clear=True), \ + patch("sys.stdin", io.StringIO("not json")): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + + +# Required for patch.dict +import os + + +# ============================================================================= +# Edge Case Tests — Stall/Idle Interaction, Concurrent Tracking +# ============================================================================= + +class TestLegacyIdleCountMigration: + """Tests for the int-to-structured-dict migration in check_idle_cleanup(). + + Legacy idle_counts.json files stored plain ints per teammate (e.g., {"coder-a": 3}). + The current format uses structured dicts (e.g., {"coder-a": {"count": 3, "task_id": "1"}}). + The migration logic in _increment() must handle both formats transparently.""" + + def test_legacy_int_migrated_to_structured_dict(self, tmp_path): + """When idle_counts.json has a plain int, check_idle_cleanup should + migrate it to structured format and continue counting correctly.""" + from teammate_idle import check_idle_cleanup, read_idle_counts, write_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + # Write legacy format: plain int + write_idle_counts(counts_path, {"coder-a": 2}) + tasks = [make_task(task_id="5", status="completed", owner="coder-a")] + + # Should migrate the int (2) to {"count": 2, "task_id": ""} then + # increment to 3, which hits the suggest threshold + msg, should_shutdown = check_idle_cleanup(tasks, "coder-a", counts_path) + assert msg is not None + assert "idle" in msg.lower() + assert should_shutdown is False + + # Verify the file now contains structured format + counts = read_idle_counts(counts_path) + entry = counts["coder-a"] + assert isinstance(entry, dict) + assert entry["count"] == 3 + assert entry["task_id"] == "5" + + def test_legacy_int_zero_migrated_correctly(self, tmp_path): + """Edge case: legacy count of 0 should migrate and increment to 1.""" + from teammate_idle import check_idle_cleanup, read_idle_counts, write_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + write_idle_counts(counts_path, {"coder-a": 0}) + tasks = [make_task(task_id="1", status="completed", owner="coder-a")] + + msg, should_shutdown = check_idle_cleanup(tasks, "coder-a", counts_path) + # Count goes from 0 -> 1, below threshold + assert msg is None + assert should_shutdown is False + + counts = read_idle_counts(counts_path) + entry = counts["coder-a"] + assert isinstance(entry, dict) + assert entry["count"] == 1 + + +class TestStallIdleInteraction: + """Verify that stalled agents get stall detection, NOT idle cleanup. + This is the key invariant: a stalled agent (in_progress task + idle) + should receive a stall warning, and should NOT be tracked for idle cleanup.""" + + def test_stall_prevents_idle_count_increment(self, tmp_path): + """When stall is detected, idle count should NOT be incremented.""" + from teammate_idle import detect_stall, check_idle_cleanup, read_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + tasks = [make_task(status="in_progress", owner="coder-a")] + + # Stall IS detected + stall_msg = detect_stall(tasks, "coder-a") + assert stall_msg is not None + + # Idle cleanup returns nothing (task not completed) + msg, shutdown = check_idle_cleanup(tasks, "coder-a", counts_path) + assert msg is None + assert shutdown is False + + # Idle count should NOT have been set + counts = read_idle_counts(counts_path) + assert "coder-a" not in counts + + def test_main_exclusive_stall_or_cleanup(self, capsys, tmp_path): + """main() should emit stall OR cleanup message, never both. + The code has 'else' branch: if stall, skip cleanup check.""" + import io + from teammate_idle import main + + # Agent with in_progress task (stall case) + tasks = [make_task(status="in_progress", owner="coder-a")] + + with patch.dict(os.environ, {"CLAUDE_CODE_TEAM_NAME": "pact-test"}, clear=True), \ + patch("sys.stdin", io.StringIO(json.dumps({"teammate_name": "coder-a"}))), \ + patch("teammate_idle.get_task_list", return_value=tasks): + with pytest.raises(SystemExit): + main() + + captured = capsys.readouterr() + if captured.out.strip(): + output = json.loads(captured.out) + msg = output.get("systemMessage", "") + # Should have stall message, NOT cleanup/shutdown message + assert "stall" in msg.lower() + assert "shutdown" not in msg.lower() + + def test_completed_then_new_work_resets_idle(self, tmp_path): + """Agent completes task (starts idle tracking), then gets new work + (in_progress). Idle count should be reset.""" + from teammate_idle import check_idle_cleanup, write_idle_counts, read_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + + # First: agent has completed task, idle count accumulates + completed_tasks = [make_task(status="completed", owner="coder-a")] + write_idle_counts(counts_path, {"coder-a": 4}) + + # Now: agent gets a new in_progress task + new_tasks = [ + make_task(task_id="2", status="in_progress", owner="coder-a"), + make_task(task_id="1", status="completed", owner="coder-a"), + ] + + # check_idle_cleanup should reset because find_teammate_task returns + # the in_progress task (not completed) + msg, shutdown = check_idle_cleanup(new_tasks, "coder-a", counts_path) + assert msg is None + assert shutdown is False + + counts = read_idle_counts(counts_path) + assert "coder-a" not in counts + + +class TestConcurrentIdleTracking: + """Test idle tracking with multiple agents being tracked simultaneously.""" + + def test_multiple_agents_tracked_independently(self, tmp_path): + """Each agent's idle count should be independent.""" + from teammate_idle import check_idle_cleanup, write_idle_counts, read_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + + tasks = [ + make_task(task_id="1", status="completed", owner="coder-a"), + make_task(task_id="2", status="completed", owner="coder-b"), + ] + + # coder-a idles 3 times (pre-seed with structured format) + write_idle_counts(counts_path, {"coder-a": {"count": 2, "task_id": "1"}}) + msg_a, _ = check_idle_cleanup(tasks, "coder-a", counts_path) + assert msg_a is not None # Suggest at 3 + assert "coder-a" in msg_a + + # coder-b idles once (count starts at 0) + msg_b, _ = check_idle_cleanup(tasks, "coder-b", counts_path) + assert msg_b is None # Below threshold + + counts = read_idle_counts(counts_path) + assert counts["coder-a"]["count"] == 3 + assert counts["coder-b"]["count"] == 1 + + def test_one_agent_shutdown_doesnt_affect_others(self, tmp_path): + """Force-shutdown of one agent should not affect others' counts.""" + from teammate_idle import check_idle_cleanup, write_idle_counts, read_idle_counts + + counts_path = str(tmp_path / "idle_counts.json") + + tasks = [ + make_task(task_id="1", status="completed", owner="coder-a"), + make_task(task_id="2", status="completed", owner="coder-b"), + ] + + write_idle_counts(counts_path, { + "coder-a": {"count": 4, "task_id": "1"}, + "coder-b": {"count": 1, "task_id": "2"}, + }) + + # coder-a hits force threshold (5) + msg_a, shutdown_a = check_idle_cleanup(tasks, "coder-a", counts_path) + assert shutdown_a is True + + # coder-b still at count 2, no action + msg_b, shutdown_b = check_idle_cleanup(tasks, "coder-b", counts_path) + assert shutdown_b is False + assert msg_b is None + + counts = read_idle_counts(counts_path) + assert counts["coder-a"]["count"] == 5 + assert counts["coder-b"]["count"] == 2 + + +class TestFindTeammateTaskEdgeCases: + """Additional edge cases for find_teammate_task().""" + + def test_pending_task_not_returned(self): + """Pending tasks (not yet started) should NOT be returned.""" + from teammate_idle import find_teammate_task + + tasks = [make_task(status="pending", owner="coder-a")] + result = find_teammate_task(tasks, "coder-a") + assert result is None + + def test_deleted_task_not_returned(self): + """Deleted tasks should NOT be returned.""" + from teammate_idle import find_teammate_task + + tasks = [make_task(status="deleted", owner="coder-a")] + result = find_teammate_task(tasks, "coder-a") + assert result is None + + def test_mixed_statuses_returns_in_progress(self): + """With pending + in_progress + completed, returns in_progress.""" + from teammate_idle import find_teammate_task + + tasks = [ + make_task(task_id="1", status="pending", owner="coder-a"), + make_task(task_id="2", status="in_progress", owner="coder-a"), + make_task(task_id="3", status="completed", owner="coder-a"), + ] + result = find_teammate_task(tasks, "coder-a") + assert result["id"] == "2" + + def test_owner_matching_is_exact(self): + """Owner matching should be exact, not substring.""" + from teammate_idle import find_teammate_task + + tasks = [make_task(status="in_progress", owner="coder-a-backend")] + result = find_teammate_task(tasks, "coder-a") + assert result is None + + +class TestDetectStallEdgeCases: + """Additional edge cases for detect_stall().""" + + def test_empty_metadata_still_detects_stall(self): + """Task with empty metadata dict should still trigger stall.""" + from teammate_idle import detect_stall + + tasks = [make_task( + status="in_progress", owner="coder-a", + metadata={} + )] + result = detect_stall(tasks, "coder-a") + assert result is not None + + def test_no_metadata_key_still_detects_stall(self): + """Task without metadata key at all should still trigger stall.""" + from teammate_idle import detect_stall + + task = {"id": "1", "subject": "work", "status": "in_progress", "owner": "coder-a"} + result = detect_stall([task], "coder-a") + assert result is not None + + def test_stalled_false_value_still_detects_stall(self): + """metadata.stalled=False should NOT suppress stall detection.""" + from teammate_idle import detect_stall + + tasks = [make_task( + status="in_progress", owner="coder-a", + metadata={"stalled": False} + )] + result = detect_stall(tasks, "coder-a") + assert result is not None # False != truthy, so stall should fire + + +class TestMainEdgeCases: + """Additional edge cases for main() entry point.""" + + def test_team_name_lowercased(self): + """CLAUDE_CODE_TEAM_NAME should be lowercased per v3.3.2 convention.""" + import io + from teammate_idle import main + + # Use uppercase team name — should work (lowercased internally) + with patch.dict(os.environ, {"CLAUDE_CODE_TEAM_NAME": "PACT-TEST"}, clear=True), \ + patch("sys.stdin", io.StringIO(json.dumps({"teammate_name": ""}))): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 # Empty teammate_name exits cleanly + + def test_get_task_list_returns_none(self): + """Should exit cleanly when get_task_list() returns None.""" + import io + from teammate_idle import main + + with patch.dict(os.environ, {"CLAUDE_CODE_TEAM_NAME": "pact-test"}, clear=True), \ + patch("sys.stdin", io.StringIO(json.dumps({"teammate_name": "coder-a"}))), \ + patch("teammate_idle.get_task_list", return_value=None): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + + def test_shutdown_message_includes_action_required(self, capsys, tmp_path): + """When force shutdown threshold hit, output should include ACTION REQUIRED.""" + import io + from teammate_idle import main, write_idle_counts + + tasks = [make_task(status="completed", owner="coder-a")] + + # Pre-set count to 4 (will become 5 = force threshold) + idle_dir = tmp_path / "teams" / "pact-test" + idle_dir.mkdir(parents=True) + write_idle_counts(str(idle_dir / "idle_counts.json"), {"coder-a": 4}) + + with patch.dict(os.environ, {"CLAUDE_CODE_TEAM_NAME": "pact-test"}, clear=True), \ + patch("sys.stdin", io.StringIO(json.dumps({"teammate_name": "coder-a"}))), \ + patch("teammate_idle.get_task_list", return_value=tasks), \ + patch("pathlib.Path.home", return_value=tmp_path): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + captured = capsys.readouterr() + if captured.out.strip(): + output = json.loads(captured.out) + msg = output.get("systemMessage", "") + assert "ACTION REQUIRED" in msg + assert "shutdown_request" in msg diff --git a/pact-plugin/tests/test_validate_handoff.py b/pact-plugin/tests/test_validate_handoff.py new file mode 100644 index 00000000..3e58a950 --- /dev/null +++ b/pact-plugin/tests/test_validate_handoff.py @@ -0,0 +1,437 @@ +# pact-plugin/tests/test_validate_handoff.py +""" +Tests for validate_handoff.py — SubagentStop hook that validates PACT +agent/teammate handoff format. + +Tests cover: +1. validate_handoff() with structured handoff section +2. validate_handoff() with implicit handoff elements +3. validate_handoff() with missing elements +4. is_pact_agent() identification +5. main() prefers last_assistant_message over transcript (SDK v2.1.47+) +6. main() falls back to transcript when last_assistant_message absent +7. main() entry point: stdin JSON, exit codes, output format +""" +import io +import json +import sys +from pathlib import Path +from unittest.mock import patch + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent / "hooks")) + + +# ============================================================================= +# Test Data +# ============================================================================= + +GOOD_HANDOFF = """ +## HANDOFF + +1. Produced: Created src/auth.py with JWT authentication middleware +2. Key decisions: Chose JWT over session tokens for stateless design +3. Areas of uncertainty: + - [HIGH] Token refresh logic untested with concurrent requests +4. Integration points: Connects to user_service.py via get_user() +5. Open questions: Should token expiry be configurable? +""" + +PARTIAL_HANDOFF = "Implemented the auth module. Used JWT tokens for the approach." + +MISSING_HANDOFF = "Hello world, here is some random text without any handoff info." + + +# ============================================================================= +# validate_handoff() Tests +# ============================================================================= + +class TestValidateHandoff: + """Tests for validate_handoff.validate_handoff().""" + + def test_explicit_handoff_section_is_valid(self): + from validate_handoff import validate_handoff + + is_valid, missing = validate_handoff(GOOD_HANDOFF) + assert is_valid is True + assert missing == [] + + def test_implicit_elements_are_detected(self): + from validate_handoff import validate_handoff + + # Contains "produced" (what_produced) and "chose" (key_decisions) + # and "next" (next_steps) — all 3 present + text = ( + "I produced the auth module. " + "I chose JWT tokens because they are stateless. " + "Next, the test engineer should verify token expiry." + ) + is_valid, missing = validate_handoff(text) + assert is_valid is True + assert missing == [] + + def test_partial_handoff_with_two_of_three(self): + from validate_handoff import validate_handoff + + # Has "implemented" (what_produced) and "approach" (key_decisions) + # Missing next_steps — but 2/3 is still valid + is_valid, missing = validate_handoff(PARTIAL_HANDOFF) + assert is_valid is True + assert len(missing) <= 1 + + def test_missing_handoff_is_invalid(self): + from validate_handoff import validate_handoff + + is_valid, missing = validate_handoff(MISSING_HANDOFF) + assert is_valid is False + assert len(missing) >= 2 + + def test_case_insensitive_section_detection(self): + from validate_handoff import validate_handoff + + text = "## handoff\nProduced: files. Decisions: none. Next: test." + is_valid, missing = validate_handoff(text) + assert is_valid is True + + +# ============================================================================= +# is_pact_agent() Tests +# ============================================================================= + +class TestIsPactAgent: + """Tests for validate_handoff.is_pact_agent().""" + + def test_recognizes_pact_prefixed_agents(self): + from validate_handoff import is_pact_agent + + assert is_pact_agent("pact-backend-coder") is True + assert is_pact_agent("PACT-architect") is True + assert is_pact_agent("pact_test_engineer") is True + assert is_pact_agent("PACT_preparer") is True + + def test_rejects_non_pact_agents(self): + from validate_handoff import is_pact_agent + + assert is_pact_agent("custom-agent") is False + assert is_pact_agent("") is False + assert is_pact_agent("my-pact-thing") is False + + def test_rejects_none(self): + from validate_handoff import is_pact_agent + + assert is_pact_agent(None) is False + + +# ============================================================================= +# main() Tests — last_assistant_message preference +# ============================================================================= + +class TestMainLastAssistantMessage: + """Tests for main() preferring last_assistant_message over transcript.""" + + def test_uses_last_assistant_message_when_present(self, capsys): + """When last_assistant_message is provided, it should be used + instead of transcript.""" + from validate_handoff import main + + input_data = json.dumps({ + "agent_id": "pact-backend-coder", + "last_assistant_message": GOOD_HANDOFF, + "transcript": MISSING_HANDOFF, # Would fail if used + }) + + with patch("sys.stdin", io.StringIO(input_data)): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + captured = capsys.readouterr() + # Good handoff => no warnings printed + assert captured.out == "" + + def test_falls_back_to_transcript_when_no_last_assistant_message(self, capsys): + """When last_assistant_message is absent, should fall back to + transcript field.""" + from validate_handoff import main + + input_data = json.dumps({ + "agent_id": "pact-backend-coder", + "transcript": GOOD_HANDOFF, + }) + + with patch("sys.stdin", io.StringIO(input_data)): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + captured = capsys.readouterr() + assert captured.out == "" + + def test_falls_back_to_transcript_when_last_assistant_message_empty(self, capsys): + """When last_assistant_message is empty string, should fall back to + transcript field.""" + from validate_handoff import main + + input_data = json.dumps({ + "agent_id": "pact-backend-coder", + "last_assistant_message": "", + "transcript": GOOD_HANDOFF, + }) + + with patch("sys.stdin", io.StringIO(input_data)): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + captured = capsys.readouterr() + assert captured.out == "" + + def test_warns_on_missing_handoff_from_last_assistant_message(self, capsys): + """When last_assistant_message has poor handoff, should emit warning.""" + from validate_handoff import main + + input_data = json.dumps({ + "agent_id": "pact-backend-coder", + "last_assistant_message": "x" * 100 + " " + MISSING_HANDOFF, + }) + + with patch("sys.stdin", io.StringIO(input_data)): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + captured = capsys.readouterr() + if captured.out: + output = json.loads(captured.out) + assert "systemMessage" in output + assert "Handoff Warning" in output["systemMessage"] + + +# ============================================================================= +# main() Entry Point Tests +# ============================================================================= + +class TestMainEntryPoint: + """Tests for main() stdin/stdout/exit behavior.""" + + def test_exits_0_for_non_pact_agent(self, capsys): + from validate_handoff import main + + input_data = json.dumps({ + "agent_id": "custom-agent", + "transcript": MISSING_HANDOFF, + }) + + with patch("sys.stdin", io.StringIO(input_data)): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + captured = capsys.readouterr() + assert captured.out == "" + + def test_exits_0_on_invalid_json(self): + from validate_handoff import main + + with patch("sys.stdin", io.StringIO("not json")): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + + def test_exits_0_for_short_transcript(self, capsys): + from validate_handoff import main + + input_data = json.dumps({ + "agent_id": "pact-backend-coder", + "last_assistant_message": "short", + }) + + with patch("sys.stdin", io.StringIO(input_data)): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + captured = capsys.readouterr() + assert captured.out == "" + + def test_exits_0_with_no_agent_id(self, capsys): + from validate_handoff import main + + input_data = json.dumps({ + "transcript": GOOD_HANDOFF, + }) + + with patch("sys.stdin", io.StringIO(input_data)): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + captured = capsys.readouterr() + assert captured.out == "" + + +# ============================================================================= +# Edge Case Tests — Field Preference, Boundary Conditions +# ============================================================================= + +class TestLastAssistantMessagePreference: + """Detailed tests for the last_assistant_message vs transcript preference logic.""" + + def test_prefers_last_assistant_message_over_transcript_content(self, capsys): + """When both fields have content, last_assistant_message wins. + Verified by: last_assistant_message has good handoff, transcript has bad. + If transcript were used, we'd get a warning — no warning = correct field used.""" + from validate_handoff import main + + input_data = json.dumps({ + "agent_id": "pact-backend-coder", + "last_assistant_message": GOOD_HANDOFF, + "transcript": "x" * 200, # Long enough to trigger validation, but no handoff + }) + + with patch("sys.stdin", io.StringIO(input_data)): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + captured = capsys.readouterr() + assert captured.out == "" # No warning = used good handoff from last_assistant_message + + def test_last_assistant_message_none_falls_back(self, capsys): + """When last_assistant_message is explicitly None, falls back to transcript.""" + from validate_handoff import main + + input_data = json.dumps({ + "agent_id": "pact-backend-coder", + "last_assistant_message": None, + "transcript": GOOD_HANDOFF, + }) + + with patch("sys.stdin", io.StringIO(input_data)): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + captured = capsys.readouterr() + assert captured.out == "" # Fallback to transcript succeeded + + def test_both_fields_missing_exits_cleanly(self, capsys): + """When both fields are missing, transcript is empty string, exits 0.""" + from validate_handoff import main + + input_data = json.dumps({ + "agent_id": "pact-backend-coder", + }) + + with patch("sys.stdin", io.StringIO(input_data)): + with pytest.raises(SystemExit) as exc_info: + main() + + assert exc_info.value.code == 0 + captured = capsys.readouterr() + assert captured.out == "" # Short transcript (< 100 chars) skips validation + + +class TestValidateHandoffEdgeCases: + """Edge cases for validate_handoff() function.""" + + def test_handoff_section_with_hash_header(self): + """Section header with # or ## should be detected.""" + from validate_handoff import validate_handoff + + text = "# Handoff\nHere is what I did." + is_valid, missing = validate_handoff(text) + assert is_valid is True + + def test_handoff_section_with_colon(self): + """'Handoff:' followed by newline should be detected.""" + from validate_handoff import validate_handoff + + text = "Handoff:\nProduced files." + is_valid, missing = validate_handoff(text) + assert is_valid is True + + def test_deliverables_section_detected(self): + """'## Deliverables' section header should count as structured handoff.""" + from validate_handoff import validate_handoff + + text = "## Deliverables\nCreated auth module." + is_valid, missing = validate_handoff(text) + assert is_valid is True + + def test_summary_section_detected(self): + """'## Summary' section header should count as structured handoff.""" + from validate_handoff import validate_handoff + + text = "## Summary\nDid the work." + is_valid, missing = validate_handoff(text) + assert is_valid is True + + def test_output_section_detected(self): + """'## Output' section header should count as structured handoff.""" + from validate_handoff import validate_handoff + + text = "## Output\nFiles produced." + is_valid, missing = validate_handoff(text) + assert is_valid is True + + def test_empty_string_is_invalid(self): + """Empty string has no handoff elements.""" + from validate_handoff import validate_handoff + + is_valid, missing = validate_handoff("") + assert is_valid is False + assert len(missing) == 3 # All 3 elements missing + + def test_exactly_at_boundary_one_missing(self): + """With exactly 1 out of 3 missing, should still be valid.""" + from validate_handoff import validate_handoff + + # Has "produced" (what_produced) and "decided to" (key_decisions) + # Missing next_steps entirely + text = "I produced the auth module. I decided to use JWT tokens." + is_valid, missing = validate_handoff(text) + assert is_valid is True + assert len(missing) <= 1 + + def test_exactly_at_boundary_two_missing(self): + """With exactly 2 out of 3 missing, should be invalid.""" + from validate_handoff import validate_handoff + + # Only has "produced" (what_produced) + # Missing key_decisions and next_steps + text = "I produced the auth module and it works great and is ready." + is_valid, missing = validate_handoff(text) + assert is_valid is False + assert len(missing) >= 2 + + +class TestIsPactAgentEdgeCases: + """Edge cases for is_pact_agent().""" + + def test_pact_in_middle_not_matched(self): + """'my-pact-agent' should NOT match (prefix check only).""" + from validate_handoff import is_pact_agent + + assert is_pact_agent("my-pact-agent") is False + + def test_just_pact_prefix_matched(self): + """Just 'pact-' should match.""" + from validate_handoff import is_pact_agent + + assert is_pact_agent("pact-") is True + + def test_empty_string_not_matched(self): + from validate_handoff import is_pact_agent + + assert is_pact_agent("") is False + + def test_integer_input_raises(self): + """Non-string input raises AttributeError (startswith not on int). + This is acceptable — main() wraps in try/except.""" + from validate_handoff import is_pact_agent + + with pytest.raises(AttributeError): + is_pact_agent(123) diff --git a/pact-plugin/tests/test_worktree_guard.py b/pact-plugin/tests/test_worktree_guard.py index 7d07abd6..12c87369 100644 --- a/pact-plugin/tests/test_worktree_guard.py +++ b/pact-plugin/tests/test_worktree_guard.py @@ -5,16 +5,19 @@ Tests cover: 1. Edit inside worktree → allow -2. Edit outside worktree to app code → block +2. Edit outside worktree to app code → block with corrected path suggestion 3. Edit outside worktree to .claude/ → allow (AI tooling) 4. Edit outside worktree to docs/ → allow (documentation) 5. No PACT_WORKTREE_PATH set → allow (inactive, no-op) 6. CLAUDE.md always allowed -7. main() entry point: stdin JSON parsing, exit codes, output format +7. Corrected path suggestion (_suggest_worktree_path) +8. main() entry point: stdin JSON parsing, exit codes, output format """ import io import json +import os import sys +import tempfile from pathlib import Path from unittest.mock import patch @@ -43,7 +46,27 @@ def test_blocks_app_code_outside_worktree(self): worktree_path="/tmp/worktrees/feat-auth" ) assert result is not None - assert "outside worktree" in result.lower() + assert "outside the active worktree" in result.lower() + + def test_block_message_includes_corrected_path(self, tmp_path): + """Error message should include 'Did you mean:' with the corrected path.""" + from worktree_guard import check_worktree_boundary + + # Create a realistic directory structure so Path.resolve() works + project_root = tmp_path / "project" + worktree_dir = project_root / ".worktrees" / "feat-auth" + src_dir = project_root / "src" + worktree_dir.mkdir(parents=True) + src_dir.mkdir(parents=True) + (src_dir / "auth.ts").touch() + + result = check_worktree_boundary( + file_path=str(src_dir / "auth.ts"), + worktree_path=str(worktree_dir) + ) + assert result is not None + assert "Did you mean:" in result + assert str(worktree_dir / "src" / "auth.ts") in result def test_allows_claude_dir_outside_worktree(self): from worktree_guard import check_worktree_boundary @@ -82,6 +105,78 @@ def test_allows_claude_md_anywhere(self): assert result is None +class TestSuggestWorktreePath: + """Tests for worktree_guard._suggest_worktree_path().""" + + def test_suggests_path_with_worktrees_dir(self, tmp_path): + from worktree_guard import _suggest_worktree_path + + project_root = tmp_path / "project" + worktree_dir = project_root / ".worktrees" / "feat-auth" + src_dir = project_root / "src" + worktree_dir.mkdir(parents=True) + src_dir.mkdir(parents=True) + (src_dir / "auth.ts").touch() + + result = _suggest_worktree_path( + str(src_dir / "auth.ts"), + str(worktree_dir) + ) + assert result is not None + assert result == str(worktree_dir / "src" / "auth.ts") + + def test_suggests_path_with_common_ancestor(self, tmp_path): + """Fallback: uses common path prefix when .worktrees dir is absent.""" + from worktree_guard import _suggest_worktree_path + + # Two unrelated directories under same tmp_path + workspace = tmp_path / "workspace" + dir_a = workspace / "main" + dir_b = workspace / "branch" + dir_a_src = dir_a / "src" + dir_a_src.mkdir(parents=True) + dir_b.mkdir(parents=True) + (dir_a_src / "app.py").touch() + # Common ancestor must have a project marker for validation + (workspace / ".git").mkdir() + + result = _suggest_worktree_path( + str(dir_a_src / "app.py"), + str(dir_b) + ) + # Should produce dir_b/main/src/app.py or similar via common ancestor + assert result is not None + assert str(dir_b) in result + + def test_rejects_common_ancestor_without_project_marker(self, tmp_path): + """Common-prefix fallback returns None when ancestor lacks project marker.""" + from worktree_guard import _suggest_worktree_path + + # Two directories sharing a common prefix but no .git/.worktrees/CLAUDE.md + workspace = tmp_path / "workspace" + dir_a = workspace / "main" / "src" + dir_b = workspace / "branch" + dir_a.mkdir(parents=True) + dir_b.mkdir(parents=True) + (dir_a / "app.py").touch() + # No project marker at workspace — should reject + + result = _suggest_worktree_path( + str(dir_a / "app.py"), + str(dir_b) + ) + assert result is None + + def test_returns_none_on_path_error(self): + from worktree_guard import _suggest_worktree_path + + # Completely unresolvable paths (won't crash) + result = _suggest_worktree_path("", "") + # Empty paths produce empty Path parts, which may or may not suggest + # Either None or a value is acceptable; must not raise + assert result is None or isinstance(result, str) + + class TestMainEntryPoint: """Tests for worktree_guard.main() stdin/stdout/exit behavior.""" @@ -116,7 +211,7 @@ def test_main_exits_2_when_edit_outside_worktree(self, capsys): "tool_input": {"file_path": "/Users/mj/project/src/auth.ts"} }) - error_msg = "File is outside worktree boundary" + error_msg = "Edit blocked: /Users/mj/project/src/auth.ts is outside the active worktree" with patch.dict("os.environ", {"PACT_WORKTREE_PATH": "/tmp/worktrees/feat-auth"}), \ patch("worktree_guard.check_worktree_boundary", return_value=error_msg), \ patch("sys.stdin", io.StringIO(input_data)): @@ -149,3 +244,175 @@ def test_main_exits_0_when_no_file_path(self): main() assert exc_info.value.code == 0 + + +# ============================================================================= +# Edge Case Tests — Path Handling, Application Code Detection +# ============================================================================= + +class TestIsAllowedPathEdgeCases: + """Edge cases for worktree_guard.is_allowed_path().""" + + def test_claude_dir_nested_deep(self): + """Deeply nested .claude path should still be allowed.""" + from worktree_guard import is_allowed_path + + assert is_allowed_path("/Users/mj/.claude/some/deep/path/file.md") is True + + def test_docs_nested_deep(self): + """Deeply nested docs path should still be allowed.""" + from worktree_guard import is_allowed_path + + assert is_allowed_path("/Users/mj/project/docs/architecture/v2/design.md") is True + + def test_gitignore_anywhere(self): + """'.gitignore' file should be allowed regardless of location.""" + from worktree_guard import is_allowed_path + + assert is_allowed_path("/Users/mj/project/.gitignore") is True + assert is_allowed_path("/Users/mj/project/subdir/.gitignore") is True + + def test_docs_as_substring_not_matched(self): + """A path component 'mydocs' should NOT match the 'docs' pattern + (tests path component matching vs substring matching).""" + from worktree_guard import is_allowed_path + + # 'mydocs' should NOT match 'docs' pattern + assert is_allowed_path("/Users/mj/project/mydocs/file.md") is False + + def test_claude_as_substring_not_matched(self): + """A directory named 'not-claude' should NOT match '.claude' pattern.""" + from worktree_guard import is_allowed_path + + assert is_allowed_path("/Users/mj/not-claude/file.md") is False + + def test_claude_md_in_any_directory(self): + """CLAUDE.md should be allowed anywhere.""" + from worktree_guard import is_allowed_path + + assert is_allowed_path("/any/path/CLAUDE.md") is True + + +class TestIsApplicationCodeEdgeCases: + """Edge cases for worktree_guard.is_application_code().""" + + def test_recognizes_common_extensions(self): + from worktree_guard import is_application_code + + assert is_application_code("/project/src/app.py") is True + assert is_application_code("/project/lib/util.ts") is True + assert is_application_code("/project/test/test_app.py") is True + assert is_application_code("/project/scripts/deploy.sh") is True + assert is_application_code("/project/infra/main.tf") is True + + def test_rejects_non_app_extensions(self): + from worktree_guard import is_application_code + + assert is_application_code("/project/readme.txt") is False + assert is_application_code("/project/image.png") is False + assert is_application_code("/project/notes.pdf") is False + + def test_recognizes_app_code_dirs(self): + from worktree_guard import is_application_code + + assert is_application_code("/project/src/component.vue") is True + assert is_application_code("/project/lib/helper.rb") is True + assert is_application_code("/project/app/main.go") is True + assert is_application_code("/project/tests/test_api.py") is True + + def test_markdown_outside_app_dirs_not_app_code(self): + """Markdown file outside app directories should NOT be app code.""" + from worktree_guard import is_application_code + + assert is_application_code("/project/README.md") is False + + def test_yaml_in_app_dir_is_app_code(self): + """YAML config in src/ is considered app code.""" + from worktree_guard import is_application_code + + assert is_application_code("/project/src/config.yaml") is True + + +class TestCheckWorktreeBoundaryEdgeCases: + """Edge cases for check_worktree_boundary().""" + + def test_non_app_code_outside_worktree_allowed(self): + """Non-application code outside worktree should be allowed.""" + from worktree_guard import check_worktree_boundary + + result = check_worktree_boundary( + file_path="/Users/mj/project/notes.txt", + worktree_path="/tmp/worktrees/feat-auth" + ) + assert result is None # Not blocked + + def test_worktree_path_trailing_slash(self): + """Should work with trailing slash on worktree path.""" + from worktree_guard import check_worktree_boundary + + result = check_worktree_boundary( + file_path="/tmp/worktrees/feat-auth/src/auth.ts", + worktree_path="/tmp/worktrees/feat-auth/" + ) + assert result is None + + def test_exact_worktree_path_file(self): + """File at the exact worktree root should be allowed.""" + from worktree_guard import check_worktree_boundary + + result = check_worktree_boundary( + file_path="/tmp/worktrees/feat-auth/package.json", + worktree_path="/tmp/worktrees/feat-auth" + ) + assert result is None + + def test_blocks_python_outside_worktree(self): + """Python file outside worktree should be blocked.""" + from worktree_guard import check_worktree_boundary + + result = check_worktree_boundary( + file_path="/Users/mj/project/hooks/session_init.py", + worktree_path="/tmp/worktrees/feat-auth" + ) + assert result is not None + assert "outside the active worktree" in result.lower() + + +class TestSuggestWorktreePathEdgeCases: + """Additional edge cases for _suggest_worktree_path().""" + + def test_nested_worktree_path(self, tmp_path): + """Should handle deeply nested worktree paths.""" + from worktree_guard import _suggest_worktree_path + + project = tmp_path / "repo" + worktree = project / ".worktrees" / "feat" / "deep-branch" + src = project / "pact-plugin" / "hooks" + worktree.mkdir(parents=True) + src.mkdir(parents=True) + (src / "session_init.py").touch() + + result = _suggest_worktree_path( + str(src / "session_init.py"), + str(worktree) + ) + assert result is not None + assert "pact-plugin" in result + assert "session_init.py" in result + + def test_same_path_returns_itself(self, tmp_path): + """File inside worktree should return the same path.""" + from worktree_guard import _suggest_worktree_path + + worktree = tmp_path / ".worktrees" / "feat" + worktree.mkdir(parents=True) + src = worktree / "src" + src.mkdir() + (src / "app.py").touch() + + result = _suggest_worktree_path( + str(src / "app.py"), + str(worktree) + ) + # Should return something (the file is inside worktree, so suggestion = same location) + assert result is not None From 358941bfa1b7f8aa92a29fb71aff8c041ad21e85 Mon Sep 17 00:00:00 2001 From: michael-wojcik <5386199+michael-wojcik@users.noreply.github.com> Date: Sat, 21 Feb 2026 02:34:34 -0500 Subject: [PATCH 2/3] docs: sync pact-protocols.md with v3.5.0 standalone protocol files MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Update 3 stale sections to match their source-of-truth files: - imPACT: "Two questions, Three outcomes" → "Three questions, Five outcomes" - Algedonic: add broadcast HALT for parallel agents - Stall Detection: already in sync, no changes needed Closes part of #206 --- pact-plugin/protocols/pact-protocols.md | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/pact-plugin/protocols/pact-protocols.md b/pact-plugin/protocols/pact-protocols.md index 020453b4..13a03d0d 100644 --- a/pact-plugin/protocols/pact-protocols.md +++ b/pact-plugin/protocols/pact-protocols.md @@ -629,6 +629,7 @@ For full protocol details, see [algedonic.md](algedonic.md). - **Any agent** can emit algedonic signals when they recognize trigger conditions - Orchestrator **MUST** surface signals to user immediately—cannot suppress or delay - HALT requires user acknowledgment before ANY work resumes +- For **HALT** with parallel agents: broadcast stop to all teammates via `SendMessage(type="broadcast")`, preserve work-in-progress, do NOT commit partial work - ALERT allows user to choose: Investigate / Continue / Stop ### Relationship to imPACT @@ -755,18 +756,21 @@ At phase transitions, briefly assess: **Trigger when**: Blocked; get similar errors repeatedly; or prior phase output is wrong. -**Two questions**: +**Three questions**: 1. **Redo prior phase?** — Is the issue upstream in P→A→C→T? -2. **Additional agents needed?** — Do I need subagents to assist? +2. **Additional agents needed?** — Do we need help beyond the blocked agent's scope/specialty? +3. **Is the agent recoverable?** — Can the blocked agent be resumed or helped, or is it unrecoverable (looping, stalled, context exhausted)? -**Three outcomes**: +**Five outcomes**: | Outcome | When | Action | |---------|------|--------| -| Redo solo | Prior phase broken, I can fix it | Loop back and fix yourself | -| Redo with help | Prior phase broken, need specialist | Loop back with subagent assistance | -| Proceed with help | Current phase correct, blocked on execution | Invoke subagents to help forward | +| Redo prior phase | Issue is upstream in P→A→C→T | Re-delegate to relevant agent(s) to redo the prior phase | +| Augment present phase | Need help in current phase | Re-invoke blocked agent with additional context + parallel agents | +| Invoke rePACT | Sub-task needs own P→A→C→T cycle | Use `/PACT:rePACT` for nested cycle | +| Terminate agent | Agent unrecoverable (infinite loop, context exhaustion, stall after resume) | `TaskStop(taskId)` + fresh spawn with partial handoff | +| Not truly blocked | Neither question is "Yes" | Instruct agent to continue with clarified guidance | -If neither question is "Yes," you're not blocked—continue. +If 3+ consecutive imPACT cycles fail to resolve, escalate to user (proto-algedonic ALERT: META-BLOCK). --- From 887d54c7a880db0a621542b56afca819ef344084 Mon Sep 17 00:00:00 2001 From: michael-wojcik <5386199+michael-wojcik@users.noreply.github.com> Date: Sat, 21 Feb 2026 02:34:40 -0500 Subject: [PATCH 3/3] docs: tighten imPACT.md and wrap-up.md verbose additions - imPACT.md: condense "When to terminate" from 12 lines to 1 paragraph - wrap-up.md: remove belt-and-suspenders note about session_end.py Closes #206 --- pact-plugin/commands/imPACT.md | 11 +---------- pact-plugin/commands/wrap-up.md | 2 -- 2 files changed, 1 insertion(+), 12 deletions(-) diff --git a/pact-plugin/commands/imPACT.md b/pact-plugin/commands/imPACT.md index e382294e..af64129e 100644 --- a/pact-plugin/commands/imPACT.md +++ b/pact-plugin/commands/imPACT.md @@ -122,16 +122,7 @@ If the blocker reveals that a sub-task is more complex than expected and needs i /PACT:rePACT backend "implement the OAuth2 token refresh that's blocking us" ``` -**When to terminate**: -Terminate is a last resort when the agent cannot be productively resumed: -- Agent was already resumed once and stalled again on the same issue -- Agent is looping on the same error after 3+ attempts -- Agent's context is exhausted (near capacity, responses becoming incoherent) -- TeammateIdle hook reported stall and resume did not resolve it - -**Note**: `TaskStop` is a **force-stop** -- it terminates the agent immediately without giving it a chance to finish current work or save state. For cooperative shutdown (letting the agent complete current work), use `SendMessage(type="shutdown_request")` instead. - -After termination, spawn a fresh agent with the partial handoff context from the terminated agent's task metadata. The fresh agent gets a clean context window without the failed approaches. +**When to terminate**: Last resort — agent resumed once and stalled again, looping on same error 3+ times, context exhausted, or TeammateIdle stall unresolved by resume. `TaskStop` is a force-stop (immediate, non-cooperative); use `SendMessage(type="shutdown_request")` for cooperative shutdown. After termination, spawn a fresh agent with partial handoff from the terminated agent's task metadata. --- diff --git a/pact-plugin/commands/wrap-up.md b/pact-plugin/commands/wrap-up.md index 6078050b..66050b4a 100644 --- a/pact-plugin/commands/wrap-up.md +++ b/pact-plugin/commands/wrap-up.md @@ -69,5 +69,3 @@ Clean up the session team to free resources: 1. **Shut down remaining teammates**: Send `shutdown_request` to each active teammate and wait for responses. 2. **Delete the team**: Call `TeamDelete` to remove the team directory (`~/.claude/teams/{team_name}/`). 3. **Handle failures**: If `TeamDelete` fails because active members remain, report which teammates are still running and ask the user whether to force shutdown or leave them. - -> Note: `hooks/session_end.py` also performs best-effort cleanup of stale team directories from prior sessions. This manual step ensures the *current* session's team is cleanly shut down.