first commit
This commit is contained in:
@@ -0,0 +1,803 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Multi-Agent Pipeline Context Injection Hook
|
||||
|
||||
Core Design Philosophy:
|
||||
- Dispatch becomes a pure dispatcher, only responsible for "calling subagents"
|
||||
- Hook is responsible for injecting all context, subagent works autonomously with complete info
|
||||
- Each agent has a dedicated jsonl file defining its context
|
||||
- No resume needed, no segmentation, behavior controlled by code not prompt
|
||||
|
||||
Trigger: PreToolUse (before Task tool call)
|
||||
|
||||
Context Source: .trellis/.current-task points to task directory
|
||||
- implement.jsonl - Implement agent dedicated context
|
||||
- check.jsonl - Check agent dedicated context
|
||||
- debug.jsonl - Debug agent dedicated context
|
||||
- research.jsonl - Research agent dedicated context (optional, usually not needed)
|
||||
- cr.jsonl - Code review dedicated context
|
||||
- prd.md - Requirements document
|
||||
- info.md - Technical design
|
||||
- codex-review-output.txt - Code Review results
|
||||
"""
|
||||
|
||||
# IMPORTANT: Suppress all warnings FIRST
|
||||
import warnings
|
||||
warnings.filterwarnings("ignore")
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# IMPORTANT: Force stdout to use UTF-8 on Windows
|
||||
# This fixes UnicodeEncodeError when outputting non-ASCII characters
|
||||
if sys.platform == "win32":
|
||||
import io as _io
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
sys.stdout.reconfigure(encoding="utf-8", errors="replace") # type: ignore[union-attr]
|
||||
elif hasattr(sys.stdout, "detach"):
|
||||
sys.stdout = _io.TextIOWrapper(sys.stdout.detach(), encoding="utf-8", errors="replace") # type: ignore[union-attr]
|
||||
|
||||
# =============================================================================
|
||||
# Path Constants (change here to rename directories)
|
||||
# =============================================================================
|
||||
|
||||
DIR_WORKFLOW = ".trellis"
|
||||
DIR_WORKSPACE = "workspace"
|
||||
DIR_TASKS = "tasks"
|
||||
DIR_SPEC = "spec"
|
||||
FILE_CURRENT_TASK = ".current-task"
|
||||
FILE_TASK_JSON = "task.json"
|
||||
|
||||
# Agents that don't update phase (can be called at any time)
|
||||
AGENTS_NO_PHASE_UPDATE = {"debug", "research"}
|
||||
|
||||
# =============================================================================
|
||||
# Subagent Constants (change here to rename subagent types)
|
||||
# =============================================================================
|
||||
|
||||
AGENT_IMPLEMENT = "implement"
|
||||
AGENT_CHECK = "check"
|
||||
AGENT_DEBUG = "debug"
|
||||
AGENT_RESEARCH = "research"
|
||||
|
||||
# Agents that require a task directory
|
||||
AGENTS_REQUIRE_TASK = (AGENT_IMPLEMENT, AGENT_CHECK, AGENT_DEBUG)
|
||||
# All supported agents
|
||||
AGENTS_ALL = (AGENT_IMPLEMENT, AGENT_CHECK, AGENT_DEBUG, AGENT_RESEARCH)
|
||||
|
||||
|
||||
def find_repo_root(start_path: str) -> str | None:
|
||||
"""
|
||||
Find git repo root from start_path upwards
|
||||
|
||||
Returns:
|
||||
Repo root path, or None if not found
|
||||
"""
|
||||
current = Path(start_path).resolve()
|
||||
while current != current.parent:
|
||||
if (current / ".git").exists():
|
||||
return str(current)
|
||||
current = current.parent
|
||||
return None
|
||||
|
||||
|
||||
def get_current_task(repo_root: str) -> str | None:
|
||||
"""
|
||||
Read current task directory path from .trellis/.current-task
|
||||
|
||||
Returns:
|
||||
Task directory relative path (relative to repo_root)
|
||||
None if not set
|
||||
"""
|
||||
current_task_file = os.path.join(repo_root, DIR_WORKFLOW, FILE_CURRENT_TASK)
|
||||
if not os.path.exists(current_task_file):
|
||||
return None
|
||||
|
||||
try:
|
||||
with open(current_task_file, "r", encoding="utf-8") as f:
|
||||
content = f.read().strip()
|
||||
if not content:
|
||||
return None
|
||||
normalized = content.replace("\\", "/")
|
||||
while normalized.startswith("./"):
|
||||
normalized = normalized[2:]
|
||||
if normalized.startswith("tasks/"):
|
||||
normalized = f".trellis/{normalized}"
|
||||
return normalized
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def update_current_phase(repo_root: str, task_dir: str, subagent_type: str) -> None:
|
||||
"""
|
||||
Update current_phase in task.json based on subagent_type.
|
||||
|
||||
This ensures phase tracking is always accurate, regardless of whether
|
||||
dispatch agent remembers to update it.
|
||||
|
||||
Logic:
|
||||
- Read next_action array from task.json
|
||||
- Find the next phase whose action matches subagent_type
|
||||
- Only move forward, never backward
|
||||
- Some agents (debug, research) don't update phase
|
||||
"""
|
||||
if subagent_type in AGENTS_NO_PHASE_UPDATE:
|
||||
return
|
||||
|
||||
task_json_path = os.path.join(repo_root, task_dir, FILE_TASK_JSON)
|
||||
if not os.path.exists(task_json_path):
|
||||
return
|
||||
|
||||
try:
|
||||
with open(task_json_path, "r", encoding="utf-8") as f:
|
||||
task_data = json.load(f)
|
||||
|
||||
current_phase = task_data.get("current_phase", 0)
|
||||
next_actions = task_data.get("next_action", [])
|
||||
|
||||
# Map action names to subagent types
|
||||
# "implement" -> "implement", "check" -> "check", "finish" -> "check"
|
||||
action_to_agent = {
|
||||
"implement": "implement",
|
||||
"check": "check",
|
||||
"finish": "check", # finish uses check agent
|
||||
}
|
||||
|
||||
# Find the next phase that matches this subagent_type
|
||||
new_phase = None
|
||||
for action in next_actions:
|
||||
phase_num = action.get("phase", 0)
|
||||
action_name = action.get("action", "")
|
||||
expected_agent = action_to_agent.get(action_name)
|
||||
|
||||
# Only consider phases after current_phase
|
||||
if phase_num > current_phase and expected_agent == subagent_type:
|
||||
new_phase = phase_num
|
||||
break
|
||||
|
||||
if new_phase is not None:
|
||||
task_data["current_phase"] = new_phase
|
||||
|
||||
with open(task_json_path, "w", encoding="utf-8") as f:
|
||||
json.dump(task_data, f, indent=2, ensure_ascii=False)
|
||||
except Exception:
|
||||
# Don't fail the hook if phase update fails
|
||||
pass
|
||||
|
||||
|
||||
def read_file_content(base_path: str, file_path: str) -> str | None:
|
||||
"""Read file content, return None if file doesn't exist"""
|
||||
full_path = os.path.join(base_path, file_path)
|
||||
if os.path.exists(full_path) and os.path.isfile(full_path):
|
||||
try:
|
||||
with open(full_path, "r", encoding="utf-8") as f:
|
||||
return f.read()
|
||||
except Exception:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def read_directory_contents(
|
||||
base_path: str, dir_path: str, max_files: int = 20
|
||||
) -> list[tuple[str, str]]:
|
||||
"""
|
||||
Read all .md files in a directory
|
||||
|
||||
Args:
|
||||
base_path: Base path (usually repo_root)
|
||||
dir_path: Directory relative path
|
||||
max_files: Max files to read (prevent huge directories)
|
||||
|
||||
Returns:
|
||||
[(file_path, content), ...]
|
||||
"""
|
||||
full_path = os.path.join(base_path, dir_path)
|
||||
if not os.path.exists(full_path) or not os.path.isdir(full_path):
|
||||
return []
|
||||
|
||||
results = []
|
||||
try:
|
||||
# Only read .md files, sorted by filename
|
||||
md_files = sorted(
|
||||
[
|
||||
f
|
||||
for f in os.listdir(full_path)
|
||||
if f.endswith(".md") and os.path.isfile(os.path.join(full_path, f))
|
||||
]
|
||||
)
|
||||
|
||||
for filename in md_files[:max_files]:
|
||||
file_full_path = os.path.join(full_path, filename)
|
||||
relative_path = os.path.join(dir_path, filename)
|
||||
try:
|
||||
with open(file_full_path, "r", encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
results.append((relative_path, content))
|
||||
except Exception:
|
||||
continue
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def read_jsonl_entries(base_path: str, jsonl_path: str) -> list[tuple[str, str]]:
|
||||
"""
|
||||
Read all file/directory contents referenced in jsonl file
|
||||
|
||||
Schema:
|
||||
{"file": "path/to/file.md", "reason": "..."}
|
||||
{"file": "path/to/dir/", "type": "directory", "reason": "..."}
|
||||
|
||||
Returns:
|
||||
[(path, content), ...]
|
||||
"""
|
||||
full_path = os.path.join(base_path, jsonl_path)
|
||||
if not os.path.exists(full_path):
|
||||
return []
|
||||
|
||||
results = []
|
||||
try:
|
||||
with open(full_path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
item = json.loads(line)
|
||||
file_path = item.get("file") or item.get("path")
|
||||
entry_type = item.get("type", "file")
|
||||
|
||||
if not file_path:
|
||||
continue
|
||||
|
||||
if entry_type == "directory":
|
||||
# Read all .md files in directory
|
||||
dir_contents = read_directory_contents(base_path, file_path)
|
||||
results.extend(dir_contents)
|
||||
else:
|
||||
# Read single file
|
||||
content = read_file_content(base_path, file_path)
|
||||
if content:
|
||||
results.append((file_path, content))
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def get_agent_context(repo_root: str, task_dir: str, agent_type: str) -> str:
|
||||
"""
|
||||
Get complete context for specified agent
|
||||
|
||||
Prioritize agent-specific jsonl, fallback to spec.jsonl if not exists
|
||||
"""
|
||||
context_parts = []
|
||||
|
||||
# 1. Try agent-specific jsonl
|
||||
agent_jsonl = f"{task_dir}/{agent_type}.jsonl"
|
||||
agent_entries = read_jsonl_entries(repo_root, agent_jsonl)
|
||||
|
||||
# 2. If agent-specific jsonl doesn't exist or empty, fallback to spec.jsonl
|
||||
if not agent_entries:
|
||||
agent_entries = read_jsonl_entries(repo_root, f"{task_dir}/spec.jsonl")
|
||||
|
||||
# 3. Add all files from jsonl
|
||||
for file_path, content in agent_entries:
|
||||
context_parts.append(f"=== {file_path} ===\n{content}")
|
||||
|
||||
return "\n\n".join(context_parts)
|
||||
|
||||
|
||||
def get_implement_context(repo_root: str, task_dir: str) -> str:
|
||||
"""
|
||||
Complete context for Implement Agent
|
||||
|
||||
Read order:
|
||||
1. All files in implement.jsonl (dev specs)
|
||||
2. prd.md (requirements)
|
||||
3. info.md (technical design)
|
||||
"""
|
||||
context_parts = []
|
||||
|
||||
# 1. Read implement.jsonl (or fallback to spec.jsonl)
|
||||
base_context = get_agent_context(repo_root, task_dir, "implement")
|
||||
if base_context:
|
||||
context_parts.append(base_context)
|
||||
|
||||
# 2. Requirements document
|
||||
prd_content = read_file_content(repo_root, f"{task_dir}/prd.md")
|
||||
if prd_content:
|
||||
context_parts.append(f"=== {task_dir}/prd.md (Requirements) ===\n{prd_content}")
|
||||
|
||||
# 3. Technical design
|
||||
info_content = read_file_content(repo_root, f"{task_dir}/info.md")
|
||||
if info_content:
|
||||
context_parts.append(
|
||||
f"=== {task_dir}/info.md (Technical Design) ===\n{info_content}"
|
||||
)
|
||||
|
||||
return "\n\n".join(context_parts)
|
||||
|
||||
|
||||
def get_check_context(repo_root: str, task_dir: str) -> str:
|
||||
"""
|
||||
Complete context for Check Agent
|
||||
|
||||
Read order:
|
||||
1. All files in check.jsonl (check specs + dev specs)
|
||||
2. prd.md (for understanding task intent)
|
||||
"""
|
||||
context_parts = []
|
||||
|
||||
# 1. Read check.jsonl (or fallback to spec.jsonl + hardcoded check files)
|
||||
check_entries = read_jsonl_entries(repo_root, f"{task_dir}/check.jsonl")
|
||||
|
||||
if check_entries:
|
||||
for file_path, content in check_entries:
|
||||
context_parts.append(f"=== {file_path} ===\n{content}")
|
||||
else:
|
||||
# Fallback: use hardcoded check files + spec.jsonl
|
||||
check_files = [
|
||||
(".claude/commands/trellis/finish-work.md", "Finish work checklist"),
|
||||
(".claude/commands/trellis/check-cross-layer.md", "Cross-layer check spec"),
|
||||
(".claude/commands/trellis/check.md", "Code quality check spec"),
|
||||
]
|
||||
for file_path, description in check_files:
|
||||
content = read_file_content(repo_root, file_path)
|
||||
if content:
|
||||
context_parts.append(f"=== {file_path} ({description}) ===\n{content}")
|
||||
|
||||
# Add spec.jsonl
|
||||
spec_entries = read_jsonl_entries(repo_root, f"{task_dir}/spec.jsonl")
|
||||
for file_path, content in spec_entries:
|
||||
context_parts.append(f"=== {file_path} (Dev spec) ===\n{content}")
|
||||
|
||||
# 2. Requirements document (for understanding task intent)
|
||||
prd_content = read_file_content(repo_root, f"{task_dir}/prd.md")
|
||||
if prd_content:
|
||||
context_parts.append(
|
||||
f"=== {task_dir}/prd.md (Requirements - for understanding intent) ===\n{prd_content}"
|
||||
)
|
||||
|
||||
return "\n\n".join(context_parts)
|
||||
|
||||
|
||||
def get_finish_context(repo_root: str, task_dir: str) -> str:
|
||||
"""
|
||||
Complete context for Finish phase (final check before PR)
|
||||
|
||||
Read order:
|
||||
1. All files in finish.jsonl (if exists)
|
||||
2. Fallback to finish-work.md only (lightweight final check)
|
||||
3. update-spec.md (for active spec sync)
|
||||
4. prd.md (for verifying requirements are met)
|
||||
"""
|
||||
context_parts = []
|
||||
|
||||
# 1. Try finish.jsonl first
|
||||
finish_entries = read_jsonl_entries(repo_root, f"{task_dir}/finish.jsonl")
|
||||
|
||||
if finish_entries:
|
||||
for file_path, content in finish_entries:
|
||||
context_parts.append(f"=== {file_path} ===\n{content}")
|
||||
else:
|
||||
# Fallback: only finish-work.md (lightweight)
|
||||
finish_work = read_file_content(
|
||||
repo_root, ".claude/commands/trellis/finish-work.md"
|
||||
)
|
||||
if finish_work:
|
||||
context_parts.append(
|
||||
f"=== .claude/commands/trellis/finish-work.md (Finish checklist) ===\n{finish_work}"
|
||||
)
|
||||
|
||||
# 2. Spec update process (for active spec sync)
|
||||
update_spec = read_file_content(
|
||||
repo_root, ".claude/commands/trellis/update-spec.md"
|
||||
)
|
||||
if update_spec:
|
||||
context_parts.append(
|
||||
f"=== .claude/commands/trellis/update-spec.md (Spec update process) ===\n{update_spec}"
|
||||
)
|
||||
|
||||
# 3. Requirements document (for verifying requirements are met)
|
||||
prd_content = read_file_content(repo_root, f"{task_dir}/prd.md")
|
||||
if prd_content:
|
||||
context_parts.append(
|
||||
f"=== {task_dir}/prd.md (Requirements - verify all met) ===\n{prd_content}"
|
||||
)
|
||||
|
||||
return "\n\n".join(context_parts)
|
||||
|
||||
|
||||
def get_debug_context(repo_root: str, task_dir: str) -> str:
|
||||
"""
|
||||
Complete context for Debug Agent
|
||||
|
||||
Read order:
|
||||
1. All files in debug.jsonl (specs needed for fixing)
|
||||
2. codex-review-output.txt (Codex Review results)
|
||||
"""
|
||||
context_parts = []
|
||||
|
||||
# 1. Read debug.jsonl (or fallback to spec.jsonl + hardcoded check files)
|
||||
debug_entries = read_jsonl_entries(repo_root, f"{task_dir}/debug.jsonl")
|
||||
|
||||
if debug_entries:
|
||||
for file_path, content in debug_entries:
|
||||
context_parts.append(f"=== {file_path} ===\n{content}")
|
||||
else:
|
||||
# Fallback: use spec.jsonl + hardcoded check files
|
||||
spec_entries = read_jsonl_entries(repo_root, f"{task_dir}/spec.jsonl")
|
||||
for file_path, content in spec_entries:
|
||||
context_parts.append(f"=== {file_path} (Dev spec) ===\n{content}")
|
||||
|
||||
check_files = [
|
||||
(".claude/commands/trellis/check.md", "Code quality check spec"),
|
||||
(".claude/commands/trellis/check-cross-layer.md", "Cross-layer check spec"),
|
||||
]
|
||||
for file_path, description in check_files:
|
||||
content = read_file_content(repo_root, file_path)
|
||||
if content:
|
||||
context_parts.append(f"=== {file_path} ({description}) ===\n{content}")
|
||||
|
||||
# 2. Codex review output (if exists)
|
||||
codex_output = read_file_content(repo_root, f"{task_dir}/codex-review-output.txt")
|
||||
if codex_output:
|
||||
context_parts.append(
|
||||
f"=== {task_dir}/codex-review-output.txt (Codex Review Results) ===\n{codex_output}"
|
||||
)
|
||||
|
||||
return "\n\n".join(context_parts)
|
||||
|
||||
|
||||
def build_implement_prompt(original_prompt: str, context: str) -> str:
|
||||
"""Build complete prompt for Implement"""
|
||||
return f"""# Implement Agent Task
|
||||
|
||||
You are the Implement Agent in the Multi-Agent Pipeline.
|
||||
|
||||
## Your Context
|
||||
|
||||
All the information you need has been prepared for you:
|
||||
|
||||
{context}
|
||||
|
||||
---
|
||||
|
||||
## Your Task
|
||||
|
||||
{original_prompt}
|
||||
|
||||
---
|
||||
|
||||
## Workflow
|
||||
|
||||
1. **Understand specs** - All dev specs are injected above, understand them
|
||||
2. **Understand requirements** - Read requirements document and technical design
|
||||
3. **Implement feature** - Implement following specs and design
|
||||
4. **Self-check** - Ensure code quality against check specs
|
||||
|
||||
## Important Constraints
|
||||
|
||||
- Do NOT execute git commit, only code modifications
|
||||
- Follow all dev specs injected above
|
||||
- Report list of modified/created files when done"""
|
||||
|
||||
|
||||
def build_check_prompt(original_prompt: str, context: str) -> str:
|
||||
"""Build complete prompt for Check"""
|
||||
return f"""# Check Agent Task
|
||||
|
||||
You are the Check Agent in the Multi-Agent Pipeline (code and cross-layer checker).
|
||||
|
||||
## Your Context
|
||||
|
||||
All check specs and dev specs you need:
|
||||
|
||||
{context}
|
||||
|
||||
---
|
||||
|
||||
## Your Task
|
||||
|
||||
{original_prompt}
|
||||
|
||||
---
|
||||
|
||||
## Workflow
|
||||
|
||||
1. **Get changes** - Run `git diff --name-only` and `git diff` to get code changes
|
||||
2. **Check against specs** - Check item by item against specs above
|
||||
3. **Self-fix** - Fix issues directly, don't just report
|
||||
4. **Run verification** - Run project's lint and typecheck commands
|
||||
|
||||
## Important Constraints
|
||||
|
||||
- Fix issues yourself, don't just report
|
||||
- Must execute complete checklist in check specs
|
||||
- Pay special attention to impact radius analysis (L1-L5)"""
|
||||
|
||||
|
||||
def build_finish_prompt(original_prompt: str, context: str) -> str:
|
||||
"""Build complete prompt for Finish (final check before PR)"""
|
||||
return f"""# Finish Agent Task
|
||||
|
||||
You are performing the final check before creating a PR.
|
||||
|
||||
## Your Context
|
||||
|
||||
Finish checklist and requirements:
|
||||
|
||||
{context}
|
||||
|
||||
---
|
||||
|
||||
## Your Task
|
||||
|
||||
{original_prompt}
|
||||
|
||||
---
|
||||
|
||||
## Workflow
|
||||
|
||||
1. **Review changes** - Run `git diff --name-only` to see all changed files
|
||||
2. **Verify requirements** - Check each requirement in prd.md is implemented
|
||||
3. **Spec sync** - Analyze whether changes introduce new patterns, contracts, or conventions
|
||||
- If new pattern/convention found: read target spec file → update it → update index.md if needed
|
||||
- If infra/cross-layer change: follow the 7-section mandatory template from update-spec.md
|
||||
- If pure code fix with no new patterns: skip this step
|
||||
4. **Run final checks** - Execute lint and typecheck
|
||||
5. **Confirm ready** - Ensure code is ready for PR
|
||||
|
||||
## Important Constraints
|
||||
|
||||
- You MAY update spec files when gaps are detected (use update-spec.md as guide)
|
||||
- MUST read the target spec file BEFORE editing (avoid duplicating existing content)
|
||||
- Do NOT update specs for trivial changes (typos, formatting, obvious fixes)
|
||||
- If critical CODE issues found, report them clearly (fix specs, not code)
|
||||
- Verify all acceptance criteria in prd.md are met"""
|
||||
|
||||
|
||||
def build_debug_prompt(original_prompt: str, context: str) -> str:
|
||||
"""Build complete prompt for Debug"""
|
||||
return f"""# Debug Agent Task
|
||||
|
||||
You are the Debug Agent in the Multi-Agent Pipeline (issue fixer).
|
||||
|
||||
## Your Context
|
||||
|
||||
Dev specs and Codex Review results:
|
||||
|
||||
{context}
|
||||
|
||||
---
|
||||
|
||||
## Your Task
|
||||
|
||||
{original_prompt}
|
||||
|
||||
---
|
||||
|
||||
## Workflow
|
||||
|
||||
1. **Understand issues** - Analyze issues pointed out in Codex Review
|
||||
2. **Locate code** - Find positions that need fixing
|
||||
3. **Fix against specs** - Fix issues following dev specs
|
||||
4. **Verify fixes** - Run typecheck to ensure no new issues
|
||||
|
||||
## Important Constraints
|
||||
|
||||
- Do NOT execute git commit, only code modifications
|
||||
- Run typecheck after each fix to verify
|
||||
- Report which issues were fixed and which files were modified"""
|
||||
|
||||
|
||||
def get_research_context(repo_root: str, task_dir: str | None) -> str:
|
||||
"""
|
||||
Context for Research Agent
|
||||
|
||||
Research doesn't need much preset context, only needs:
|
||||
1. Project structure overview (where spec directories are)
|
||||
2. Optional research.jsonl (if there are specific search needs)
|
||||
"""
|
||||
context_parts = []
|
||||
|
||||
# 1. Project structure overview (dynamically discover spec directories)
|
||||
spec_path = f"{DIR_WORKFLOW}/{DIR_SPEC}"
|
||||
spec_root = Path(repo_root) / DIR_WORKFLOW / DIR_SPEC
|
||||
|
||||
# Build spec tree dynamically
|
||||
tree_lines = [f"{spec_path}/"]
|
||||
if spec_root.is_dir():
|
||||
pkg_dirs = sorted(d for d in spec_root.iterdir() if d.is_dir())
|
||||
for i, pkg_dir in enumerate(pkg_dirs):
|
||||
is_last = i == len(pkg_dirs) - 1
|
||||
prefix = "└── " if is_last else "├── "
|
||||
layers = sorted(d.name for d in pkg_dir.iterdir() if d.is_dir())
|
||||
layer_info = f" ({', '.join(layers)})" if layers else ""
|
||||
tree_lines.append(f"{prefix}{pkg_dir.name}/{layer_info}")
|
||||
|
||||
spec_tree = "\n".join(tree_lines)
|
||||
|
||||
project_structure = f"""## Project Spec Directory Structure
|
||||
|
||||
```
|
||||
{spec_tree}
|
||||
```
|
||||
|
||||
To get structured package info, run: `python3 ./{DIR_WORKFLOW}/scripts/get_context.py --mode packages`
|
||||
|
||||
## Search Tips
|
||||
|
||||
- Spec files: `{spec_path}/**/*.md`
|
||||
- Code search: Use Glob and Grep tools
|
||||
- Tech solutions: Use mcp__exa__web_search_exa or mcp__exa__get_code_context_exa"""
|
||||
|
||||
context_parts.append(project_structure)
|
||||
|
||||
# 2. If task directory exists, try reading research.jsonl (optional)
|
||||
if task_dir:
|
||||
research_entries = read_jsonl_entries(repo_root, f"{task_dir}/research.jsonl")
|
||||
if research_entries:
|
||||
context_parts.append(
|
||||
"\n## Additional Search Context (from research.jsonl)\n"
|
||||
)
|
||||
for file_path, content in research_entries:
|
||||
context_parts.append(f"=== {file_path} ===\n{content}")
|
||||
|
||||
return "\n\n".join(context_parts)
|
||||
|
||||
|
||||
def build_research_prompt(original_prompt: str, context: str) -> str:
|
||||
"""Build complete prompt for Research"""
|
||||
return f"""# Research Agent Task
|
||||
|
||||
You are the Research Agent in the Multi-Agent Pipeline (search researcher).
|
||||
|
||||
## Core Principle
|
||||
|
||||
**You do one thing: find and explain information.**
|
||||
|
||||
You are a documenter, not a reviewer.
|
||||
|
||||
## Project Info
|
||||
|
||||
{context}
|
||||
|
||||
---
|
||||
|
||||
## Your Task
|
||||
|
||||
{original_prompt}
|
||||
|
||||
---
|
||||
|
||||
## Workflow
|
||||
|
||||
1. **Understand query** - Determine search type (internal/external) and scope
|
||||
2. **Plan search** - List search steps for complex queries
|
||||
3. **Execute search** - Execute multiple independent searches in parallel
|
||||
4. **Organize results** - Output structured report
|
||||
|
||||
## Search Tools
|
||||
|
||||
| Tool | Purpose |
|
||||
|------|---------|
|
||||
| Glob | Search by filename pattern |
|
||||
| Grep | Search by content |
|
||||
| Read | Read file content |
|
||||
| mcp__exa__web_search_exa | External web search |
|
||||
| mcp__exa__get_code_context_exa | External code/doc search |
|
||||
|
||||
## Strict Boundaries
|
||||
|
||||
**Only allowed**: Describe what exists, where it is, how it works
|
||||
|
||||
**Forbidden** (unless explicitly asked):
|
||||
- Suggest improvements
|
||||
- Criticize implementation
|
||||
- Recommend refactoring
|
||||
- Modify any files
|
||||
|
||||
## Report Format
|
||||
|
||||
Provide structured search results including:
|
||||
- List of files found (with paths)
|
||||
- Code pattern analysis (if applicable)
|
||||
- Related spec documents
|
||||
- External references (if any)"""
|
||||
|
||||
|
||||
def main():
|
||||
try:
|
||||
input_data = json.load(sys.stdin)
|
||||
except json.JSONDecodeError:
|
||||
sys.exit(0)
|
||||
|
||||
tool_name = input_data.get("tool_name", "")
|
||||
|
||||
if tool_name not in ("Task", "Agent"):
|
||||
sys.exit(0)
|
||||
|
||||
tool_input = input_data.get("tool_input", {})
|
||||
subagent_type = tool_input.get("subagent_type", "")
|
||||
original_prompt = tool_input.get("prompt", "")
|
||||
cwd = input_data.get("cwd", os.getcwd())
|
||||
|
||||
# Only handle subagent types we care about
|
||||
if subagent_type not in AGENTS_ALL:
|
||||
sys.exit(0)
|
||||
|
||||
# Find repo root
|
||||
repo_root = find_repo_root(cwd)
|
||||
if not repo_root:
|
||||
sys.exit(0)
|
||||
|
||||
# Get current task directory (research doesn't require it)
|
||||
task_dir = get_current_task(repo_root)
|
||||
|
||||
# implement/check/debug need task directory
|
||||
if subagent_type in AGENTS_REQUIRE_TASK:
|
||||
if not task_dir:
|
||||
sys.exit(0)
|
||||
# Check if task directory exists
|
||||
task_dir_full = os.path.join(repo_root, task_dir)
|
||||
if not os.path.exists(task_dir_full):
|
||||
sys.exit(0)
|
||||
|
||||
# Update current_phase in task.json (system-level enforcement)
|
||||
update_current_phase(repo_root, task_dir, subagent_type)
|
||||
|
||||
# Check for [finish] marker in prompt (check agent with finish context)
|
||||
is_finish_phase = "[finish]" in original_prompt.lower()
|
||||
|
||||
# Get context and build prompt based on subagent type
|
||||
if subagent_type == AGENT_IMPLEMENT:
|
||||
assert task_dir is not None # validated above
|
||||
context = get_implement_context(repo_root, task_dir)
|
||||
new_prompt = build_implement_prompt(original_prompt, context)
|
||||
elif subagent_type == AGENT_CHECK:
|
||||
assert task_dir is not None # validated above
|
||||
if is_finish_phase:
|
||||
# Finish phase: use finish context (lighter, focused on final verification)
|
||||
context = get_finish_context(repo_root, task_dir)
|
||||
new_prompt = build_finish_prompt(original_prompt, context)
|
||||
else:
|
||||
# Regular check phase: use check context (full specs for self-fix loop)
|
||||
context = get_check_context(repo_root, task_dir)
|
||||
new_prompt = build_check_prompt(original_prompt, context)
|
||||
elif subagent_type == AGENT_DEBUG:
|
||||
assert task_dir is not None # validated above
|
||||
context = get_debug_context(repo_root, task_dir)
|
||||
new_prompt = build_debug_prompt(original_prompt, context)
|
||||
elif subagent_type == AGENT_RESEARCH:
|
||||
# Research can work without task directory
|
||||
context = get_research_context(repo_root, task_dir)
|
||||
new_prompt = build_research_prompt(original_prompt, context)
|
||||
else:
|
||||
sys.exit(0)
|
||||
|
||||
if not context:
|
||||
sys.exit(0)
|
||||
|
||||
# Return updated input with correct Claude Code PreToolUse format
|
||||
output = {
|
||||
"hookSpecificOutput": {
|
||||
"hookEventName": "PreToolUse",
|
||||
"permissionDecision": "allow",
|
||||
"updatedInput": {**tool_input, "prompt": new_prompt},
|
||||
}
|
||||
}
|
||||
|
||||
print(json.dumps(output, ensure_ascii=False))
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,396 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Ralph Loop - SubagentStop Hook for Check Agent Loop Control
|
||||
|
||||
Based on the Ralph Wiggum technique for autonomous agent loops.
|
||||
Uses completion promises to control when the check agent can stop.
|
||||
|
||||
Mechanism:
|
||||
- Intercepts when check subagent tries to stop (SubagentStop event)
|
||||
- If verify commands configured in worktree.yaml, runs them to verify
|
||||
- Otherwise, reads check.jsonl to get dynamic completion markers ({reason}_FINISH)
|
||||
- Blocks stopping until verification passes or all markers found
|
||||
- Has max iterations as safety limit
|
||||
|
||||
State file: .trellis/.ralph-state.json
|
||||
- Tracks current iteration count per session
|
||||
- Resets when task changes
|
||||
"""
|
||||
|
||||
# IMPORTANT: Suppress all warnings FIRST
|
||||
import warnings
|
||||
warnings.filterwarnings("ignore")
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
# IMPORTANT: Force stdout to use UTF-8 on Windows
|
||||
# This fixes UnicodeEncodeError when outputting non-ASCII characters
|
||||
if sys.platform == "win32":
|
||||
import io as _io
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
sys.stdout.reconfigure(encoding="utf-8", errors="replace") # type: ignore[union-attr]
|
||||
elif hasattr(sys.stdout, "detach"):
|
||||
sys.stdout = _io.TextIOWrapper(sys.stdout.detach(), encoding="utf-8", errors="replace") # type: ignore[union-attr]
|
||||
|
||||
# =============================================================================
|
||||
# Configuration
|
||||
# =============================================================================
|
||||
|
||||
MAX_ITERATIONS = 5 # Safety limit to prevent infinite loops
|
||||
STATE_TIMEOUT_MINUTES = 30 # Reset state if older than this
|
||||
STATE_FILE = ".trellis/.ralph-state.json"
|
||||
WORKTREE_YAML = ".trellis/worktree.yaml"
|
||||
DIR_WORKFLOW = ".trellis"
|
||||
FILE_CURRENT_TASK = ".current-task"
|
||||
|
||||
# Only control loop for check agent
|
||||
TARGET_AGENT = "check"
|
||||
|
||||
|
||||
def find_repo_root(start_path: str) -> str | None:
|
||||
"""Find git repo root from start_path upwards"""
|
||||
current = Path(start_path).resolve()
|
||||
while current != current.parent:
|
||||
if (current / ".git").exists():
|
||||
return str(current)
|
||||
current = current.parent
|
||||
return None
|
||||
|
||||
|
||||
def get_current_task(repo_root: str) -> str | None:
|
||||
"""Read current task directory path"""
|
||||
current_task_file = os.path.join(repo_root, DIR_WORKFLOW, FILE_CURRENT_TASK)
|
||||
if not os.path.exists(current_task_file):
|
||||
return None
|
||||
|
||||
try:
|
||||
with open(current_task_file, "r", encoding="utf-8") as f:
|
||||
content = f.read().strip()
|
||||
if not content:
|
||||
return None
|
||||
normalized = content.replace("\\", "/")
|
||||
while normalized.startswith("./"):
|
||||
normalized = normalized[2:]
|
||||
if normalized.startswith("tasks/"):
|
||||
normalized = f".trellis/{normalized}"
|
||||
return normalized
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def get_verify_commands(repo_root: str) -> list[str]:
|
||||
"""
|
||||
Read verify commands from worktree.yaml.
|
||||
|
||||
Returns list of commands to run, or empty list if not configured.
|
||||
Uses simple YAML parsing without external dependencies.
|
||||
"""
|
||||
yaml_path = os.path.join(repo_root, WORKTREE_YAML)
|
||||
if not os.path.exists(yaml_path):
|
||||
return []
|
||||
|
||||
try:
|
||||
with open(yaml_path, "r", encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
|
||||
# Simple YAML parsing for verify section
|
||||
# Look for "verify:" followed by list items
|
||||
lines = content.split("\n")
|
||||
in_verify_section = False
|
||||
commands = []
|
||||
|
||||
for line in lines:
|
||||
stripped = line.strip()
|
||||
|
||||
# Check for section start
|
||||
if stripped.startswith("verify:"):
|
||||
in_verify_section = True
|
||||
continue
|
||||
|
||||
# Check for new section (not indented, ends with :)
|
||||
if (
|
||||
not line.startswith(" ")
|
||||
and not line.startswith("\t")
|
||||
and stripped.endswith(":")
|
||||
and stripped != ""
|
||||
):
|
||||
in_verify_section = False
|
||||
continue
|
||||
|
||||
# If in verify section, look for list items
|
||||
if in_verify_section:
|
||||
# Skip comments and empty lines
|
||||
if stripped.startswith("#") or stripped == "":
|
||||
continue
|
||||
# Parse list item (- command)
|
||||
if stripped.startswith("- "):
|
||||
cmd = stripped[2:].strip()
|
||||
if cmd:
|
||||
commands.append(cmd)
|
||||
|
||||
return commands
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
def run_verify_commands(repo_root: str, commands: list[str]) -> tuple[bool, str]:
|
||||
"""
|
||||
Run verify commands and return (success, message).
|
||||
|
||||
All commands must pass for success.
|
||||
"""
|
||||
for cmd in commands:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
shell=True,
|
||||
cwd=repo_root,
|
||||
capture_output=True,
|
||||
timeout=120, # 2 minute timeout per command
|
||||
)
|
||||
if result.returncode != 0:
|
||||
stderr = result.stderr.decode("utf-8", errors="replace")
|
||||
stdout = result.stdout.decode("utf-8", errors="replace")
|
||||
error_output = stderr or stdout
|
||||
# Truncate long output
|
||||
if len(error_output) > 500:
|
||||
error_output = error_output[:500] + "..."
|
||||
return False, f"Command failed: {cmd}\n{error_output}"
|
||||
except subprocess.TimeoutExpired:
|
||||
return False, f"Command timed out: {cmd}"
|
||||
except Exception as e:
|
||||
return False, f"Command error: {cmd} - {str(e)}"
|
||||
|
||||
return True, "All verify commands passed"
|
||||
|
||||
|
||||
def get_completion_markers(repo_root: str, task_dir: str) -> list[str]:
|
||||
"""
|
||||
Read check.jsonl and generate completion markers from reasons.
|
||||
|
||||
Each entry's "reason" field becomes {REASON}_FINISH marker.
|
||||
Example: {"file": "...", "reason": "TypeCheck"} -> "TYPECHECK_FINISH"
|
||||
"""
|
||||
check_jsonl_path = os.path.join(repo_root, task_dir, "check.jsonl")
|
||||
markers = []
|
||||
|
||||
if not os.path.exists(check_jsonl_path):
|
||||
# Fallback: if no check.jsonl, use default marker
|
||||
return ["ALL_CHECKS_FINISH"]
|
||||
|
||||
try:
|
||||
with open(check_jsonl_path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
item = json.loads(line)
|
||||
reason = item.get("reason", "")
|
||||
if reason:
|
||||
# Convert to uppercase and add _FINISH suffix
|
||||
marker = f"{reason.upper().replace(' ', '_')}_FINISH"
|
||||
if marker not in markers:
|
||||
markers.append(marker)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# If no markers found, use default
|
||||
if not markers:
|
||||
markers = ["ALL_CHECKS_FINISH"]
|
||||
|
||||
return markers
|
||||
|
||||
|
||||
def load_state(repo_root: str) -> dict:
|
||||
"""Load Ralph Loop state from file"""
|
||||
state_path = os.path.join(repo_root, STATE_FILE)
|
||||
if not os.path.exists(state_path):
|
||||
return {"task": None, "iteration": 0, "started_at": None}
|
||||
|
||||
try:
|
||||
with open(state_path, "r", encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
except Exception:
|
||||
return {"task": None, "iteration": 0, "started_at": None}
|
||||
|
||||
|
||||
def save_state(repo_root: str, state: dict) -> None:
|
||||
"""Save Ralph Loop state to file"""
|
||||
state_path = os.path.join(repo_root, STATE_FILE)
|
||||
try:
|
||||
# Ensure directory exists
|
||||
os.makedirs(os.path.dirname(state_path), exist_ok=True)
|
||||
with open(state_path, "w", encoding="utf-8") as f:
|
||||
json.dump(state, f, indent=2, ensure_ascii=False)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def check_completion(agent_output: str, markers: list[str]) -> tuple[bool, list[str]]:
|
||||
"""
|
||||
Check if all completion markers are present in agent output.
|
||||
|
||||
Returns:
|
||||
(all_complete, missing_markers)
|
||||
"""
|
||||
missing = []
|
||||
for marker in markers:
|
||||
if marker not in agent_output:
|
||||
missing.append(marker)
|
||||
|
||||
return len(missing) == 0, missing
|
||||
|
||||
|
||||
def main():
|
||||
try:
|
||||
input_data = json.load(sys.stdin)
|
||||
except json.JSONDecodeError:
|
||||
# If can't parse input, allow stop
|
||||
sys.exit(0)
|
||||
|
||||
# Get event info
|
||||
hook_event = input_data.get("hook_event_name", "")
|
||||
|
||||
# Only handle SubagentStop event
|
||||
if hook_event != "SubagentStop":
|
||||
sys.exit(0)
|
||||
|
||||
# Get subagent info
|
||||
# Field names per Claude Code SubagentStop event schema:
|
||||
# agent_type, last_assistant_message, agent_id, agent_transcript_path, cwd
|
||||
# The event does NOT carry a `prompt` field, so finish-phase detection
|
||||
# based on a `[finish]` marker in the user prompt is no longer possible
|
||||
# here; finish-phase skip logic should be reintroduced via task.json
|
||||
# state (e.g. current_phase) in a follow-up.
|
||||
agent_type = input_data.get("agent_type", "")
|
||||
last_assistant_message = input_data.get("last_assistant_message", "")
|
||||
cwd = input_data.get("cwd", os.getcwd())
|
||||
|
||||
# Only control check agent
|
||||
if agent_type != TARGET_AGENT:
|
||||
sys.exit(0)
|
||||
|
||||
# Find repo root
|
||||
repo_root = find_repo_root(cwd)
|
||||
if not repo_root:
|
||||
sys.exit(0)
|
||||
|
||||
# Get current task
|
||||
task_dir = get_current_task(repo_root)
|
||||
if not task_dir:
|
||||
sys.exit(0)
|
||||
|
||||
# Load state
|
||||
state = load_state(repo_root)
|
||||
|
||||
# Reset state if task changed or state is too old
|
||||
should_reset = False
|
||||
if state.get("task") != task_dir:
|
||||
should_reset = True
|
||||
elif state.get("started_at"):
|
||||
try:
|
||||
started = datetime.fromisoformat(state["started_at"])
|
||||
if (datetime.now() - started).total_seconds() > STATE_TIMEOUT_MINUTES * 60:
|
||||
should_reset = True
|
||||
except (ValueError, TypeError):
|
||||
should_reset = True
|
||||
|
||||
if should_reset:
|
||||
state = {
|
||||
"task": task_dir,
|
||||
"iteration": 0,
|
||||
"started_at": datetime.now().isoformat(),
|
||||
}
|
||||
|
||||
# Increment iteration
|
||||
state["iteration"] = state.get("iteration", 0) + 1
|
||||
current_iteration = state["iteration"]
|
||||
|
||||
# Save state
|
||||
save_state(repo_root, state)
|
||||
|
||||
# Safety check: max iterations
|
||||
if current_iteration >= MAX_ITERATIONS:
|
||||
# Allow stop, reset state for next run
|
||||
state["iteration"] = 0
|
||||
save_state(repo_root, state)
|
||||
output = {
|
||||
"decision": "allow",
|
||||
"reason": f"Max iterations ({MAX_ITERATIONS}) reached. Stopping to prevent infinite loop.",
|
||||
}
|
||||
print(json.dumps(output, ensure_ascii=False))
|
||||
sys.exit(0)
|
||||
|
||||
# Check if verify commands are configured
|
||||
verify_commands = get_verify_commands(repo_root)
|
||||
|
||||
if verify_commands:
|
||||
# Use programmatic verification
|
||||
passed, message = run_verify_commands(repo_root, verify_commands)
|
||||
|
||||
if passed:
|
||||
# All verify commands passed, allow stop
|
||||
state["iteration"] = 0
|
||||
save_state(repo_root, state)
|
||||
output = {
|
||||
"decision": "allow",
|
||||
"reason": "All verify commands passed. Check phase complete.",
|
||||
}
|
||||
print(json.dumps(output, ensure_ascii=False))
|
||||
sys.exit(0)
|
||||
else:
|
||||
# Verification failed, block stop
|
||||
output = {
|
||||
"decision": "block",
|
||||
"reason": f"Iteration {current_iteration}/{MAX_ITERATIONS}. Verification failed:\n{message}\n\nPlease fix the issues and try again.",
|
||||
}
|
||||
print(json.dumps(output, ensure_ascii=False))
|
||||
sys.exit(0)
|
||||
else:
|
||||
# No verify commands, fall back to completion markers
|
||||
markers = get_completion_markers(repo_root, task_dir)
|
||||
all_complete, missing = check_completion(last_assistant_message, markers)
|
||||
|
||||
if all_complete:
|
||||
# All checks complete, allow stop
|
||||
state["iteration"] = 0
|
||||
save_state(repo_root, state)
|
||||
output = {
|
||||
"decision": "allow",
|
||||
"reason": "All completion markers found. Check phase complete.",
|
||||
}
|
||||
print(json.dumps(output, ensure_ascii=False))
|
||||
sys.exit(0)
|
||||
else:
|
||||
# Missing markers, block stop and continue
|
||||
output = {
|
||||
"decision": "block",
|
||||
"reason": f"""Iteration {current_iteration}/{MAX_ITERATIONS}. Missing completion markers: {", ".join(missing)}.
|
||||
|
||||
IMPORTANT: You must ACTUALLY run the checks, not just output the markers.
|
||||
- Did you run lint? What was the output?
|
||||
- Did you run typecheck? What was the output?
|
||||
- Did they actually pass with zero errors?
|
||||
|
||||
Only output a marker (e.g., LINT_FINISH) AFTER:
|
||||
1. You have executed the corresponding command
|
||||
2. The command completed with zero errors
|
||||
3. You have shown the command output in your response
|
||||
|
||||
Do NOT output markers just to escape the loop. The loop exists to ensure quality.""",
|
||||
}
|
||||
print(json.dumps(output, ensure_ascii=False))
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,414 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Session Start Hook - Inject structured context
|
||||
"""
|
||||
|
||||
# IMPORTANT: Suppress all warnings FIRST
|
||||
import warnings
|
||||
warnings.filterwarnings("ignore")
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from io import StringIO
|
||||
from pathlib import Path
|
||||
|
||||
# IMPORTANT: Force stdout to use UTF-8 on Windows
|
||||
# This fixes UnicodeEncodeError when outputting non-ASCII characters
|
||||
if sys.platform == "win32":
|
||||
import io as _io
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
sys.stdout.reconfigure(encoding="utf-8", errors="replace") # type: ignore[union-attr]
|
||||
elif hasattr(sys.stdout, "detach"):
|
||||
sys.stdout = _io.TextIOWrapper(sys.stdout.detach(), encoding="utf-8", errors="replace") # type: ignore[union-attr]
|
||||
|
||||
|
||||
def should_skip_injection() -> bool:
|
||||
return (
|
||||
os.environ.get("CLAUDE_NON_INTERACTIVE") == "1"
|
||||
or os.environ.get("OPENCODE_NON_INTERACTIVE") == "1"
|
||||
)
|
||||
|
||||
|
||||
def read_file(path: Path, fallback: str = "") -> str:
|
||||
try:
|
||||
return path.read_text(encoding="utf-8")
|
||||
except (FileNotFoundError, PermissionError):
|
||||
return fallback
|
||||
|
||||
|
||||
def run_script(script_path: Path) -> str:
|
||||
try:
|
||||
if script_path.suffix == ".py":
|
||||
# Add PYTHONIOENCODING to force UTF-8 in subprocess
|
||||
env = os.environ.copy()
|
||||
env["PYTHONIOENCODING"] = "utf-8"
|
||||
cmd = [sys.executable, "-W", "ignore", str(script_path)]
|
||||
else:
|
||||
env = os.environ
|
||||
cmd = [str(script_path)]
|
||||
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
timeout=5,
|
||||
cwd=script_path.parent.parent.parent,
|
||||
env=env,
|
||||
)
|
||||
return result.stdout if result.returncode == 0 else "No context available"
|
||||
except (subprocess.TimeoutExpired, FileNotFoundError, PermissionError):
|
||||
return "No context available"
|
||||
|
||||
|
||||
def _normalize_task_ref(task_ref: str) -> str:
|
||||
normalized = task_ref.strip()
|
||||
if not normalized:
|
||||
return ""
|
||||
|
||||
path_obj = Path(normalized)
|
||||
if path_obj.is_absolute():
|
||||
return str(path_obj)
|
||||
|
||||
normalized = normalized.replace("\\", "/")
|
||||
while normalized.startswith("./"):
|
||||
normalized = normalized[2:]
|
||||
|
||||
if normalized.startswith("tasks/"):
|
||||
return f".trellis/{normalized}"
|
||||
|
||||
return normalized
|
||||
|
||||
|
||||
def _resolve_task_dir(trellis_dir: Path, task_ref: str) -> Path:
|
||||
normalized = _normalize_task_ref(task_ref)
|
||||
path_obj = Path(normalized)
|
||||
if path_obj.is_absolute():
|
||||
return path_obj
|
||||
if normalized.startswith(".trellis/"):
|
||||
return trellis_dir.parent / path_obj
|
||||
return trellis_dir / "tasks" / path_obj
|
||||
|
||||
|
||||
def _get_task_status(trellis_dir: Path) -> str:
|
||||
"""Check current task status and return structured status string."""
|
||||
current_task_file = trellis_dir / ".current-task"
|
||||
if not current_task_file.is_file():
|
||||
return "Status: NO ACTIVE TASK\nNext: Describe what you want to work on"
|
||||
|
||||
task_ref = _normalize_task_ref(current_task_file.read_text(encoding="utf-8").strip())
|
||||
if not task_ref:
|
||||
return "Status: NO ACTIVE TASK\nNext: Describe what you want to work on"
|
||||
|
||||
# Resolve task directory
|
||||
task_dir = _resolve_task_dir(trellis_dir, task_ref)
|
||||
if not task_dir.is_dir():
|
||||
return f"Status: STALE POINTER\nTask: {task_ref}\nNext: Task directory not found. Run: python3 ./.trellis/scripts/task.py finish"
|
||||
|
||||
# Read task.json
|
||||
task_json_path = task_dir / "task.json"
|
||||
task_data = {}
|
||||
if task_json_path.is_file():
|
||||
try:
|
||||
task_data = json.loads(task_json_path.read_text(encoding="utf-8"))
|
||||
except (json.JSONDecodeError, PermissionError):
|
||||
pass
|
||||
|
||||
task_title = task_data.get("title", task_ref)
|
||||
task_status = task_data.get("status", "unknown")
|
||||
|
||||
if task_status == "completed":
|
||||
return f"Status: COMPLETED\nTask: {task_title}\nNext: Archive with `python3 ./.trellis/scripts/task.py archive {task_dir.name}` or start a new task"
|
||||
|
||||
# Check if context is configured (jsonl files exist and non-empty)
|
||||
has_context = False
|
||||
for jsonl_name in ("implement.jsonl", "check.jsonl", "spec.jsonl"):
|
||||
jsonl_path = task_dir / jsonl_name
|
||||
if jsonl_path.is_file() and jsonl_path.stat().st_size > 0:
|
||||
has_context = True
|
||||
break
|
||||
|
||||
has_prd = (task_dir / "prd.md").is_file()
|
||||
|
||||
if not has_prd:
|
||||
return f"Status: NOT READY\nTask: {task_title}\nMissing: prd.md not created\nNext: Write PRD, then research → init-context → start"
|
||||
|
||||
if not has_context:
|
||||
return f"Status: NOT READY\nTask: {task_title}\nMissing: Context not configured (no jsonl files)\nNext: Complete Phase 2 (research → init-context → start) before implementing"
|
||||
|
||||
return f"Status: READY\nTask: {task_title}\nNext: Continue with implement or check"
|
||||
|
||||
|
||||
def _load_trellis_config(trellis_dir: Path) -> tuple:
|
||||
"""Load Trellis config for session-start decisions.
|
||||
|
||||
Returns:
|
||||
(is_mono, packages_dict, spec_scope, task_pkg, default_pkg)
|
||||
"""
|
||||
scripts_dir = trellis_dir / "scripts"
|
||||
if str(scripts_dir) not in sys.path:
|
||||
sys.path.insert(0, str(scripts_dir))
|
||||
|
||||
try:
|
||||
from common.config import get_default_package, get_packages, get_spec_scope, is_monorepo # type: ignore[import-not-found]
|
||||
from common.paths import get_current_task # type: ignore[import-not-found]
|
||||
|
||||
repo_root = trellis_dir.parent
|
||||
is_mono = is_monorepo(repo_root)
|
||||
packages = get_packages(repo_root) or {}
|
||||
scope = get_spec_scope(repo_root)
|
||||
|
||||
# Get active task's package
|
||||
task_pkg = None
|
||||
current = get_current_task(repo_root)
|
||||
if current:
|
||||
task_json = repo_root / current / "task.json"
|
||||
if task_json.is_file():
|
||||
try:
|
||||
data = json.loads(task_json.read_text(encoding="utf-8"))
|
||||
if isinstance(data, dict):
|
||||
tp = data.get("package")
|
||||
if isinstance(tp, str) and tp:
|
||||
task_pkg = tp
|
||||
except (json.JSONDecodeError, OSError):
|
||||
pass
|
||||
|
||||
default_pkg = get_default_package(repo_root)
|
||||
return is_mono, packages, scope, task_pkg, default_pkg
|
||||
except Exception:
|
||||
return False, {}, None, None, None
|
||||
|
||||
|
||||
def _check_legacy_spec(trellis_dir: Path, is_mono: bool, packages: dict) -> str | None:
|
||||
"""Check for legacy spec directory structure in monorepo.
|
||||
|
||||
Returns warning message if legacy structure detected, None otherwise.
|
||||
"""
|
||||
if not is_mono or not packages:
|
||||
return None
|
||||
|
||||
spec_dir = trellis_dir / "spec"
|
||||
if not spec_dir.is_dir():
|
||||
return None
|
||||
|
||||
# Check for legacy flat spec dirs (spec/backend/, spec/frontend/ with index.md)
|
||||
has_legacy = False
|
||||
for legacy_name in ("backend", "frontend"):
|
||||
legacy_dir = spec_dir / legacy_name
|
||||
if legacy_dir.is_dir() and (legacy_dir / "index.md").is_file():
|
||||
has_legacy = True
|
||||
break
|
||||
|
||||
if not has_legacy:
|
||||
return None
|
||||
|
||||
# Check which packages are missing spec/<pkg>/ directory
|
||||
missing = [
|
||||
name for name in sorted(packages.keys())
|
||||
if not (spec_dir / name).is_dir()
|
||||
]
|
||||
|
||||
if not missing:
|
||||
return None # All packages have spec dirs
|
||||
|
||||
if len(missing) == len(packages):
|
||||
return (
|
||||
f"[!] Legacy spec structure detected: found `spec/backend/` or `spec/frontend/` "
|
||||
f"but no package-scoped `spec/<package>/` directories.\n"
|
||||
f"Monorepo packages: {', '.join(sorted(packages.keys()))}\n"
|
||||
f"Please reorganize: `spec/backend/` -> `spec/<package>/backend/`"
|
||||
)
|
||||
return (
|
||||
f"[!] Partial spec migration detected: packages {', '.join(missing)} "
|
||||
f"still missing `spec/<pkg>/` directory.\n"
|
||||
f"Please complete migration for all packages."
|
||||
)
|
||||
|
||||
|
||||
def _resolve_spec_scope(
|
||||
is_mono: bool,
|
||||
packages: dict,
|
||||
scope,
|
||||
task_pkg: str | None,
|
||||
default_pkg: str | None,
|
||||
) -> set | None:
|
||||
"""Resolve which packages should have their specs injected.
|
||||
|
||||
Returns:
|
||||
Set of package names to include, or None for full scan.
|
||||
"""
|
||||
if not is_mono or not packages:
|
||||
return None # Single-repo: full scan
|
||||
|
||||
if scope is None:
|
||||
return None # No scope configured: full scan
|
||||
|
||||
if isinstance(scope, str) and scope == "active_task":
|
||||
if task_pkg and task_pkg in packages:
|
||||
return {task_pkg}
|
||||
if default_pkg and default_pkg in packages:
|
||||
return {default_pkg}
|
||||
return None # Fallback to full scan
|
||||
|
||||
if isinstance(scope, list):
|
||||
valid = set()
|
||||
for entry in scope:
|
||||
if entry in packages:
|
||||
valid.add(entry)
|
||||
else:
|
||||
print(
|
||||
f"Warning: spec_scope contains unknown package: {entry}, ignoring",
|
||||
file=sys.stderr,
|
||||
)
|
||||
|
||||
if valid:
|
||||
# Warn if active task is out of scope
|
||||
if task_pkg and task_pkg not in valid:
|
||||
print(
|
||||
f"Warning: active task package '{task_pkg}' is out of configured spec_scope",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return valid
|
||||
|
||||
# All entries invalid: fallback chain
|
||||
print(
|
||||
"Warning: all spec_scope entries invalid, falling back to task/default/full",
|
||||
file=sys.stderr,
|
||||
)
|
||||
if task_pkg and task_pkg in packages:
|
||||
return {task_pkg}
|
||||
if default_pkg and default_pkg in packages:
|
||||
return {default_pkg}
|
||||
return None # Full scan
|
||||
|
||||
return None # Unknown scope type: full scan
|
||||
|
||||
|
||||
def _build_workflow_toc(workflow_path: Path) -> str:
|
||||
"""Build a compact section index for workflow.md (lazy-load the full file on demand).
|
||||
|
||||
Replaces full-file injection to keep additionalContext payload small.
|
||||
The full file is accessible via: Read tool on .trellis/workflow.md
|
||||
"""
|
||||
content = read_file(workflow_path)
|
||||
if not content:
|
||||
return "No workflow.md found"
|
||||
|
||||
toc_lines = [
|
||||
"# Development Workflow — Section Index",
|
||||
"Full guide: .trellis/workflow.md (read on demand)",
|
||||
"",
|
||||
]
|
||||
for line in content.splitlines():
|
||||
if line.startswith("## "):
|
||||
toc_lines.append(line)
|
||||
|
||||
toc_lines += [
|
||||
"",
|
||||
"To read a section: use the Read tool on .trellis/workflow.md",
|
||||
]
|
||||
return "\n".join(toc_lines)
|
||||
|
||||
|
||||
def main():
|
||||
if should_skip_injection():
|
||||
sys.exit(0)
|
||||
|
||||
project_dir = Path(os.environ.get("CLAUDE_PROJECT_DIR", ".")).resolve()
|
||||
trellis_dir = project_dir / ".trellis"
|
||||
|
||||
# Load config for scope filtering and legacy detection
|
||||
is_mono, packages, scope_config, task_pkg, default_pkg = _load_trellis_config(trellis_dir)
|
||||
allowed_pkgs = _resolve_spec_scope(is_mono, packages, scope_config, task_pkg, default_pkg)
|
||||
|
||||
output = StringIO()
|
||||
|
||||
output.write("""<session-context>
|
||||
You are starting a new session in a Trellis-managed project.
|
||||
Read and follow all instructions below carefully.
|
||||
</session-context>
|
||||
|
||||
""")
|
||||
|
||||
# Legacy migration warning
|
||||
legacy_warning = _check_legacy_spec(trellis_dir, is_mono, packages)
|
||||
if legacy_warning:
|
||||
output.write(f"<migration-warning>\n{legacy_warning}\n</migration-warning>\n\n")
|
||||
|
||||
output.write("<current-state>\n")
|
||||
context_script = trellis_dir / "scripts" / "get_context.py"
|
||||
output.write(run_script(context_script))
|
||||
output.write("\n</current-state>\n\n")
|
||||
|
||||
output.write("<workflow>\n")
|
||||
output.write(_build_workflow_toc(trellis_dir / "workflow.md"))
|
||||
output.write("\n</workflow>\n\n")
|
||||
|
||||
output.write("<guidelines>\n")
|
||||
output.write("**Note**: The guidelines below are index files — they list available guideline documents and their locations.\n")
|
||||
output.write("During actual development, you MUST read the specific guideline files listed in each index's Pre-Development Checklist.\n\n")
|
||||
|
||||
spec_dir = trellis_dir / "spec"
|
||||
if spec_dir.is_dir():
|
||||
for sub in sorted(spec_dir.iterdir()):
|
||||
if not sub.is_dir() or sub.name.startswith("."):
|
||||
continue
|
||||
|
||||
# Always include guides/ regardless of scope
|
||||
if sub.name == "guides":
|
||||
index_file = sub / "index.md"
|
||||
if index_file.is_file():
|
||||
output.write(f"## {sub.name}\n")
|
||||
output.write(read_file(index_file))
|
||||
output.write("\n\n")
|
||||
continue
|
||||
|
||||
index_file = sub / "index.md"
|
||||
if index_file.is_file():
|
||||
# Flat spec dir (single-repo layer like spec/backend/)
|
||||
output.write(f"## {sub.name}\n")
|
||||
output.write(read_file(index_file))
|
||||
output.write("\n\n")
|
||||
else:
|
||||
# Nested package dirs (monorepo: spec/<pkg>/<layer>/index.md)
|
||||
# Apply scope filter
|
||||
if allowed_pkgs is not None and sub.name not in allowed_pkgs:
|
||||
continue
|
||||
for nested in sorted(sub.iterdir()):
|
||||
if not nested.is_dir():
|
||||
continue
|
||||
nested_index = nested / "index.md"
|
||||
if nested_index.is_file():
|
||||
output.write(f"## {sub.name}/{nested.name}\n")
|
||||
output.write(read_file(nested_index))
|
||||
output.write("\n\n")
|
||||
|
||||
output.write("</guidelines>\n\n")
|
||||
|
||||
# Check task status and inject structured tag
|
||||
task_status = _get_task_status(trellis_dir)
|
||||
output.write(f"<task-status>\n{task_status}\n</task-status>\n\n")
|
||||
|
||||
output.write("""<ready>
|
||||
Context loaded. Workflow index, project state, and guidelines are already injected above — do NOT re-read them.
|
||||
Wait for the user's first message, then handle it following the workflow guide.
|
||||
If there is an active task, ask whether to continue it.
|
||||
</ready>""")
|
||||
|
||||
result = {
|
||||
"hookSpecificOutput": {
|
||||
"hookEventName": "SessionStart",
|
||||
"additionalContext": output.getvalue(),
|
||||
}
|
||||
}
|
||||
|
||||
# Output JSON - stdout is already configured for UTF-8
|
||||
print(json.dumps(result, ensure_ascii=False), flush=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,218 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Trellis StatusLine — project-level status display for Claude Code.
|
||||
|
||||
Reads Claude Code session JSON from stdin + Trellis task data from filesystem.
|
||||
Outputs 1-2 lines:
|
||||
With active task: [P1] Task title (status) + info line
|
||||
Without task: info line only
|
||||
Info line: model · ctx% · branch · duration · developer · tasks · rate limits
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Fix: Windows Python defaults to GBK encoding, which corrupts UTF-8
|
||||
# characters like the middle dot (·). Wrap stdout/stderr with UTF-8.
|
||||
if sys.platform == "win32":
|
||||
sys.stdout = io.TextIOWrapper(sys.stdout.detach(), encoding="utf-8")
|
||||
sys.stderr = io.TextIOWrapper(sys.stderr.detach(), encoding="utf-8")
|
||||
|
||||
|
||||
def _read_text(path: Path) -> str:
|
||||
try:
|
||||
return path.read_text(encoding="utf-8").strip()
|
||||
except (FileNotFoundError, PermissionError, OSError):
|
||||
return ""
|
||||
|
||||
|
||||
def _read_json(path: Path) -> dict:
|
||||
text = _read_text(path)
|
||||
if not text:
|
||||
return {}
|
||||
try:
|
||||
return json.loads(text)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
return {}
|
||||
|
||||
|
||||
def _normalize_task_ref(task_ref: str) -> str:
|
||||
normalized = task_ref.strip()
|
||||
if not normalized:
|
||||
return ""
|
||||
|
||||
path_obj = Path(normalized)
|
||||
if path_obj.is_absolute():
|
||||
return str(path_obj)
|
||||
|
||||
normalized = normalized.replace("\\", "/")
|
||||
while normalized.startswith("./"):
|
||||
normalized = normalized[2:]
|
||||
|
||||
if normalized.startswith("tasks/"):
|
||||
return f".trellis/{normalized}"
|
||||
|
||||
return normalized
|
||||
|
||||
|
||||
def _resolve_task_dir(trellis_dir: Path, task_ref: str) -> Path:
|
||||
normalized = _normalize_task_ref(task_ref)
|
||||
path_obj = Path(normalized)
|
||||
if path_obj.is_absolute():
|
||||
return path_obj
|
||||
if normalized.startswith(".trellis/"):
|
||||
return trellis_dir.parent / path_obj
|
||||
return trellis_dir / "tasks" / path_obj
|
||||
|
||||
|
||||
def _find_trellis_dir() -> Path | None:
|
||||
"""Walk up from cwd to find .trellis/ directory."""
|
||||
current = Path.cwd()
|
||||
for parent in [current, *current.parents]:
|
||||
candidate = parent / ".trellis"
|
||||
if candidate.is_dir():
|
||||
return candidate
|
||||
return None
|
||||
|
||||
|
||||
def _get_current_task(trellis_dir: Path) -> dict | None:
|
||||
"""Load current task info. Returns dict with title/status/priority or None."""
|
||||
task_ref = _normalize_task_ref(_read_text(trellis_dir / ".current-task"))
|
||||
if not task_ref:
|
||||
return None
|
||||
|
||||
# Resolve task directory
|
||||
task_path = _resolve_task_dir(trellis_dir, task_ref)
|
||||
task_data = _read_json(task_path / "task.json")
|
||||
if not task_data:
|
||||
return None
|
||||
|
||||
return {
|
||||
"title": task_data.get("title") or task_data.get("name") or "unknown",
|
||||
"status": task_data.get("status", "unknown"),
|
||||
"priority": task_data.get("priority", "P2"),
|
||||
}
|
||||
|
||||
|
||||
def _count_active_tasks(trellis_dir: Path) -> int:
|
||||
"""Count non-archived task directories with valid task.json."""
|
||||
tasks_dir = trellis_dir / "tasks"
|
||||
if not tasks_dir.is_dir():
|
||||
return 0
|
||||
count = 0
|
||||
for d in tasks_dir.iterdir():
|
||||
if d.is_dir() and d.name != "archive" and (d / "task.json").is_file():
|
||||
count += 1
|
||||
return count
|
||||
|
||||
|
||||
def _get_developer(trellis_dir: Path) -> str:
|
||||
content = _read_text(trellis_dir / ".developer")
|
||||
if not content:
|
||||
return "unknown"
|
||||
for line in content.splitlines():
|
||||
if line.startswith("name="):
|
||||
return line[5:].strip()
|
||||
return content.splitlines()[0].strip() or "unknown"
|
||||
|
||||
|
||||
def _get_git_branch() -> str:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "branch", "--show-current"],
|
||||
capture_output=True, text=True, timeout=3,
|
||||
)
|
||||
return result.stdout.strip() if result.returncode == 0 else ""
|
||||
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||
return ""
|
||||
|
||||
|
||||
def _format_ctx_size(size: int) -> str:
|
||||
if size >= 1_000_000:
|
||||
return f"{size // 1_000_000}M"
|
||||
if size >= 1_000:
|
||||
return f"{size // 1_000}K"
|
||||
return str(size)
|
||||
|
||||
|
||||
def _format_duration(ms: int) -> str:
|
||||
secs = ms // 1000
|
||||
hours, remainder = divmod(secs, 3600)
|
||||
mins = remainder // 60
|
||||
if hours > 0:
|
||||
return f"{hours}h{mins}m"
|
||||
return f"{mins}m"
|
||||
|
||||
|
||||
def main() -> None:
|
||||
# Read Claude Code session JSON from stdin
|
||||
try:
|
||||
cc_data = json.loads(sys.stdin.read())
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
cc_data = {}
|
||||
|
||||
trellis_dir = _find_trellis_dir()
|
||||
SEP = " \033[90m·\033[0m "
|
||||
|
||||
# --- Trellis data ---
|
||||
task = _get_current_task(trellis_dir) if trellis_dir else None
|
||||
dev = _get_developer(trellis_dir) if trellis_dir else ""
|
||||
task_count = _count_active_tasks(trellis_dir) if trellis_dir else 0
|
||||
|
||||
# --- CC session data ---
|
||||
model = cc_data.get("model", {}).get("display_name", "?")
|
||||
ctx_pct = int(cc_data.get("context_window", {}).get("used_percentage") or 0)
|
||||
ctx_size = _format_ctx_size(cc_data.get("context_window", {}).get("context_window_size") or 0)
|
||||
duration = _format_duration(cc_data.get("cost", {}).get("total_duration_ms") or 0)
|
||||
branch = _get_git_branch()
|
||||
|
||||
# Avoid "Opus 4.6 (1M context) (1M)"
|
||||
if re.search(r"\d+[KMG]\b", model, re.IGNORECASE):
|
||||
model_label = model
|
||||
else:
|
||||
model_label = f"{model} ({ctx_size})"
|
||||
|
||||
# Context % with color
|
||||
if ctx_pct >= 90:
|
||||
ctx_color = "\033[31m"
|
||||
elif ctx_pct >= 70:
|
||||
ctx_color = "\033[33m"
|
||||
else:
|
||||
ctx_color = "\033[32m"
|
||||
|
||||
# Build info line: model · ctx · branch · duration · dev · tasks [· rate limits]
|
||||
parts = [
|
||||
model_label,
|
||||
f"ctx {ctx_color}{ctx_pct}%\033[0m",
|
||||
]
|
||||
if branch:
|
||||
parts.append(f"\033[35m{branch}\033[0m")
|
||||
parts.append(duration)
|
||||
if dev:
|
||||
parts.append(f"\033[32m{dev}\033[0m")
|
||||
if task_count:
|
||||
parts.append(f"{task_count} task(s)")
|
||||
|
||||
five_hr = cc_data.get("rate_limits", {}).get("five_hour", {}).get("used_percentage")
|
||||
if five_hr is not None:
|
||||
parts.append(f"5h {int(five_hr)}%")
|
||||
seven_day = cc_data.get("rate_limits", {}).get("seven_day", {}).get("used_percentage")
|
||||
if seven_day is not None:
|
||||
parts.append(f"7d {int(seven_day)}%")
|
||||
|
||||
info_line = SEP.join(parts)
|
||||
|
||||
# Output: task line (only if active) + info line
|
||||
if task:
|
||||
print(f"\033[36m[{task['priority']}]\033[0m {task['title']} \033[33m({task['status']})\033[0m")
|
||||
print(info_line)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user