TalexDreamSoul/teatea-pension
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs: add Trellis planning and project specs

Browse Source
This commit is contained in:
admin
2026-07-01 06:27:55 -07:00
parent 7f227c5f2a
commit aafd9caaac
349 changed files with 55801 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

183
.cursor/hooks/inject-shell-session-context.py Normal file
View File

@@ -0,0 +1,183 @@
#!/usr/bin/env python3
"""Cursor beforeShellExecution hook: bridge conversation identity to task.py.
Cursor's shell command environment does not inherit SessionStart data. This
hook writes a short-lived runtime ticket before Cursor runs a shell command
that calls `task.py start/current/finish`. The task script then consumes the
ticket only when it has no native session environment.
"""
from __future__ import annotations
import hashlib
import json
import os
import shlex
import sys
import time
from pathlib import Path
from typing import Any
DIR_WORKFLOW = ".trellis"
DIR_RUNTIME = ".runtime"
DIR_CURSOR_SHELL = "cursor-shell"
SESSION_SUBCOMMANDS = {"start", "current", "finish"}
TICKET_TTL_SECONDS = 30
CONTEXT_IDENTITY_KEYS = (
"session_id",
"sessionId",
"sessionID",
"conversation_id",
"conversationId",
"conversationID",
"transcript_path",
"transcriptPath",
"transcript",
)
def _string_value(value: Any) -> str | None:
if isinstance(value, str):
stripped = value.strip()
return stripped or None
return None
def _find_trellis_root(start: Path) -> Path | None:
current = start.resolve()
while True:
if (current / DIR_WORKFLOW).is_dir():
return current
if current == current.parent:
return None
current = current.parent
def _runtime_ticket_dir(root: Path) -> Path:
return root / DIR_WORKFLOW / DIR_RUNTIME / DIR_CURSOR_SHELL
def _load_active_task_resolver(root: Path):
scripts_dir = root / DIR_WORKFLOW / "scripts"
if str(scripts_dir) not in sys.path:
sys.path.insert(0, str(scripts_dir))
from common.active_task import resolve_context_key # type: ignore[import-not-found]
return resolve_context_key
def _extract_task_subcommands(command: str) -> list[dict[str, str]]:
try:
tokens = shlex.split(command, posix=os.name != "nt")
except ValueError:
return []
subcommands: list[dict[str, str]] = []
for index, token in enumerate(tokens[:-1]):
if Path(token.strip("\"'")).name != "task.py":
continue
name = tokens[index + 1]
if name not in SESSION_SUBCOMMANDS:
continue
item = {"name": name}
if name == "start" and index + 2 < len(tokens):
item["task_ref"] = tokens[index + 2]
subcommands.append(item)
return subcommands
def _cleanup_expired_tickets(ticket_dir: Path, now: float) -> None:
if not ticket_dir.is_dir():
return
for ticket_path in ticket_dir.glob("*.json"):
try:
data = json.loads(ticket_path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
continue
expires_at = data.get("expires_at_epoch")
if isinstance(expires_at, (int, float)) and expires_at < now:
try:
ticket_path.unlink()
except OSError:
pass
def _has_context_identity(hook_input: dict[str, Any]) -> bool:
return any(_string_value(hook_input.get(key)) for key in CONTEXT_IDENTITY_KEYS)
def _write_ticket(
root: Path,
hook_input: dict[str, Any],
context_key: str,
subcommands: list[dict[str, str]],
) -> None:
now = time.time()
ticket_dir = _runtime_ticket_dir(root)
ticket_dir.mkdir(parents=True, exist_ok=True)
_cleanup_expired_tickets(ticket_dir, now)
command = _string_value(hook_input.get("command")) or ""
digest = hashlib.sha256(
f"{context_key}\0{command}\0{now}".encode("utf-8"),
).hexdigest()[:16]
ticket_path = ticket_dir / f"{int(now * 1000)}-{digest}.json"
payload = {
"platform": "cursor",
"context_key": context_key,
"conversation_id": _string_value(hook_input.get("conversation_id")),
"session_id": _string_value(hook_input.get("session_id")),
"generation_id": _string_value(hook_input.get("generation_id")),
"cwd": _string_value(hook_input.get("cwd")),
"command": command,
"subcommands": subcommands,
"created_at_epoch": now,
"expires_at_epoch": now + TICKET_TTL_SECONDS,
}
ticket_path.write_text(
json.dumps(payload, indent=2, ensure_ascii=False) + "\n",
encoding="utf-8",
)
def main() -> int:
if os.environ.get("TRELLIS_HOOKS") == "0" or os.environ.get("TRELLIS_DISABLE_HOOKS") == "1":
return 0
try:
hook_input = json.loads(sys.stdin.read())
except (json.JSONDecodeError, ValueError):
hook_input = {}
if not isinstance(hook_input, dict):
hook_input = {}
command = _string_value(hook_input.get("command")) or ""
subcommands = _extract_task_subcommands(command)
if not subcommands:
return 0
cwd = Path(_string_value(hook_input.get("cwd")) or os.getcwd())
root = _find_trellis_root(cwd)
if root is None:
return 0
if not _has_context_identity(hook_input):
return 0
resolve_context_key = _load_active_task_resolver(root)
context_key = resolve_context_key(hook_input, platform="cursor")
if not context_key:
return 0
try:
_write_ticket(root, hook_input, context_key, subcommands)
except OSError:
return 0
print(json.dumps({"permission": "allow"}, ensure_ascii=False))
return 0
if __name__ == "__main__":
sys.exit(main())

771
.cursor/hooks/inject-subagent-context.py Normal file
View File

@@ -0,0 +1,771 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Multi-Platform Sub-Agent Context Injection Hook
Injects task-specific context when sub-agents (implement, check, research) are spawned.
Core Design Philosophy:
- 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 active task resolver points to task directory
- implement.jsonl - Implement agent dedicated context
- check.jsonl - Check agent dedicated context
- prd.md - Requirements document
- design.md - Technical design for complex tasks
- implement.md - Execution plan for complex tasks
- codex-review-output.txt - Code Review results
"""
from __future__ import annotations
# IMPORTANT: Suppress all warnings FIRST
import warnings
warnings.filterwarnings("ignore")
import json
import os
import sys
from pathlib import Path
from typing import Any
# IMPORTANT: Force stdout to use UTF-8 on Windows
# This fixes UnicodeEncodeError when outputting non-ASCII characters
if sys.platform.startswith("win"):
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_SPEC = "spec"
FILE_TASK_JSON = "task.json"
# =============================================================================
# Subagent Constants (change here to rename subagent types)
# =============================================================================
AGENT_IMPLEMENT = "trellis-implement"
AGENT_CHECK = "trellis-check"
AGENT_RESEARCH = "trellis-research"
# Agents that require a task directory
AGENTS_REQUIRE_TASK = (AGENT_IMPLEMENT, AGENT_CHECK)
# All supported agents
AGENTS_ALL = (AGENT_IMPLEMENT, AGENT_CHECK, 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 _detect_platform(input_data: dict) -> str | None:
if isinstance(input_data.get("cursor_version"), str):
return "cursor"
env_map = {
"CLAUDE_PROJECT_DIR": "claude",
"CURSOR_PROJECT_DIR": "cursor",
"CODEBUDDY_PROJECT_DIR": "codebuddy",
"FACTORY_PROJECT_DIR": "droid",
"GEMINI_PROJECT_DIR": "gemini",
"QODER_PROJECT_DIR": "qoder",
"KIRO_PROJECT_DIR": "kiro",
"COPILOT_PROJECT_DIR": "copilot",
}
for env_name, platform in env_map.items():
if os.environ.get(env_name):
return platform
script_parts = set(Path(sys.argv[0]).parts)
if ".claude" in script_parts:
return "claude"
if ".cursor" in script_parts:
return "cursor"
if ".gemini" in script_parts:
return "gemini"
if ".qoder" in script_parts:
return "qoder"
if ".codebuddy" in script_parts:
return "codebuddy"
if ".factory" in script_parts:
return "droid"
if ".kiro" in script_parts:
return "kiro"
return None
def get_current_task(repo_root: str, input_data: dict) -> str | None:
"""Resolve current task directory through the unified active task resolver."""
scripts_dir = Path(repo_root) / DIR_WORKFLOW / "scripts"
if str(scripts_dir) not in sys.path:
sys.path.insert(0, str(scripts_dir))
try:
from common.active_task import resolve_active_task # type: ignore[import-not-found]
except Exception:
return None
active = resolve_active_task(
Path(repo_root),
input_data,
platform=_detect_platform(input_data),
)
return active.task_path
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": "..."}
{"_example": "..."} # seed row — skipped (no `file` field)
Rows without a ``file`` field (e.g. the self-describing seed line written
by ``task.py create`` before the agent has curated entries) are skipped
silently. If the resulting entry list is empty, a stderr warning is
emitted so the operator can debug missing context.
Returns:
[(path, content), ...]
"""
full_path = os.path.join(base_path, jsonl_path)
if not os.path.exists(full_path):
print(
f"[inject-subagent-context] WARN: {jsonl_path} not found — "
f"sub-agent will receive only task artifacts",
file=sys.stderr,
)
return []
results = []
saw_real_entry = False
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:
# Seed / comment row — skip silently
continue
saw_real_entry = True
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
if not saw_real_entry:
print(
f"[inject-subagent-context] WARN: {jsonl_path} has no curated "
f"entries (only seed / empty) — sub-agent will receive only "
f"task artifacts. See workflow.md planning artifact guidance.",
file=sys.stderr,
)
return results
def get_agent_context(repo_root: str, task_dir: str, agent_type: str) -> str:
"""
Get context from {agent_type}.jsonl for the specified agent.
Only reads implement.jsonl or check.jsonl (the two JSONL files the task system creates).
"""
context_parts = []
agent_jsonl = f"{task_dir}/{agent_type}.jsonl"
for file_path, content in read_jsonl_entries(repo_root, agent_jsonl):
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 (spec/research manifests)
2. prd.md (requirements)
3. design.md if present (technical design)
4. implement.md if present (execution plan)
"""
context_parts = []
# 1. Read implement.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 for complex tasks
design_content = read_file_content(repo_root, f"{task_dir}/design.md")
if design_content:
context_parts.append(
f"=== {task_dir}/design.md (Technical Design) ===\n{design_content}"
)
# 4. Execution plan for complex tasks
implement_plan_content = read_file_content(repo_root, f"{task_dir}/implement.md")
if implement_plan_content:
context_parts.append(
f"=== {task_dir}/implement.md (Execution Plan) ===\n{implement_plan_content}"
)
return "\n\n".join(context_parts)
def get_check_context(repo_root: str, task_dir: str) -> str:
"""
Context for Check Agent: check.jsonl + task artifacts.
"""
context_parts = []
for file_path, content in read_jsonl_entries(repo_root, f"{task_dir}/check.jsonl"):
context_parts.append(f"=== {file_path} ===\n{content}")
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}")
design_content = read_file_content(repo_root, f"{task_dir}/design.md")
if design_content:
context_parts.append(
f"=== {task_dir}/design.md (Technical Design) ===\n{design_content}"
)
implement_plan_content = read_file_content(repo_root, f"{task_dir}/implement.md")
if implement_plan_content:
context_parts.append(
f"=== {task_dir}/implement.md (Execution Plan) ===\n{implement_plan_content}"
)
return "\n\n".join(context_parts)
def get_finish_context(repo_root: str, task_dir: str) -> str:
"""
Context for Finish phase: reuses check.jsonl + prd.md
(Finish is a final check, same context source.)
"""
return get_check_context(repo_root, task_dir)
def build_implement_prompt(original_prompt: str, context: str) -> str:
"""Build complete prompt for Implement"""
return f"""<!-- trellis-hook-injected -->
# 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 task artifacts** - Read requirements, technical design if present, and execution plan if present
3. **Implement feature** - Implement following specs and task artifacts
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"""<!-- trellis-hook-injected -->
# 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"""<!-- trellis-hook-injected -->
# 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 task artifacts** - Check requirements in prd.md and, when present, design.md / implement.md
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
- Verify design.md and implement.md constraints when those files are present"""
def get_research_context(repo_root: str, task_dir: str | None) -> str:
"""
Context for Research Agent — project structure overview for spec directories.
`task_dir` kept for signature parity with get_implement_context / get_check_context
so the dispatcher can call them uniformly.
"""
_ = task_dir
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)
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 _string_value(value: Any) -> str:
if isinstance(value, str):
stripped = value.strip()
return stripped
return ""
def _extract_subagent_name(value: Any) -> str:
"""Extract a sub-agent name from common platform encodings.
Cursor's native Task args encode custom sub-agents as a protobuf oneof,
which can appear in hook JSON as either ``{"custom": {"name": "..."}}``
or ``{"type": {"case": "custom", "value": {"name": "..."}}}``.
"""
direct = _string_value(value)
if direct:
return direct
if not isinstance(value, dict):
return ""
for key in ("name", "subagent_type_name", "subagentTypeName"):
direct = _string_value(value.get(key))
if direct:
return direct
custom = value.get("custom")
if isinstance(custom, dict):
custom_name = _string_value(custom.get("name"))
if custom_name:
return custom_name
oneof = value.get("type")
if isinstance(oneof, dict):
case_name = _string_value(oneof.get("case"))
if case_name == "custom":
nested_value = oneof.get("value")
if isinstance(nested_value, dict):
custom_name = _string_value(nested_value.get("name"))
if custom_name:
return custom_name
if case_name:
return case_name
case_name = _string_value(value.get("case"))
if case_name == "custom":
nested_value = value.get("value")
if isinstance(nested_value, dict):
custom_name = _string_value(nested_value.get("name"))
if custom_name:
return custom_name
if case_name:
return case_name
for agent_name in AGENTS_ALL:
if agent_name in value:
return agent_name
return ""
def _extract_subagent_type(tool_input: dict) -> str:
for key in (
"subagent_type",
"subagentType",
"subagent_type_name",
"subagentTypeName",
"agent_type",
"agentType",
"name",
):
agent_name = _extract_subagent_name(tool_input.get(key))
if agent_name:
return agent_name
return ""
def _parse_hook_input(input_data: dict) -> tuple[str, str, dict]:
"""Parse hook input across different platform formats.
Returns (subagent_type, original_prompt, tool_input).
Handles:
- Claude Code / Qoder / CodeBuddy / Droid: tool_name=Task|Agent, tool_input.subagent_type
- Cursor: tool_name=Task|Subagent, tool_input.subagent_type
- Copilot CLI: toolName=task (camelCase key, lowercase value)
- Gemini CLI: tool_name IS the agent name (BeforeTool matcher already filtered)
- Kiro: agentSpawn hook, agent_name field at top level
"""
tool_input = input_data.get("tool_input", {})
# Standard format: Task/Agent tool with subagent_type
tool_name = input_data.get("tool_name", "") or input_data.get("toolName", "")
if tool_name.lower() in ("task", "agent", "subagent"):
return (
_extract_subagent_type(tool_input),
tool_input.get("prompt", ""),
tool_input,
)
# Kiro: agentSpawn hook passes agent_name at top level
agent_name = input_data.get("agent_name", "")
if agent_name:
return agent_name, tool_input.get("prompt", input_data.get("prompt", "")), tool_input
# Gemini CLI: BeforeTool where tool_name IS the agent name
# (matcher already ensured it's one of our agents)
if tool_name in AGENTS_ALL:
return tool_name, tool_input.get("prompt", ""), tool_input
# Copilot CLI: toolName field (camelCase), value might be the agent name
tool_name_camel = input_data.get("toolName", "")
if tool_name_camel in AGENTS_ALL:
return tool_name_camel, input_data.get("toolArgs", ""), tool_input
return "", "", tool_input
def main():
if os.environ.get("TRELLIS_HOOKS") == "0" or os.environ.get("TRELLIS_DISABLE_HOOKS") == "1":
sys.exit(0)
try:
input_data = json.load(sys.stdin)
except json.JSONDecodeError:
sys.exit(0)
subagent_type, original_prompt, tool_input = _parse_hook_input(input_data)
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, input_data)
# implement/check 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)
# 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_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 — use a multi-format output that covers all platforms.
# Most platforms ignore unrecognized fields, so we include multiple formats.
# The platform picks whichever fields it understands.
updated = {**tool_input, "prompt": new_prompt}
output = {
# Claude Code / Qoder / CodeBuddy / Droid format
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "allow",
"updatedInput": updated,
},
# Cursor format
"permission": "allow",
"updated_input": updated,
# Gemini format
"updatedInput": updated,
}
print(json.dumps(output, ensure_ascii=False))
sys.exit(0)
if __name__ == "__main__":
main()

844
.cursor/hooks/session-start.py Normal file
View File

@@ -0,0 +1,844 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Session Start Hook - Inject structured context
"""
from __future__ import annotations
# IMPORTANT: Suppress all warnings FIRST
import warnings
warnings.filterwarnings("ignore")
import json
import os
import re
import shlex
import subprocess
import sys
from io import StringIO
from pathlib import Path
def _normalize_windows_shell_path(path_str: str) -> str:
"""Normalize Unix-style shell paths to real Windows paths.
On Windows, shells like Git Bash / MSYS2 / Cygwin may report paths like
`/d/Users/...` or `/cygdrive/d/Users/...`. `Path.resolve()` will misinterpret
these as `D:/d/Users...` on drive D: (or similar), breaking repo root
detection.
This function is intentionally conservative: it only rewrites patterns that
unambiguously represent a drive letter mount.
"""
if not isinstance(path_str, str) or not path_str:
return path_str
# Only relevant on Windows; keep other platforms untouched.
if not sys.platform.startswith("win"):
return path_str
p = path_str.strip()
# Already a Windows drive path (C:\... or C:/...)
if re.match(r"^[A-Za-z]:[\/]", p):
return p
# MSYS/Git-Bash style: /c/Users/... or /d/Work/...
m = re.match(r"^/([A-Za-z])/(.*)", p)
if m:
drive, rest = m.group(1).upper(), m.group(2)
rest = rest.replace('/', '\\')
return f"{drive}:\\{rest}"
# Cygwin style: /cygdrive/c/Users/...
m = re.match(r"^/cygdrive/([A-Za-z])/(.*)", p)
if m:
drive, rest = m.group(1).upper(), m.group(2)
rest = rest.replace('/', '\\')
return f"{drive}:\\{rest}"
# WSL mounted drive (sometimes leaked into env): /mnt/c/Users/...
m = re.match(r"^/mnt/([A-Za-z])/(.*)", p)
if m:
drive, rest = m.group(1).upper(), m.group(2)
rest = rest.replace('/', '\\')
return f"{drive}:\\{rest}"
return path_str
FIRST_REPLY_NOTICE = """<first-reply-notice>
First visible reply: say once in Chinese that Trellis SessionStart context is loaded, then answer directly.
This notice is one-shot: do not repeat it after the first assistant reply in the same session.
</first-reply-notice>"""
# Force UTF-8 on stdin/stdout/stderr on Windows. Default codepage there is
# cp936 / cp1252 / etc. — non-ASCII content (Chinese task names, prd snippets)
# both in stdin (hook payload from host CLI) and stdout (our emitted blocks)
# raises UnicodeDecodeError / UnicodeEncodeError. Equivalent to `python -X utf8`
# but applied per-stream so we don't depend on host CLI's command wiring.
if sys.platform.startswith("win"):
import io as _io
for _stream_name in ("stdin", "stdout", "stderr"):
_stream = getattr(sys, _stream_name, None)
if _stream is None:
continue
if hasattr(_stream, "reconfigure"):
try:
_stream.reconfigure(encoding="utf-8", errors="replace") # type: ignore[union-attr]
except Exception:
pass
elif hasattr(_stream, "detach"):
try:
setattr(sys, _stream_name, _io.TextIOWrapper(_stream.detach(), encoding="utf-8", errors="replace"))
except Exception:
pass
def _has_curated_jsonl_entry(jsonl_path: Path) -> bool:
"""Return True iff jsonl has at least one row with a ``file`` field.
A freshly seeded jsonl only contains a ``{"_example": ...}`` row (no
``file`` key) — that is NOT "ready". Readiness requires at least one
curated entry. Matches the contract used by hook-inject and pull-based
sub-agent context loaders.
"""
try:
for line in jsonl_path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line:
continue
try:
row = json.loads(line)
except json.JSONDecodeError:
continue
if isinstance(row, dict) and row.get("file"):
return True
except (OSError, UnicodeDecodeError):
return False
return False
def should_skip_injection() -> bool:
"""Check if any platform's non-interactive flag is set, or if Trellis
hooks are explicitly disabled via TRELLIS_HOOKS=0 / TRELLIS_DISABLE_HOOKS=1.
"""
if os.environ.get("TRELLIS_HOOKS") == "0":
return True
if os.environ.get("TRELLIS_DISABLE_HOOKS") == "1":
return True
non_interactive_vars = [
"CLAUDE_NON_INTERACTIVE",
"QODER_NON_INTERACTIVE",
"CODEBUDDY_NON_INTERACTIVE",
"FACTORY_NON_INTERACTIVE",
"CURSOR_NON_INTERACTIVE",
"GEMINI_NON_INTERACTIVE",
"KIRO_NON_INTERACTIVE",
"COPILOT_NON_INTERACTIVE",
"TRAE_NON_INTERACTIVE",
]
return any(os.environ.get(var) == "1" for var in non_interactive_vars)
def read_file(path: Path, fallback: str = "") -> str:
try:
return path.read_text(encoding="utf-8")
except (FileNotFoundError, PermissionError):
return fallback
def _repo_relative(repo_root: Path, path: Path) -> str:
try:
return path.relative_to(repo_root).as_posix()
except ValueError:
return str(path)
def _run_git(repo_root: Path, args: list[str]) -> str:
try:
result = subprocess.run(
["git", *args],
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=3,
cwd=str(repo_root),
)
except (subprocess.TimeoutExpired, FileNotFoundError, PermissionError):
return ""
if result.returncode != 0:
return ""
return result.stdout.strip()
def _format_git_state(repo_root: Path) -> str:
branch = _run_git(repo_root, ["branch", "--show-current"]) or "(detached)"
dirty_lines = [
line for line in _run_git(repo_root, ["status", "--porcelain"]).splitlines()
if line.strip()
]
dirty_text = "clean" if not dirty_lines else f"dirty {len(dirty_lines)} paths"
return f"Git: branch {branch}; {dirty_text}."
def _detect_platform(input_data: dict) -> str | None:
if isinstance(input_data.get("cursor_version"), str):
return "cursor"
env_map = {
"CLAUDE_PROJECT_DIR": "claude",
"CURSOR_PROJECT_DIR": "cursor",
"CODEBUDDY_PROJECT_DIR": "codebuddy",
"FACTORY_PROJECT_DIR": "droid",
"GEMINI_PROJECT_DIR": "gemini",
"QODER_PROJECT_DIR": "qoder",
"KIRO_PROJECT_DIR": "kiro",
"COPILOT_PROJECT_DIR": "copilot",
"TRAE_PROJECT_DIR": "trae",
}
for env_name, platform in env_map.items():
if os.environ.get(env_name):
return platform
script_parts = set(Path(sys.argv[0]).parts)
if ".claude" in script_parts:
return "claude"
if ".cursor" in script_parts:
return "cursor"
if ".codex" in script_parts:
return "codex"
if ".gemini" in script_parts:
return "gemini"
if ".qoder" in script_parts:
return "qoder"
if ".codebuddy" in script_parts:
return "codebuddy"
if ".factory" in script_parts:
return "droid"
if ".kiro" in script_parts:
return "kiro"
if ".trae" in script_parts:
return "trae"
return None
def _resolve_context_key(trellis_dir: Path, input_data: dict) -> str | None:
scripts_dir = trellis_dir / "scripts"
if str(scripts_dir) not in sys.path:
sys.path.insert(0, str(scripts_dir))
from common.active_task import resolve_context_key # type: ignore[import-not-found]
return resolve_context_key(input_data, platform=_detect_platform(input_data))
def _persist_context_key_for_bash(context_key: str | None) -> None:
"""Expose Trellis session identity to later Claude Code Bash commands.
Claude Code SessionStart hooks can append exports to CLAUDE_ENV_FILE; those
variables are then available to Bash tools in the same conversation. Without
this bridge, `task.py start` has hook stdin during SessionStart but no
session identity when the AI later runs it as a normal shell command.
"""
if not context_key:
return
env_file = os.environ.get("CLAUDE_ENV_FILE")
if not env_file:
return
try:
with open(env_file, "a", encoding="utf-8") as handle:
handle.write(f"export TRELLIS_CONTEXT_ID={shlex.quote(context_key)}\n")
except OSError:
pass
def _resolve_active_task(trellis_dir: Path, input_data: dict):
scripts_dir = trellis_dir / "scripts"
if str(scripts_dir) not in sys.path:
sys.path.insert(0, str(scripts_dir))
from common.active_task import resolve_active_task # type: ignore[import-not-found]
return resolve_active_task(
trellis_dir.parent,
input_data,
platform=_detect_platform(input_data),
)
def run_script(script_path: Path, context_key: str | None = None) -> str:
try:
if script_path.suffix == ".py":
# Add PYTHONIOENCODING to force UTF-8 in subprocess
env = os.environ.copy()
env["PYTHONIOENCODING"] = "utf-8"
if context_key:
env["TRELLIS_CONTEXT_ID"] = context_key
cmd = [sys.executable, "-W", "ignore", str(script_path)]
else:
env = os.environ.copy()
if context_key:
env["TRELLIS_CONTEXT_ID"] = context_key
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, input_data: dict) -> str:
"""Return compact active-task status, artifact presence, and next action."""
active = _resolve_active_task(trellis_dir, input_data)
if not active.task_path:
return (
"Status: NO ACTIVE TASK\n"
"Next-Action: Classify the current turn before creating any Trellis task. "
"Simple conversation / small task asks only whether this turn should create a Trellis task. "
"Complex task asks whether task creation and planning are allowed."
)
task_ref = active.task_path
task_dir = _resolve_task_dir(trellis_dir, task_ref)
if active.stale or not task_dir.is_dir():
return (
f"Status: STALE POINTER\nTask: {task_ref}\n"
f"Next-Action: Run `python3 ./.trellis/scripts/task.py finish` to clear the stale pointer, "
"then ask the user what to work on next."
)
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")
artifact_names = ("prd.md", "design.md", "implement.md", "implement.jsonl", "check.jsonl")
present = [name for name in artifact_names if (task_dir / name).is_file()]
if (task_dir / "research").is_dir():
present.append("research/")
present_line = ", ".join(present) if present else "(none)"
if task_status == "completed":
return (
f"Status: COMPLETED\nTask: {task_title}\n"
f"Present: {present_line}\n"
"Next-Action: Run `/trellis:finish-work`. If the working tree is dirty, return to Phase 3.4 first."
)
has_prd = (task_dir / "prd.md").is_file()
has_design = (task_dir / "design.md").is_file()
has_implement_plan = (task_dir / "implement.md").is_file()
implement_jsonl = task_dir / "implement.jsonl"
check_jsonl = task_dir / "check.jsonl"
jsonl_ready = (
(not implement_jsonl.is_file() or _has_curated_jsonl_entry(implement_jsonl))
and (not check_jsonl.is_file() or _has_curated_jsonl_entry(check_jsonl))
)
if task_status == "planning" and not has_prd:
return (
f"Status: PLANNING\nTask: {task_title}\n"
f"Present: {present_line}\n"
"Next-Action: Load `trellis-brainstorm` and write `prd.md`. Stay in planning."
)
if task_status == "planning":
missing_complex = [
name for name, exists in (
("design.md", has_design),
("implement.md", has_implement_plan),
)
if not exists
]
next_bits: list[str] = []
if missing_complex:
next_bits.append(
"Lightweight task can request start review with PRD-only; "
f"complex task must add {', '.join(missing_complex)} before start"
)
else:
next_bits.append("Planning artifacts are present; ask for review before `task.py start`")
if not jsonl_ready:
next_bits.append("curate `implement.jsonl` and `check.jsonl` before sub-agent mode start")
return (
f"Status: PLANNING\nTask: {task_title}\n"
f"Present: {present_line}\n"
f"Next-Action: {'; '.join(next_bits)}. Do not enter implementation until the user confirms start."
)
return (
f"Status: {str(task_status).upper()}\nTask: {task_title}\n"
f"Present: {present_line}\n"
"Next-Action: Follow the matching per-turn workflow-state. "
"Implementation/check context order is jsonl entries -> `prd.md` -> `design.md if present` -> `implement.md if present`."
)
def _load_trellis_config(trellis_dir: Path, input_data: dict) -> 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,
input_data,
platform=_detect_platform(input_data),
)
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 _collect_spec_index_paths(trellis_dir: Path, allowed_pkgs: set | None) -> list[str]:
paths: list[str] = []
guides_index = trellis_dir / "spec" / "guides" / "index.md"
if guides_index.is_file():
paths.append(".trellis/spec/guides/index.md")
spec_dir = trellis_dir / "spec"
if not spec_dir.is_dir():
return paths
for sub in sorted(spec_dir.iterdir()):
if not sub.is_dir() or sub.name.startswith(".") or sub.name == "guides":
continue
index_file = sub / "index.md"
if index_file.is_file():
paths.append(f".trellis/spec/{sub.name}/index.md")
continue
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():
paths.append(f".trellis/spec/{sub.name}/{nested.name}/index.md")
return paths
def _build_compact_current_state(
trellis_dir: Path,
input_data: dict,
spec_index_paths: list[str],
) -> str:
repo_root = trellis_dir.parent
lines: list[str] = []
try:
from common.paths import get_active_journal_file, get_developer, get_tasks_dir, count_lines # type: ignore[import-not-found]
from common.tasks import iter_active_tasks # type: ignore[import-not-found]
except Exception:
get_active_journal_file = None # type: ignore[assignment]
get_developer = None # type: ignore[assignment]
get_tasks_dir = None # type: ignore[assignment]
count_lines = None # type: ignore[assignment]
iter_active_tasks = None # type: ignore[assignment]
developer = get_developer(repo_root) if get_developer else None
lines.append(f"Developer: {developer or '(not initialized)'}")
lines.append(_format_git_state(repo_root))
active = _resolve_active_task(trellis_dir, input_data)
if active.task_path:
task_dir = _resolve_task_dir(trellis_dir, active.task_path)
status = "unknown"
task_json = task_dir / "task.json"
if task_json.is_file():
try:
data = json.loads(task_json.read_text(encoding="utf-8"))
if isinstance(data, dict):
status = str(data.get("status") or "unknown")
except (json.JSONDecodeError, OSError):
pass
lines.append(f"Current task: {_repo_relative(repo_root, task_dir)}; status={status}.")
else:
lines.append("Current task: none.")
if get_tasks_dir and iter_active_tasks:
try:
task_count = sum(1 for _ in iter_active_tasks(get_tasks_dir(repo_root)))
lines.append(
f"Active tasks: {task_count} total. Use `python3 ./.trellis/scripts/task.py list --mine` only if needed."
)
except Exception:
pass
if get_active_journal_file and count_lines:
journal = get_active_journal_file(repo_root)
if journal:
lines.append(
f"Journal: {_repo_relative(repo_root, journal)}, {count_lines(journal)} / 2000 lines."
)
if spec_index_paths:
lines.append(f"Spec indexes: {len(spec_index_paths)} available.")
return "\n".join(lines)
def _extract_range(content: str, start_header: str, end_header: str) -> str:
"""Extract lines starting at `## start_header` up to (but excluding) `## end_header`.
Both parameters are full header lines WITHOUT the `## ` prefix (e.g. "Phase Index").
Returns empty string if start header is not found.
End header missing → extracts to end of file.
"""
lines = content.splitlines()
start: int | None = None
end: int = len(lines)
start_match = f"## {start_header}"
end_match = f"## {end_header}"
for i, line in enumerate(lines):
stripped = line.strip()
if start is None and stripped == start_match:
start = i
continue
if start is not None and stripped == end_match:
end = i
break
if start is None:
return ""
return "\n".join(lines[start:end]).rstrip()
_BREADCRUMB_TAG_RE = re.compile(
r"\[workflow-state:([A-Za-z0-9_-]+)\]\s*\n.*?\n\s*\[/workflow-state:\1\]",
re.DOTALL,
)
def _strip_breadcrumb_tag_blocks(content: str) -> str:
"""Remove `[workflow-state:STATUS]...[/workflow-state:STATUS]` blocks.
The tag blocks live inside `## Phase Index` (since v0.5.0-rc.0, when
they were colocated with their phase summaries) and are consumed by the
UserPromptSubmit hook (`inject-workflow-state.py`). The session-start
payload already covers the full step bodies, so re-inlining the
breadcrumbs here would just duplicate context.
"""
stripped = _BREADCRUMB_TAG_RE.sub("", content)
stripped = re.sub(r"<!--.*?-->", "", stripped, flags=re.DOTALL)
stripped = re.sub(r"^\[(?!/?workflow-state:)/?[^\]\n]+\]\s*\n?", "", stripped, flags=re.MULTILINE)
return re.sub(r"\n{3,}", "\n\n", stripped).strip()
def _build_workflow_overview(workflow_path: Path) -> str:
"""Inject only the compact Phase Index summary for SessionStart."""
content = read_file(workflow_path)
if not content:
return "No workflow.md found"
out_lines = [
"# Development Workflow - Session Summary",
"Full guide: .trellis/workflow.md. Step detail: `python3 ./.trellis/scripts/get_context.py --mode phase --step <X.Y>`.",
"",
]
phases = _extract_range(content, "Phase Index", "Phase 1: Plan")
if phases:
out_lines.append(_strip_breadcrumb_tag_blocks(phases).rstrip())
return "\n".join(out_lines).rstrip()
def main():
if should_skip_injection():
sys.exit(0)
try:
hook_input = json.loads(sys.stdin.read())
if not isinstance(hook_input, dict):
hook_input = {}
except (json.JSONDecodeError, ValueError):
hook_input = {}
# Try platform-specific env vars, hook cwd, fallback to cwd
project_dir_env_vars = [
"CLAUDE_PROJECT_DIR",
"QODER_PROJECT_DIR",
"CODEBUDDY_PROJECT_DIR",
"FACTORY_PROJECT_DIR",
"CURSOR_PROJECT_DIR",
"GEMINI_PROJECT_DIR",
"KIRO_PROJECT_DIR",
"COPILOT_PROJECT_DIR",
"TRAE_PROJECT_DIR",
]
project_dir = None
for var in project_dir_env_vars:
val = os.environ.get(var)
if val:
project_dir = Path(_normalize_windows_shell_path(val)).resolve()
break
if project_dir is None:
project_dir = Path(_normalize_windows_shell_path(hook_input.get("cwd", "."))).resolve()
trellis_dir = project_dir / ".trellis"
context_key = _resolve_context_key(trellis_dir, hook_input)
_persist_context_key_for_bash(context_key)
# Load config for scope filtering and legacy detection
is_mono, packages, scope_config, task_pkg, default_pkg = _load_trellis_config(
trellis_dir,
hook_input,
)
allowed_pkgs = _resolve_spec_scope(is_mono, packages, scope_config, task_pkg, default_pkg)
output = StringIO()
spec_index_paths = _collect_spec_index_paths(trellis_dir, allowed_pkgs)
output.write("""<session-context>
Trellis compact SessionStart context. Use it to orient the session; load details on demand.
</session-context>
""")
output.write(FIRST_REPLY_NOTICE)
output.write("\n\n")
# 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")
output.write(_build_compact_current_state(trellis_dir, hook_input, spec_index_paths))
output.write("\n</current-state>\n\n")
output.write("<trellis-workflow>\n")
output.write(_build_workflow_overview(trellis_dir / "workflow.md"))
output.write("\n</trellis-workflow>\n\n")
output.write("<guidelines>\n")
output.write(
"Task context order for implementation/check: jsonl entries -> `prd.md` -> "
"`design.md if present` -> `implement.md if present`. Missing optional artifacts "
"are skipped for lightweight tasks.\n\n"
)
if spec_index_paths:
output.write("## Available indexes (read on demand)\n")
for p in spec_index_paths:
output.write(f"- {p}\n")
output.write("\n")
output.write(
"Discover more via: "
"`python3 ./.trellis/scripts/get_context.py --mode packages`\n"
)
output.write("</guidelines>\n\n")
# Check task status and inject structured tag
task_status = _get_task_status(trellis_dir, hook_input)
output.write(f"<task-status>\n{task_status}\n</task-status>\n\n")
output.write("""<ready>
Context loaded. Follow <task-status>. Load workflow/spec/task details only when needed.
</ready>""")
context_text = output.getvalue()
# Kiro (CLI trellis agent agentSpawn) adds a hook's stdout directly to the
# conversation context — no JSON envelope. Emit the bare overview text.
# Conditionally isolated: all other platforms keep the JSON path below.
if _detect_platform(hook_input) == "kiro":
print(context_text, flush=True)
return
result = {
# Claude Code / Qoder / CodeBuddy / Droid / Gemini / Copilot format
"hookSpecificOutput": {
"hookEventName": "SessionStart",
"additionalContext": context_text,
},
# Cursor sessionStart format (top-level snake_case per Cursor docs)
"additional_context": context_text,
}
# Output JSON - stdout is already configured for UTF-8
print(json.dumps(result, ensure_ascii=False), flush=True)
if __name__ == "__main__":
main()