#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""Claude Code / Factory SubagentStop hook for tracking subagent (Task) results.
This hook runs when a subagent (Task tool call) finishes. It captures
the subagent's work for memory context, enabling better coordination
between subagents and the main agent.
Usage:
Configure in ~/.claude/settings.json (Claude Code) or ~/.factory/settings.json (Factory):
{
"hooks": {
"SubagentStop": [
{
"hooks": [
{
"type": "command",
"command": "python /path/to/recall/hooks/recall-subagent.py",
"timeout": 10
}
]
}
]
}
}
Input (via stdin JSON):
{
"session_id": "abc123",
"transcript_path": "/path/to/transcript.jsonl",
"cwd": "/project/root",
"permission_mode": "default",
"hook_event_name": "SubagentStop",
"stop_hook_active": false
}
Output:
- JSON with decision: "block" to continue, or nothing to allow stop
"""
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from recall_client import DaemonClient
def read_hook_input() -> dict:
"""Read hook input from stdin."""
try:
if sys.stdin.isatty():
return {}
stdin_data = sys.stdin.read()
if stdin_data:
return json.loads(stdin_data)
except (OSError, json.JSONDecodeError):
pass
return {}
def get_project_namespace() -> str:
"""Derive project namespace from current working directory."""
cwd = str(Path.cwd())
project_name = Path(cwd).name
project_indicators = [
".git",
"pyproject.toml",
"package.json",
"Cargo.toml",
"go.mod",
]
for indicator in project_indicators:
if Path(cwd, indicator).exists():
return f"project:{project_name}"
return "global"
def read_transcript_tail(transcript_path: str | None, lines: int = 30) -> str | None:
"""Read the last N lines of the transcript."""
if not transcript_path:
return None
try:
path = Path(transcript_path).expanduser()
if not path.exists():
return None
content = path.read_text()
all_lines = content.strip().split("\n")
return "\n".join(all_lines[-lines:])
except Exception:
return None
def extract_subagent_summary(transcript_tail: str) -> str | None:
"""Extract a summary of what the subagent accomplished.
Parses the transcript to find the subagent's final output/report.
"""
if not transcript_tail:
return None
# Look for common subagent completion patterns
lines = transcript_tail.split("\n")
summary_lines = []
for line in lines[-20:]: # Focus on last 20 lines
# Skip empty lines and tool metadata
if not line.strip():
continue
if line.startswith("{") and '"tool' in line:
continue
# Look for result/summary indicators
lower_line = line.lower()
if any(
indicator in lower_line
for indicator in [
"completed",
"finished",
"done",
"result",
"summary",
"found",
"created",
"updated",
"fixed",
"implemented",
]
):
summary_lines.append(line.strip())
if summary_lines:
return " | ".join(summary_lines[-3:]) # Last 3 relevant lines
return None
def get_daemon_client() -> DaemonClient:
"""Get a DaemonClient instance with default settings.
Returns:
Configured DaemonClient for IPC communication.
"""
return DaemonClient(
connect_timeout=2.0,
request_timeout=5.0,
auto_fallback=True,
)
def record_subagent_activity(
session_id: str,
namespace: str,
summary: str | None,
) -> None:
"""Record subagent activity for tracking."""
log_path = Path.home() / ".claude" / "hooks" / "logs" / "recall-subagent.log"
log_path.parent.mkdir(parents=True, exist_ok=True)
try:
with log_path.open("a") as f:
summary_preview = (
(summary[:100] + "...") if summary and len(summary) > 100 else summary
)
f.write(
f"{datetime.now(tz=timezone.utc).isoformat()} | SUBAGENT_STOP | session={session_id} | namespace={namespace} | summary={summary_preview}\n",
)
except Exception:
pass
def main():
"""Main hook entry point.
Tracks subagent completion for:
1. Analytics on subagent usage
2. Capturing subagent results for future reference
3. Enabling memory sharing between subagents
"""
try:
hook_input = read_hook_input()
session_id = hook_input.get("session_id") or hook_input.get(
"sessionId",
"unknown",
)
transcript_path = hook_input.get("transcript_path") or hook_input.get(
"transcriptPath",
)
cwd = hook_input.get("cwd", str(Path.cwd()))
stop_hook_active = hook_input.get("stop_hook_active", False)
# Prevent infinite loops
if stop_hook_active:
return
# Change to session's working directory
if cwd:
os.chdir(cwd)
namespace = get_project_namespace()
# Read recent transcript to understand what the subagent did
transcript_tail = read_transcript_tail(transcript_path)
summary = extract_subagent_summary(transcript_tail)
# Record the activity
record_subagent_activity(session_id, namespace, summary)
# ALWAYS store subagent activity - no exceptions
with get_daemon_client() as client:
if summary and len(summary) > 20:
client.store(
content=f"Subagent completed: {summary}",
namespace=namespace,
memory_type="session",
importance=0.4,
metadata={
"source": "recall-subagent",
"session_id": session_id,
"is_subagent_result": True,
},
)
else:
# FALLBACK: Always store at least a record
client.store(
content=f"Subagent session {session_id} completed in {namespace}",
namespace=namespace,
memory_type="session",
importance=0.2,
metadata={
"source": "recall-subagent",
"session_id": session_id,
"is_subagent_result": True,
"fallback": True,
},
)
# Output memory reflection prompt for Claude
reflection_prompt = """🤖 SUBAGENT COMPLETE - Memory Storage REQUIRED
You MUST evaluate subagent results for storage:
- What subagent discovered (MUST store if novel)
- Patterns during delegation (SHOULD store)
- Insights for future tasks (MUST store if actionable)
You MUST NOT discard subagent insights without evaluation.
Store via mcp-exec:
```javascript
await recall.memory_store_tool({
content: "Subagent insight: ...",
memory_type: "pattern", // or "preference", "decision", "golden_rule"
importance: 0.0-1.0, // 0.3 minor, 0.5 useful, 0.7 important, 0.9+ critical/golden_rule
namespace: "project:name" // or "global"
});
```
Use mcp__mcp-exec__execute_code_with_wrappers with wrappers: ["recall"]"""
import json
output = {"systemMessage": reflection_prompt}
print(json.dumps(output))
sys.stdout.flush()
except BrokenPipeError:
pass
except Exception as e:
# Log error but don't block
try:
log_path = (
Path.home() / ".claude" / "hooks" / "logs" / "recall-subagent.log"
)
log_path.parent.mkdir(parents=True, exist_ok=True)
with log_path.open("a") as f:
f.write(f"{datetime.now(tz=timezone.utc).isoformat()} | ERROR | {e}\n")
except Exception:
pass
if __name__ == "__main__":
main()
