#!/usr/bin/env python3 """ FlyDocs Hook: prompt-submit.py Triggered: When user submits a prompt Purpose: Inject context and validate workflow state Exit codes: 0 - Success (plain text output adds context to conversation) 2 - Block prompt (stderr shown as reason) NOTE: Uses plain text output instead of JSON due to Claude Code bug (Issue #13912) where JSON output from UserPromptSubmit hooks causes "hook error" despite documentation. """ import json import os import re import subprocess import sys from pathlib import Path DEBUG_HOOK = os.environ.get('DEBUG_HOOK', '0') == '1' SCRIPT_DIR = Path(__file__).parent.resolve() DEBUG_LOG = SCRIPT_DIR.parent / 'logs' / 'hook-debug.log' # FLY-1272: the status vocabulary lives in one place. Appended, not inserted, # so the workflow scripts directory can never shadow the standard library. sys.path.append( str(SCRIPT_DIR.parent / 'skills' / 'flydocs-workflow' / 'scripts') ) from status_vocab import normalize # noqa: E402 def debug_log(message: str) -> None: """Write debug message to log file if DEBUG_HOOK is enabled.""" if not DEBUG_HOOK: return try: DEBUG_LOG.parent.mkdir(parents=True, exist_ok=True) from datetime import datetime timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') with open(DEBUG_LOG, 'a') as f: f.write(f'[{timestamp}] {message}\n') except (OSError, IOError): pass def get_git_context() -> str | None: """Get git branch and uncommitted status in a single subprocess call.""" try: result = subprocess.run( ['git', 'status', '--porcelain', '-b'], capture_output=True, text=True, timeout=5 ) if result.returncode != 0: return None lines = result.stdout.splitlines() if not lines: return None # First line: ## branch...tracking branch = 'detached' header = lines[0] if header.startswith('## '): branch_part = header[3:].split('...')[0] if branch_part and branch_part != 'HEAD (no branch)': branch = branch_part # Any remaining lines = uncommitted changes uncommitted = 'yes' if len(lines) > 1 else 'no' return f'Git: branch={branch}, uncommitted={uncommitted}' except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError): return None def _session_dir(repo_dir: str | None = None) -> Path: """Resolve the workspace-scoped session directory (FLY-674).""" try: from repo_context import resolve_session_dir return resolve_session_dir(repo_dir) except (ImportError, Exception): base = Path(repo_dir) if repo_dir else Path('.') return base / '.flydocs' / 'session' / 'default' # FLY-1186: issue-ref shape, matching the server contract the collector # validates against (usage-attribution.ts ISSUE_REF_PATTERN). ISSUE_REF_RE = re.compile(r'[A-Z][A-Z0-9]{1,9}-[0-9]{1,6}') def get_issue_context( repo_dir: str | None = None, ) -> tuple[str | None, str | None, str | None]: """Get active issue, status, and any focus warning from session files. `focus.md` and `status` are independent files that can describe different issues. Pairing them unconditionally fabricates a state that never existed — e.g. reporting "FLY-1028 | DUPLICATE" for an issue genuinely In Review, by combining one issue's ref with another's status (FLY-1064). `status-ref` records which issue `status` actually describes. FLY-1186: focus.md is a cache, not the source. When it yields no valid ref (missing, or holding garbage like a leaked CLI flag), fall back to `status-ref` — written and validated by `issues.py transition`. A malformed focus.md is a visible warning, never a silent "no issue": that silence once cost a full day of issue attribution. """ sd = _session_dir(repo_dir) focus_file = sd / 'focus.md' status_file = sd / 'status' status_ref_file = sd / 'status-ref' issue_id = None status = None warning = None focus_content = None if focus_file.exists(): try: focus_content = focus_file.read_text() match = ISSUE_REF_RE.search(focus_content) if match: issue_id = match.group(0) except (OSError, IOError): pass if not issue_id and status_ref_file.exists(): try: candidate = status_ref_file.read_text().strip().upper() if ISSUE_REF_RE.fullmatch(candidate): issue_id = candidate except (OSError, IOError): pass if focus_content and focus_content.strip() and not ISSUE_REF_RE.search(focus_content): if issue_id: warning = ( f'[focus.md holds no valid issue ref — using status-ref ' f'{issue_id}. Run /activate or transition to repair it]' ) else: warning = ( '[focus.md holds no valid issue ref and status-ref is empty — ' 'issue attribution is off. Run /activate or transition an issue]' ) if issue_id and not status_file.exists() and warning is None: # FLY-1471: focused with nothing recorded for it. The status line # already omits the status; what was missing is that this is a state # the session should get out of — every gate reads it as unknown and # stands down. warning = ( f'[{issue_id} is focused but has no recorded status — the gates ' f'are off. Transition {issue_id} to re-arm them]' ) if issue_id and status_file.exists(): try: tracked_ref = ( status_ref_file.read_text().strip().upper() if status_ref_file.exists() else None ) if tracked_ref == issue_id.upper(): status = status_file.read_text().strip() elif warning is None: # FLY-1471: the status stays withheld — pairing this issue # with another's status is the fabricated "FLY-1028 | # DUPLICATE" line FLY-1064 removed — but the silence is what # let the drift survive a whole session, with the Stop gate # and the edit gate quietly standing down the entire time. # # Same channel as the malformed-focus.md warning above: `main` # prints it (~663-665), and a UserPromptSubmit hook's stdout # is what the agent sees. Guarded on `warning` so the repair # instruction for an unreadable focus.md is never displaced by # this one. described = tracked_ref or "(none)" warning = ( f'[session status describes {described}, not the focused ' f'{issue_id} — status hidden. Transition {issue_id} to ' f'resync]' ) except (OSError, IOError): pass return issue_id, status, warning def get_active_project_id(repo_dir: str | None = None) -> str | None: """Resolve the active provider project id from config (ADR-011). Prefers the singular top-level `activeProjectId`; falls back to the first entry of `activeProjects`. Best-effort — never raises. """ try: base = Path(repo_dir) if repo_dir else Path('.') cfg = json.loads((base / '.flydocs' / 'config.json').read_text()) project_id = cfg.get('activeProjectId') if not project_id: active = cfg.get('activeProjects') or [] project_id = active[0] if active else None return project_id if isinstance(project_id, str) and project_id else None except Exception: return None def append_usage_attribution( session_id: str, issue_id: str | None, repo_dir: str | None = None, project_id: str | None = None, ) -> None: """Append a correlated attribution tuple for AI spend apportioning. Local-only JSONL log at `.flydocs/session/usage-attribution.jsonl` (flat, unscoped — like the active-repo pointer). `flydocs usage report` reads it to compute per-issue and per-project token weights for a session's time window (FLY-1013 issue attribution; FLY-1054 adds the active provider project so planning/refinement work that precedes an issue is still attributed). Must never crash or alter the hook's status-line behavior. """ try: import time import zlib base = Path(repo_dir) if repo_dir else Path('.') log_file = base / '.flydocs' / 'session' / 'usage-attribution.jsonl' log_file.parent.mkdir(parents=True, exist_ok=True) ts = int(time.time()) tuple_obj = {'ts': ts, 'sid': session_id or None, 'issue': issue_id} # Only record project when known — keeps legacy lines unchanged and # the log compact when no provider project is active. if project_id: tuple_obj['project'] = project_id line = json.dumps(tuple_obj, separators=(',', ':')) with open(log_file, 'a') as f: f.write(line + '\n') # Cap check is O(file) — only run on ~1-in-50 appends (deterministic # hash of sid+ts, no state needed). Keep the newest 2500 lines when # the log exceeds 5000. if zlib.crc32(f'{session_id}{ts}'.encode()) % 50 == 0: lines = log_file.read_text().splitlines() if len(lines) > 5000: log_file.write_text('\n'.join(lines[-2500:]) + '\n') except Exception: pass def get_focus_descriptor(repo_dir: str | None = None) -> str | None: """Describe the active focus for the prompt context (FLY-1098). Read from config only — this runs on every prompt, so a relay call here would tax each turn. `activeContexts` carries a sprint or board name when present; the flat `activeSprintId` does not, so the fallback states that a sprint focus exists rather than printing a bare UUID. """ base = Path(repo_dir) if repo_dir else Path('.') config_file = base / '.flydocs' / 'config.json' if not config_file.exists(): return None try: config = json.loads(config_file.read_text()) except (json.JSONDecodeError, OSError): return None contexts = config.get('activeContexts') or [] if contexts: ctx = contexts[0] name = ctx.get('sprintName') or ctx.get('name') if ctx.get('sprintId'): return f"Sprint: {name}" if name else "Sprint: active" if ctx.get('id') and ctx.get('type') == 'board': return f"Board: {name}" if name else "Board: active" if config.get('activeSprintId'): return "Sprint: active" return None def get_issue_context_line(repo_dir: str | None = None) -> str | None: """Build a rich single-line issue context for prompt injection. Consolidates issue ID, status, AC progress, and assignment into one compact line that gives the agent full situational awareness. """ issue_id, status, _warning = get_issue_context(repo_dir) if not issue_id: return None sd = _session_dir(repo_dir) parts = [issue_id] # Status — always display canonical FlyDocs names (FLY-688) if status: parts.append(status) # FLY-1098: the acceptance-criteria snapshot was counted here. FLY-1065 # removed its writer, so this read now always finds nothing. # FLY-1098: name the narrower set so the agent knows focus is available. focus = get_focus_descriptor(repo_dir) if focus: parts.append(focus) # Assignee from focus file focus_file = sd / 'focus.md' if focus_file.exists(): try: content = focus_file.read_text() assignee_match = re.search(r'[Aa]ssignee:\s*(.+)', content) if assignee_match: parts.append(f'Assigned: {assignee_match.group(1).strip()}') except (OSError, IOError): pass # FLY-1267: the per-status directive used to be repeated here. It is # printed once per turn by get_workflow_directive() — which covers more # statuses and says more — so emitting it twice only cost tokens. return f'Issue: {" | ".join(parts)}' def get_workflow_directive(status: str | None, has_issue: bool) -> str | None: """Get directive workflow instruction based on current state. These are not reminders — they are required actions the agent must follow. """ if not has_issue: return '[No active issue. Run /activate to pick an issue or /capture to create one before writing code]' if not status: return None # Statuses that carry a standing instruction. Keyed on the canonical # vocabulary in status_vocab.py — resolving through normalize() means a # provider-native name reaching this far still finds its directive instead # of silently matching nothing (FLY-1272). directives = { 'BACKLOG': '[REQUIRED: Read stages/activate.md. Transition to READY or IMPLEMENTING before starting work]', 'READY': '[REQUIRED: Read stages/activate.md. Transition to IMPLEMENTING before writing code]', 'IMPLEMENTING': '[REQUIRED: Tick AC checkboxes with `issue_acceptance_update` (fallback: `issues.py acceptance REF --check N`) as you finish them. Add progress comments for milestones. Transition to REVIEW when implementation is complete]', 'REVIEW': '[REQUIRED: Verify ALL acceptance criteria are checked. Read stages/review.md for the full review procedure]', 'TESTING': '[REQUIRED: Validate all AC met. Read stages/validate.md before marking COMPLETE]', 'BLOCKED': '[Issue is BLOCKED. Resolve the blocker or escalate. Transition back to IMPLEMENTING when unblocked]', } return directives.get(normalize(status)) def check_integrity_drift() -> str | None: """Lightweight integrity drift check with 30-minute TTL. Reads .flydocs/integrity.json and checks owned files/directories exist. Caches result to avoid re-checking on every prompt. """ integrity_file = Path('.flydocs/integrity.json') if not integrity_file.exists(): return None # TTL check — skip if checked within 30 minutes cache_file = Path('.flydocs/integrity-cache.json') if cache_file.exists(): try: cache = json.loads(cache_file.read_text()) from datetime import datetime, timezone cached_at = datetime.fromisoformat(cache.get('checkedAt', '').replace('Z', '+00:00')) now = datetime.now(timezone.utc) age_minutes = (now - cached_at).total_seconds() / 60 if age_minutes < 30: # Return cached result if still fresh missing = cache.get('missing', []) if missing: return f'[Integrity drift: {len(missing)} owned file(s) missing — run /flydocs-update]' return None except (json.JSONDecodeError, OSError, ValueError, KeyError): pass # Run check try: data = json.loads(integrity_file.read_text()) except (json.JSONDecodeError, OSError): return None missing = [] for f in data.get('ownedFiles', []): if not Path(f).exists(): missing.append(f) for d in data.get('ownedDirectories', []): if not Path(d).exists(): missing.append(d) # Write cache try: from datetime import datetime, timezone cache_data = { 'checkedAt': datetime.now(timezone.utc).isoformat(), 'missing': missing, } cache_file.write_text(json.dumps(cache_data)) except (OSError, IOError): pass if missing: return f'[Integrity drift: {len(missing)} owned file(s) missing — run /flydocs-update]' return None def get_setup_nudge() -> str | None: """Check if setup has been completed, return nudge if not. Reads validation cache (written by validate_setup.py) for specific missing items. Falls back to generic nudge if cache is absent. """ config_file = Path('.flydocs/config.json') if not config_file.exists(): return None try: config = json.loads(config_file.read_text()) if config.get('setupComplete') is not False: return None # Check validation cache for specific missing items cache_file = Path('.flydocs/validation-cache.json') if cache_file.exists(): try: cache = json.loads(cache_file.read_text()) missing = cache.get('missing', []) warnings = cache.get('warnings', []) parts = [] if missing: parts.append(f'missing: {", ".join(missing)}') if warnings: parts.append(f'warnings: {", ".join(warnings)}') if parts: detail = '; '.join(parts) return f'[Setup incomplete — {detail}. Run /start-session or fix in dashboard]' except (json.JSONDecodeError, OSError, IOError): pass return '[Setup incomplete — run /start-session to configure your project]' except (json.JSONDecodeError, OSError, IOError): pass return None def _nudge_already_shown(cache_file: Path, session_id: str) -> bool: """Return True if the staleness nudge already fired this session (FLY-1066).""" if not session_id: return False try: cache = json.loads(cache_file.read_text()) if cache_file.exists() else {} except (json.JSONDecodeError, OSError): cache = {} if cache.get('staleNudgeSession') == session_id: return True cache['staleNudgeSession'] = session_id try: cache_file.parent.mkdir(parents=True, exist_ok=True) cache_file.write_text(json.dumps(cache, indent=2)) except OSError: pass return False def get_config_freshness_nudge(session_id: str = '') -> str | None: """Nudge if validation cache is stale (>24h old). Only applies to cloud tier with setupComplete=true. Encourages periodic re-validation so config stays in sync with the server. """ config_file = Path('.flydocs/config.json') if not config_file.exists(): return None try: config = json.loads(config_file.read_text()) # Only check freshness for cloud tier with completed setup if config.get('tier') != 'cloud' or config.get('setupComplete') is not True: return None cache_file = Path('.flydocs/validation-cache.json') if not cache_file.exists(): if _nudge_already_shown(cache_file, session_id): return None return '[Config not validated — run: python3 .claude/skills/flydocs-workflow/scripts/workspace.py validate]' from datetime import datetime, timezone cache = json.loads(cache_file.read_text()) timestamp_str = cache.get('timestamp') if not timestamp_str: return None # Parse ISO timestamp cached_at = datetime.fromisoformat(timestamp_str.replace('Z', '+00:00')) now = datetime.now(timezone.utc) age_hours = (now - cached_at).total_seconds() / 3600 if age_hours > 24: if _nudge_already_shown(cache_file, session_id): return None return ( f'[Config stale ({int(age_hours)}h) — run: python3 ' f'.claude/skills/flydocs-workflow/scripts/workspace.py validate]' ) except (json.JSONDecodeError, OSError, IOError, ValueError): pass return None def main() -> None: """Main hook execution.""" debug_log('=== Hook invoked ===') debug_log(f'PWD: {os.getcwd()}') debug_log(f'SCRIPT_DIR: {SCRIPT_DIR}') debug_log(f'CLAUDE_PROJECT_DIR: {os.environ.get("CLAUDE_PROJECT_DIR", "")}') # Read hook input from stdin try: input_data = json.loads(sys.stdin.read()) except (json.JSONDecodeError, ValueError): input_data = {} prompt = input_data.get('prompt', '') # This hook is registered for Claude Code (UserPromptSubmit) and for Cursor # (beforeSubmitPrompt, see .cursor/hooks.json), which name the same things # differently: Cursor sends `conversation_id` and a `workspace_roots` array # rather than `session_id` and `cwd`. Accepting both matters for AI Spend — # Cursor's conversation id is the same id its usage records carry, so # capturing it here attributes Cursor work to the right issue precisely # instead of falling back to a time-window guess (FLY-1027). session_id = input_data.get('session_id') or input_data.get('conversation_id') or '' workspace_roots = input_data.get('workspace_roots') or [] cwd = input_data.get('cwd') or ( workspace_roots[0] if isinstance(workspace_roots, list) and workspace_roots else '' ) debug_log(f'Parsed PROMPT: {prompt[:100]}...' if prompt else 'Parsed PROMPT: ') debug_log(f'Parsed SESSION_ID: {session_id}') debug_log(f'Parsed CWD: {cwd}') # Change to working directory debug_log('Attempting to change directory...') if cwd and Path(cwd).is_dir(): debug_log(f'Using CWD from input: {cwd}') os.chdir(cwd) elif os.environ.get('CLAUDE_PROJECT_DIR') and Path(os.environ['CLAUDE_PROJECT_DIR']).is_dir(): debug_log(f'Using CLAUDE_PROJECT_DIR: {os.environ["CLAUDE_PROJECT_DIR"]}') os.chdir(os.environ['CLAUDE_PROJECT_DIR']) else: debug_log(f'No valid directory to change to, staying in: {os.getcwd()}') debug_log(f'Now in directory: {os.getcwd()}') # Resolve active repo for sibling-repos topology from repo_context import resolve_repo_dir repo_dir = resolve_repo_dir() is_workspace = repo_dir != os.getcwd() debug_log(f'Resolved repo_dir: {repo_dir} (workspace={is_workspace})') # Build context parts context_parts = [] # Git context git_context = get_git_context() branch = None if git_context: debug_log(f'Git context: {git_context}') context_parts.append(git_context) # Extract branch name for graph context branch_match = re.search(r'branch=(\S+)', git_context) if branch_match: branch = branch_match.group(1).rstrip(',') # Active project + board context # In sibling-repos, read config from the active child repo config_base = Path(repo_dir) if is_workspace else Path('.') config_file = config_base / '.flydocs/config.json' cfg = None if config_file.exists(): try: cfg = json.loads(config_file.read_text()) # ADR-011: direct top-level reads, activeProjectId (singular) active_project_id = cfg.get('activeProjectId') if not active_project_id: ap = cfg.get('activeProjects', []) if ap: active_project_id = ap[0] active_contexts = cfg.get('activeContexts', []) # FLY-1267: the project id itself is no longer printed. It is a raw # UUID — the agent cannot act on it, and every script that needs it # reads it from config directly. Only its absence is actionable. # It is still written to the attribution log (FLY-1054), which is # what AI Spend actually joins on. if (not active_project_id and not active_contexts and cfg.get('tier') == 'cloud' and cfg.get('setupComplete')): context_parts.append('[No active project/board. Run /start-session to set one]') # FLY-692: Board-aware context — show board type + sprint for agents if active_contexts: primary = active_contexts[0] board_type = primary.get('boardType', '') # Canonical: id/name. Compat: boardId/boardName (stored before fix) board_name = primary.get('name') or primary.get('boardName', '') if board_type == 'scrum': sprint_name = primary.get('sprintName') or primary.get('sprintId') if sprint_name: context_parts.append(f'Board: {board_name} (Scrum) | Sprint: {sprint_name}') else: context_parts.append(f'Board: {board_name} (Scrum) | No active sprint') elif board_type == 'kanban': context_parts.append(f'Board: {board_name} (Kanban)') elif board_name: context_parts.append(f'Board: {board_name}') # Note secondary contexts briefly if len(active_contexts) > 1: others = [c.get('name') or c.get('boardName', '?') for c in active_contexts[1:]] context_parts.append(f'Also: {", ".join(others)}') # FLY-1267: the type->label-ID map used to be printed here on every # turn. It was unusable by construction — the ids were truncated to # 8 characters — and `issues.py create` resolves the category label # from the issue type itself. Pure per-turn cost, no reader. except (json.JSONDecodeError, OSError, IOError): pass # Rich issue context (consolidated: ID, status, AC, assignment, nudge) issue_line = get_issue_context_line(repo_dir if is_workspace else None) if issue_line: context_parts.append(issue_line) # Setup completion nudge OR onboard nudge OR config freshness nudge setup_nudge = get_setup_nudge() if setup_nudge: context_parts.append(setup_nudge) else: # Check if onboarding has been completed (from repo config) try: config_data = json.loads(config_file.read_text()) if config_file.exists() else {} if config_data.get('setupComplete') and not config_data.get('onboardComplete'): context_parts.append('[Run /onboard to get oriented to this project]') except (json.JSONDecodeError, OSError, IOError): pass freshness_nudge = get_config_freshness_nudge(session_id) if freshness_nudge: context_parts.append(freshness_nudge) # Integrity drift check (TTL-cached, runs at most every 30 min) drift = check_integrity_drift() if drift: context_parts.append(drift) # FLY-1267: the FlyDocs version was appended here every turn. It is static # for the life of an install, sits in `.flydocs/version` when anything needs # it, and no agent decision depends on it. # Output status line context = ' | '.join(context_parts) debug_log(f'Final CONTEXT: {context}') if context: debug_log(f'Outputting plain text context: {context}') print(context) # Workflow directive — tells the agent what it MUST do based on current state issue_id, status, focus_warning = get_issue_context( repo_dir if is_workspace else None ) if focus_warning: debug_log(f'Focus warning: {focus_warning}') print(focus_warning) # Correlated attribution tuple for AI spend apportioning (FLY-1013 issue, # FLY-1054 provider project). append_usage_attribution( session_id, issue_id, repo_dir, get_active_project_id(repo_dir if is_workspace else None), ) directive = get_workflow_directive(status, has_issue=issue_id is not None) if directive: debug_log(f'Workflow directive: {directive}') print(directive) # Capture directive — when the prompt looks like issue creation if prompt and not issue_id: prompt_lower = prompt.lower() capture_signals = ['capture', 'log a bug', 'new issue', 'add to backlog', 'found a bug', 'new idea', 'quick capture'] if any(signal in prompt_lower for signal in capture_signals): print('[REQUIRED: Read stages/capture.md for the full capture procedure including templates and label config]') # FLY-1267: the orientation block (last-session counts, topology, product # title) was printed on every turn. session-start.py already delivers # last-session state and the repo list once per session, and topology and # product live in CLAUDE.md / AGENTS.md / project.md — so this was the same # standing facts re-paid for on turn 2 through turn 60. debug_log('=== Hook completed successfully ===') sys.exit(0) if __name__ == '__main__': main()