ce-compound — Skillopedia

/ce-compound Outcome: the lesson of one qualifying solved problem is held in the cheapest form that stops the repeat, grounded against the current tree, discoverable by the next agent. Done: the chosen form is written or updated and validated, vocabulary capture recorded, and the mode's completion report emitted; when no lesson qualifies, nothing is written and the report says why. One learning per run. A session that produced several gets several sequential runs, never one batched run. Read ; it explains what batching breaks. Preconditions Document only a problem that is solved and verified.…

\n description: \"Date documented (YYYY-MM-DD)\"\n\n problem_type:\n type: enum\n values:\n - build_error\n - test_failure\n - runtime_error\n - performance_issue\n - database_issue\n - security_issue\n - ui_bug\n - integration_issue\n - logic_error\n - developer_experience\n - workflow_issue\n - best_practice\n - documentation_gap\n - architecture_pattern\n - design_pattern\n - tooling_decision\n - convention\n description: \"Primary category — determines track (bug vs knowledge). Prefer the narrowest applicable value; best_practice is the fallback when no narrower knowledge-track value fits.\"\n\n component:\n type: string\n suggested_values:\n - data_model\n - api_layer\n - service_layer\n - background_job\n - database\n - frontend\n - messaging\n - infrastructure\n - observability\n - authentication\n - payments\n - development_workflow\n - testing_framework\n - documentation\n - tooling\n description: \"Component or area involved. Open vocabulary: choose it per the corpus-first rule in yaml-schema.md (corpus value for this area, most-used spelling when the corpus disagrees, suggested_values only when no doc covers the area; corpus values reused as spelled). A newly chosen suggested_values fallback is lowercase, underscore-separated.\"\n\n severity:\n type: enum\n values:\n - critical\n - high\n - medium\n - low\n description: \"Impact severity\"\n\n# --- Track-specific rules ----------------------------------------------------\ntrack_rules:\n bug:\n required:\n symptoms:\n type: array[string]\n min_items: 1\n max_items: 5\n description: \"Observable symptoms such as errors or broken behavior\"\n root_cause:\n type: string\n suggested_values:\n - wrong_api\n - data_integrity\n - concurrency\n - async_timing\n - memory_leak\n - config_error\n - logic_error\n - test_isolation\n - missing_validation\n - missing_permission\n - missing_workflow_step\n - inadequate_documentation\n - missing_tooling\n - incomplete_setup\n description: \"Fundamental technical cause of the problem. Open vocabulary, corpus-first rule as for component, matched by the cause itself rather than the area.\"\n resolution_type:\n type: enum\n values:\n - code_fix\n - migration\n - config_change\n - test_fix\n - dependency_update\n - environment_setup\n - workflow_improvement\n - documentation_update\n - tooling_addition\n - seed_data_update\n description: \"Type of fix applied\"\n\n knowledge:\n optional:\n applies_when:\n type: array[string]\n max_items: 5\n description: \"Conditions or situations where this guidance applies\"\n symptoms:\n type: array[string]\n max_items: 5\n description: \"Observable gaps or friction that prompted this guidance (optional for knowledge track)\"\n root_cause:\n type: string\n suggested_values:\n - wrong_api\n - data_integrity\n - concurrency\n - async_timing\n - memory_leak\n - config_error\n - logic_error\n - test_isolation\n - missing_validation\n - missing_permission\n - missing_workflow_step\n - inadequate_documentation\n - missing_tooling\n - incomplete_setup\n description: \"Underlying cause, if there is a specific one (optional for knowledge track). Open vocabulary, corpus-first rule as for component, matched by the cause itself rather than the area.\"\n resolution_type:\n type: enum\n values:\n - code_fix\n - migration\n - config_change\n - test_fix\n - dependency_update\n - environment_setup\n - workflow_improvement\n - documentation_update\n - tooling_addition\n - seed_data_update\n description: \"Type of change, if applicable (optional for knowledge track)\"\n\n# --- Fields optional for BOTH tracks ----------------------------------------\noptional_fields:\n related_components:\n type: array[string]\n description: \"Other components involved\"\n\n tags:\n type: array[string]\n max_items: 8\n description: \"Search keywords, lowercase and hyphen-separated\"\n\n retire_when:\n type: string\n description: \"The change outside the repo that would retire this learning, and how a refresh can check for it without changing the repo, e.g. \\\"example-org/sdk#1234 closed as fixed in a released sdk version; check the issue and the sdk release notes\\\". Set it only when the guidance holds just while something outside the repo holds: an open upstream bug, a tool or platform version, a dependency's behavior. Omit it otherwise. A verification that needs a repo change, such as removing a workaround and rerunning CI, belongs in the body as the step before removal.\"\n\n# --- Fields optional for bug track only -------------------------------------\nbug_optional_fields:\n framework_version:\n type: string\n description: \"Framework or runtime name and version the bug was observed on, e.g. \\\"rails 7.1.2\\\" or \\\"node 22.4.0\\\". Only relevant for bug-track docs.\"\n\n# --- Backward compatibility --------------------------------------------------\n# Docs created before the track system was introduced may have bug-track\n# fields (symptoms, root_cause, resolution_type) on knowledge-type\n# problem_types. These are valid legacy docs:\n# - Bug-track fields present on a knowledge-track doc are harmless. Do not\n# strip them during refresh unless the doc is being rewritten for other reasons.\n# - When creating NEW docs, follow the track rules above.\n# Docs written before component/root_cause became open vocabulary may carry\n# values from the earlier closed list (rails_*, hotwire_turbo, missing_include,\n# ...) or a rails_version field. They stay valid; the corpus-first rule means a\n# repo whose docs consistently use those values keeps using them.\n\n# --- Validation rules --------------------------------------------------------\nvalidation_rules:\n - \"Determine track from problem_type using the tracks section above\"\n - \"All shared required_fields must be present\"\n - \"Bug-track required fields (symptoms, root_cause, resolution_type) must be present on bug-track docs\"\n - \"Knowledge-track docs have no additional required fields beyond the shared ones\"\n - \"Bug-track fields on existing knowledge-track docs are harmless (see backward compatibility note)\"\n - \"Track-specific optional fields may be included but are not required\"\n - \"Enum fields (problem_type, severity, resolution_type) must match allowed values exactly\"\n - \"Open-vocabulary fields (component, root_cause) follow the corpus-first rule in yaml-schema.md: component is matched by area (the value existing docs under \u003croot>/solutions/ use for that area, most-used spelling when existing docs disagree), root_cause is matched by the cause itself (the value existing docs use for that same cause, most-used spelling when existing docs disagree), suggested_values only when no existing doc covers the area (component) or names the cause (root_cause). Do not coin a near-synonym of a value the corpus already uses\"\n - \"Array fields must respect min_items/max_items when specified\"\n - \"date must match YYYY-MM-DD format\"\n - \"framework_version, if provided, only applies to bug-track docs\"\n - \"retire_when is optional on both tracks; when present it is one string, and a doc without it stays valid\"\n - \"tags should be lowercase and hyphen-separated\"\n - \"Array-of-strings frontmatter items (symptoms, applies_when, tags, related_components, or any future array field) must be wrapped in double quotes when the value starts with a YAML reserved indicator (`, [, *, &, !, |, >, %, @, ?) or contains the substring `: ` — otherwise strict YAML parsers reject the file\"\n","content_type":"application/yaml; charset=utf-8","language":"yaml","size":9756,"content_sha256":"3adbcdcebe83593830769ca7b22103e72df82bda3b626f503d8fa2a6b20e8b97"},{"filename":"references/session-history.md","content":"# Session history (Phase 1 step 4)\n\nSession history is the final Phase 1 input, not a workflow stop. When it returns, including with \"no relevant prior sessions\", go straight to the form choice without pausing or summarizing.\n\n## Session context\n\nResolve two values at runtime with the shell tool before Phase 1 session-history filtering. Run each as its own command and read its exit status. A non-zero exit is a normal state here, not an error to route around:\n\n- **Git branch**: run `git rev-parse --abbrev-ref HEAD`. Use the branch name to filter session history in Phase 1. If it returns `HEAD` (detached) or exits non-zero (not a git repo), skip branch filtering.\n- **Repo root**: run `git rev-parse --show-toplevel`. Use it as the session-history repo filter in Phase 1. If it exits non-zero (not a git repo), fall back to the working directory.\n\n#### 4. **Session History** (internal flow after dispatching the Related Docs Finder; automatic in Full mode, including non-interactive)\n - This is a two-stage probe: the cheap discovery+metadata pass below always executes, and the expensive extraction+synthesis executes only when the probe clears the relevance bar (see **Escalation check** below).\n - Run session discovery, branch/keyword filtering, scan-window selection, deep-dive selection, and per-session extraction directly inside this skill using `scripts/session-history/`.\n - Read the skill-local synthesis prompt at `references/agents/session-historian.md`, then dispatch a generic subagent using that prompt content. Do not dispatch a standalone agent by type/name.\n\n **Keep the session-history payload tight.** A long, keyword-rich payload invites the flow to widen its search. Use this shape:\n\n - **Session context** (only if the values resolved cleanly above; otherwise omit): repo name, current git branch.\n - **Time window**: explicit `7 days` unless the documented problem clearly spans a longer arc.\n - **Problem topic**: one sentence naming the concrete issue — error message, module name, what broke and how it was fixed. Not a paragraph; not a bullet list of related topics.\n - **Filter rule (one line)**: \"Only surface findings directly relevant to this specific problem. Ignore unrelated work from the same sessions or branches.\"\n - **Output schema**:\n\n ```\n Structure your response with these sections (omit any with no findings):\n - What was tried before\n - What didn't work\n - Key decisions\n - Related context\n ```\n\n Do not append additional context blocks, exclusion lists, or topic-keyword bullets. Verbose payloads let the session-history flow keep widening the search, which rapidly compounds wall time. If keyword search is needed, the internal flow decides that from the topic.\n - Returns a structured digest, or `no relevant prior sessions`. Either is a valid final Phase 1 input, and the documentation gets written without session context when it is the latter.\n\n **Script resolution.** Set `SKILL_DIR` to the absolute path of the directory containing the SKILL.md you just read, and run the bundled scripts from `\"$SKILL_DIR/scripts/session-history/\"`. Set `SKILL_DIR` inline in each bash block below (shell state does not persist between commands). If the bundled scripts are genuinely not present on disk under `\"$SKILL_DIR/scripts/session-history/\"`, skip session history visibly with: \"Session history bundled scripts were not found in this skill's directory; skipping the session-history probe for this run.\" Continue Phase 2 without session context.\n\n **Discovery pipeline.** Infer the scan window from the problem topic, starting with 7 days. Run discovery and metadata extraction:\n\n ```bash\n SKILL_DIR=\"\u003cabsolute path of the directory containing the SKILL.md you just read>\";\n if [ -f \"$SKILL_DIR/scripts/session-history/discover-sessions.sh\" ] && [ -f \"$SKILL_DIR/scripts/session-history/extract-metadata.py\" ]; then\n PY=\"$(for c in python3 python py; do command -v \"$c\" >/dev/null 2>&1 && \"$c\" -c '' >/dev/null 2>&1 && { echo \"$c\"; break; }; done)\"; [ -n \"$PY\" ] || { echo \"no working Python 3 interpreter on PATH\" >&2; exit 1; };\n REPO_ROOT=$(git rev-parse --show-toplevel 2>/dev/null || pwd); REPO_NAME=$(basename \"$REPO_ROOT\"); SCAN_DAYS=\"7\"; bash \"$SKILL_DIR/scripts/session-history/discover-sessions.sh\" \"$REPO_NAME\" \"$SCAN_DAYS\" --cwd \"$REPO_ROOT\" | tr '\\n' '\\0' | xargs -0 \"$PY\" \"$SKILL_DIR/scripts/session-history/extract-metadata.py\" --cwd-filter \"$REPO_ROOT\";\n else echo \"Session history bundled scripts were not found in this skill's directory; skipping the session-history probe for this run.\"; fi\n ```\n\n The command prints JSONL. `xargs` may split a long file list across several runs of the script; each run prints one object per kept session, then its own `_meta` line. Parse the output one line at a time, because the stream as a whole is not one JSON document. A pass's `files_processed` total is the sum across its `_meta` lines; a `--keyword` re-run is a new pass, not more files. Optional fields are omitted, not set to null, so a missing key is not an error. Rank recency by `last_ts`, falling back to `ts`. If you process the output with code, look at a few lines first and use the keys you see.\n\n Claude sessions live under `~/.claude/projects/` unless `CLAUDE_CONFIG_DIR` is set, in which case they live under that directory's `projects/`. Codex sessions live under `~/.codex/sessions/` unless `CODEX_HOME` is set, in which case they live under that directory's `sessions/`; `~/.agents/sessions/` is also scanned. Pi sessions are included when present under `~/.pi/agent/sessions/` (`PI_CODING_AGENT_DIR` / `PI_CODING_AGENT_SESSION_DIR` override), and oh-my-pi (`omp`) sessions under `~/.omp/agent/sessions/` (named profiles: `~/.omp/profiles/\u003cname>/agent/sessions/`; XDG: `$XDG_DATA_HOME/omp/sessions` and `$XDG_DATA_HOME/omp/profiles/\u003cname>/sessions`; `PI_CONFIG_DIR` / `PI_CODING_AGENT_DIR` / `PI_CODING_AGENT_SESSION_DIR` override). Claude, Codex, Pi, and oh-my-pi (`omp`) sessions carry a recorded `cwd` (Claude also has git branch; Pi/omp do not). `--cwd-filter` keeps a session when that cwd is the repo root, an ancestor of it, or a path inside it (path-component boundaries: `my-repo` vs `my-repo-old` do not match). Claude, Codex, Pi, and omp sessions with no recorded cwd are dropped. Cursor has no cwd and stays. Do not drop a Claude session because its project folder name lacks the repo basename. That is how Claude stores parent-started sessions. If the first pass's `files_processed` total is `0`, return `no relevant prior sessions`. If the first pass finds no relevant branch matches, or if processing Codex, Pi, or oh-my-pi (`omp`) sessions, derive 2-4 keywords from the topic and re-run metadata extraction with `--keyword K1,K2,...`. Keep at most 5 sessions across Claude Code, Codex, Cursor, Pi, and oh-my-pi (`omp`), ranked by branch match, keyword match count, file size over 30KB, and recency. Exclude the current session.\n\n **Escalation check.** The discovery+metadata pass above is the cheap probe and always runs in Full mode. Escalate to the extraction and synthesis stages below **only** when at least one retained candidate clears the relevance bar: a current-branch match, or ≥2 topic-keyword matches. If no candidate clears the bar (including a `files_processed` total of `0`), stop here, record `no relevant prior sessions` as the session-history input, and skip extraction and synthesis. This check is what keeps the always-on probe cheap. The expensive synthesis runs only when a prior session is genuinely relevant.\n\n **Extraction pipeline.** Create `SCRATCH=$(mktemp -d \"${TMPDIR:-/tmp}/ce-compound-sessions-XXXXXX\")`. For each selected session, write extracted content to scratch files:\n\n ```bash\n SKILL_DIR=\"\u003cabsolute path of the directory containing the SKILL.md you just read>\";\n if [ -f \"$SKILL_DIR/scripts/session-history/extract-skeleton.py\" ]; then\n PY=\"$(for c in python3 python py; do command -v \"$c\" >/dev/null 2>&1 && \"$c\" -c '' >/dev/null 2>&1 && { echo \"$c\"; break; }; done)\"; [ -n \"$PY\" ] || { echo \"no working Python 3 interpreter on PATH\" >&2; exit 1; };\n \"$PY\" \"$SKILL_DIR/scripts/session-history/extract-skeleton.py\" --output \"$SCRATCH/\u003csession-id>.skeleton.txt\" \u003c \u003csession-file>;\n else echo \"Session history bundled scripts were not found in this skill's directory; skipping the session-history probe for this run.\"; fi\n ```\n\n Use `extract-errors.py` selectively when dead ends or recurring errors are likely useful. Pass only the scratch file paths and metadata to the synthesis subagent.\n\n **Synthesis dispatch.** Build a generic subagent prompt containing:\n - the full content of `references/agents/session-historian.md`\n - `problem_topic`\n - `scratch_dir`\n - a `sessions` array with extracted file paths and metadata\n - the output schema above\n - the filter rule above\n\n The subagent reads only the scratch paths, **writes its prose findings to `{run_dir}/session-history.md`, and returns only that artifact path once the write is confirmed** (the same #956 reliability reason applies, because session-history findings are long-form prose prone to summary-collapse). If `{run_id}` did not resolve or the artifact write failed, it returns the prose inline instead (per the inline-fallback rule above). If synthesis fails, note the failure and continue without session context.\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":9434,"content_sha256":"c65fbb9ca4cc0ba2a77eca9d41bec65e352ab6e0f1ae251f3e9b9463baf6f662"},{"filename":"references/yaml-schema.md","content":"# YAML Frontmatter Schema\n\n`schema.yaml` in this directory is the canonical contract for `\u003croot>/solutions/` frontmatter written by `ce-compound`.\n\nUse this file as the quick reference for:\n- required fields\n- enum values and open-vocabulary defaults\n- validation expectations\n- category mapping\n- track classification (bug vs knowledge)\n\n## Corpus-First Vocabulary\n\n`component` and `root_cause` are open vocabulary, and the category directories are a default layout, not a fixed one. Repos that already hold learnings usually have their own self-consistent values and directory names, and their retrieval (agent instructions, greps, tooling) is keyed on those. A doc written to this file's defaults instead of the repo's vocabulary is a doc the repo cannot find, which defeats the point of writing it.\n\nSo, before classifying: when `\u003croot>/solutions/` already contains docs, sample their frontmatter (`component`, `root_cause`, `problem_type`, `tags`) and directory names for this area. For `component`, use the value the corpus's existing docs use for this area (the component/directory grouping this change belongs to). For `root_cause`, match by the cause itself rather than the area: use the value existing docs use for this same underlying cause, wherever in the corpus those docs live — an area's typical root_cause does not carry over to a doc whose verified cause is different. For either field, when the corpus has more than one spelling for the same area (component) or the same cause (root_cause), use the spelling the most docs use (tie: the most recently dated doc); use the schema's suggested default only when no existing doc covers the area (component) or names the cause (root_cause). Reuse corpus values as spelled — do not coin a near-synonym of a value the corpus already uses. Directory choice is a separate condition: file the doc in the existing directory that covers this area; use the Category Mapping below only when no existing directory covers it. `problem_type`, `severity`, and `resolution_type` remain closed enums (`problem_type` drives track selection).\n\n## Tracks\n\nThe `problem_type` determines which **track** applies. Each track has different required and optional fields.\n\n| Track | problem_types | Description |\n|-------|--------------|-------------|\n| **Bug** | `build_error`, `test_failure`, `runtime_error`, `performance_issue`, `database_issue`, `security_issue`, `ui_bug`, `integration_issue`, `logic_error` | Defects and failures that were diagnosed and fixed |\n| **Knowledge** | `best_practice`, `documentation_gap`, `workflow_issue`, `developer_experience`, `architecture_pattern`, `design_pattern`, `tooling_decision`, `convention` | Practices, patterns, conventions, decisions, workflow improvements, and documentation. Prefer the narrowest applicable value; `best_practice` is the fallback. |\n\n## Required Fields (both tracks)\n\n- **module**: Module or area affected\n- **date**: ISO date in `YYYY-MM-DD`\n- **problem_type**: One of the values listed in the Tracks table above\n- **component**: Component or area involved — open vocabulary (see Corpus-First Vocabulary). Suggested defaults when the corpus has no value for the area: `data_model`, `api_layer`, `service_layer`, `background_job`, `database`, `frontend`, `messaging`, `infrastructure`, `observability`, `authentication`, `payments`, `development_workflow`, `testing_framework`, `documentation`, `tooling`\n- **severity**: One of `critical`, `high`, `medium`, `low`\n\n## Bug Track Fields\n\nRequired:\n- **symptoms**: YAML array with 1-5 observable symptoms (errors, broken behavior)\n- **root_cause**: Fundamental technical cause — open vocabulary (see Corpus-First Vocabulary). Suggested defaults: `wrong_api`, `data_integrity`, `concurrency`, `async_timing`, `memory_leak`, `config_error`, `logic_error`, `test_isolation`, `missing_validation`, `missing_permission`, `missing_workflow_step`, `inadequate_documentation`, `missing_tooling`, `incomplete_setup`\n- **resolution_type**: One of `code_fix`, `migration`, `config_change`, `test_fix`, `dependency_update`, `environment_setup`, `workflow_improvement`, `documentation_update`, `tooling_addition`, `seed_data_update`\n\n## Knowledge Track Fields\n\nNo additional required fields beyond the shared ones. All fields below are optional:\n\n- **applies_when**: Conditions or situations where this guidance applies\n- **symptoms**: Observable gaps or friction that prompted this guidance\n- **root_cause**: Underlying cause, if there is a specific one\n- **resolution_type**: Type of change, if applicable\n\n## Optional Fields (both tracks)\n\n- **related_components**: Other components involved\n- **tags**: Search keywords, lowercase and hyphen-separated\n- **retire_when**: The change outside the repo that would retire this learning, and how a refresh can check for it without changing the repo. Set it only when the guidance holds just while something outside the repo holds: an open upstream bug, a tool or platform version, a dependency's behavior. Omit it otherwise. A verification that needs a repo change, such as removing a workaround and rerunning CI, belongs in the body as the step before removal. It is one string; quote the whole value when it contains ` #` or `: `.\n\n## Optional Fields (bug track only)\n\n- **framework_version**: Framework or runtime name and version the bug was observed on, e.g. `rails 7.1.2` or `node 22.4.0`\n\n## Backward Compatibility\n\nDocs created before the track system may have `symptoms`/`root_cause`/`resolution_type` on knowledge-type problem_types. These are valid legacy docs:\n\n- Bug-track fields present on a knowledge-track doc are harmless. Do not strip them during refresh unless the doc is being rewritten for other reasons.\n- Docs written before `component`/`root_cause` became open vocabulary may carry values from the earlier closed list or the earlier `rails_version` field (now `framework_version`). They stay valid, and a corpus that consistently uses them keeps using them.\n- When creating **new** docs, follow the track rules above.\n\n## Category Mapping\n\nDefault layout for a repo with no existing learnings. When `\u003croot>/solutions/` already has an established directory taxonomy, place the doc in the existing directory that covers this area (the corpus-first rule above) rather than creating a new directory from this table that nothing else uses.\n\n- `build_error` -> `\u003croot>/solutions/build-errors/`\n- `test_failure` -> `\u003croot>/solutions/test-failures/`\n- `runtime_error` -> `\u003croot>/solutions/runtime-errors/`\n- `performance_issue` -> `\u003croot>/solutions/performance-issues/`\n- `database_issue` -> `\u003croot>/solutions/database-issues/`\n- `security_issue` -> `\u003croot>/solutions/security-issues/`\n- `ui_bug` -> `\u003croot>/solutions/ui-bugs/`\n- `integration_issue` -> `\u003croot>/solutions/integration-issues/`\n- `logic_error` -> `\u003croot>/solutions/logic-errors/`\n- `developer_experience` -> `\u003croot>/solutions/developer-experience/`\n- `workflow_issue` -> `\u003croot>/solutions/workflow-issues/`\n- `best_practice` -> `\u003croot>/solutions/best-practices/`\n- `documentation_gap` -> `\u003croot>/solutions/documentation-gaps/`\n- `architecture_pattern` -> `\u003croot>/solutions/architecture-patterns/`\n- `design_pattern` -> `\u003croot>/solutions/design-patterns/`\n- `tooling_decision` -> `\u003croot>/solutions/tooling-decisions/`\n- `convention` -> `\u003croot>/solutions/conventions/`\n\n## Validation Rules\n\n1. Determine the track from `problem_type` using the Tracks table.\n2. All shared required fields must be present.\n3. Bug-track required fields (`symptoms`, `root_cause`, `resolution_type`) must be present on bug-track docs.\n4. Knowledge-track docs have no additional required fields beyond the shared ones.\n5. Bug-track fields on existing knowledge-track docs are harmless (see Backward Compatibility).\n6. Enum fields (`problem_type`, `severity`, `resolution_type`) must match the allowed values exactly.\n7. Open-vocabulary fields (`component`, `root_cause`) follow the Corpus-First Vocabulary rule above.\n8. Array fields must respect min/max item counts.\n9. `date` must match `YYYY-MM-DD`.\n10. `framework_version`, if present, only applies to bug-track docs.\n11. `retire_when` is optional on both tracks. When present it is one string, and a doc without it stays valid.\n\n## YAML Safety Rules\n\nStrict YAML 1.2 parsers (`yq`, `js-yaml` strict, PyYAML) reject array items\nthat start with a reserved indicator character as unquoted scalars. When\nwriting items for any array-of-strings field (`symptoms`, `applies_when`,\n`tags`, `related_components`, or any future array field), wrap the value in\ndouble quotes if it starts with any of:\n\n`` ` ``, `[`, `*`, `&`, `!`, `|`, `>`, `%`, `@`, `?`\n\nAlso quote if the value contains the substring `\": \"` — that punctuation\nconfuses flow-style parsers.\n\nExample — before (breaks strict YAML):\n\n symptoms:\n - `sudo dscacheutil -flushcache` does not restore in-container mDNS\n\nExample — after (parses cleanly):\n\n symptoms:\n - \"`sudo dscacheutil -flushcache` does not restore in-container mDNS\"\n\nThis rule applies to all array-of-strings frontmatter fields. Scalar string\nfields like `description:` have their own quoting rules (see plugin\n`AGENTS.md` under \"YAML Frontmatter\").\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":9203,"content_sha256":"b9c7027253398bfeaee503bde228f0b9279a047d01369cca01719b43210efd9a"},{"filename":"scripts/packs-resolve.py","content":"#!/usr/bin/env python3\n\"\"\"Resolve the Compound Packs declared in this repo's CE config into pack roots.\n\nReads the `packs:` list from `\u003crepo-root>/.compound-engineering/config.yaml`\nand `config.local.yaml` (both layers concatenate; local adds, never replaces),\nvalidates each entry, resolves path and git sources, enumerates the packs each\nsource publishes, applies selection, and prints one JSON object to stdout:\n\n {\"roots\": [{\"id\": \"...\", \"dir\": \"/abs/path\", \"nested_rule_shaped\": 0}],\n \"warnings\": [...], \"errors\": [...], \"entries\": \u003cnumber of parsed packs entries>}\n\n`nested_rule_shaped` counts the rule-shaped `.md` files one level below the\npack's top level. Discovery never reads them (subdirectories are storage), so\nthe count is informational and is reported by the health check, not warned.\n\nExit 0 whenever resolution ran (per-entry failures are data in `errors` /\n`warnings`); non-zero only when the resolver itself cannot run. Consumers treat\n`errors` as loud per-entry configuration problems and `warnings` as degraded\navailability (e.g. an unreachable git source skipped per the warn-and-continue\ncontract).\n\n`--declared-only` answers the config-only question without touching git or the\ncache: it parses both config layers, shape-checks each entry, and prints\n\n {\"declared\": \u003cbool|null>, \"entries\": \u003cn>, \"errors\": [...]}\n\nwhere `declared` is true when any entry parsed or the `packs:` block is\nmalformed (a broken declaration is still a declaration the consumer should\nsurface), false when neither config layer names packs, and null when no\nrepository -- and so no config -- could be located.\n\nEntry shape (documented subset -- anything else under `packs:` is a loud error):\n\n packs:\n - source: packs/local-rules # repo-relative path\n - source: ~/packs/kk-style # ~ or absolute path\n - source: https://github.com/o/r # git URL: ref required\n ref: v1.2.0 # tag, sha, or branch\n path: packs # optional subfolder (git only)\n pack: [rails, inertia] # one id, a list, or omit = all\n id: rails-core # rename (single-pack entries)\n - source: https://github.com/o/r/tree/main/packs # tree-URL sugar\n\nGit sources cache under `\u003cscratch-root>/ce-packs/\u003csha256(url\\\\nref)>` with an\natomic temp-clone-then-rename, so a keyed path's existence proves a complete\nclone. All git subprocesses run non-interactively (GIT_TERMINAL_PROMPT=0, ssh\nBatchMode, bounded timeout): missing credentials degrade to a warning, never a\nhang. A missing `git` binary degrades git sources only (each warns and is\nskipped); path sources still resolve, with the repository located by walking up\nfrom the working directory to the nearest `.git` entry. Environment overrides:\nCE_PACKS_CACHE_ROOT (cache base for tests), CE_PACKS_GIT_TIMEOUT (seconds,\ndefault 60).\n\nA published pack is a directory a consumer lists and reads itself, so nothing\nin it may link outside its source: a pack whose tree holds such a link is not\npublished (a loud per-entry error), never trimmed a file at a time.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport functools\nimport hashlib\nimport json\nimport os\nimport re\nimport shutil\nimport stat\nimport subprocess\nimport sys\nimport tempfile\n\nIS_WINDOWS = os.name == \"nt\"\n_uid_getter = getattr(os, \"geteuid\", None) or getattr(os, \"getuid\", None)\n_EFFECTIVE_UID = _uid_getter() if _uid_getter is not None else None\nGIT_TIMEOUT = float(os.environ.get(\"CE_PACKS_GIT_TIMEOUT\") or 60)\n\nCONFIG_FILES = (\"config.yaml\", \"config.local.yaml\")\nKNOWN_KEYS = {\"source\", \"ref\", \"path\", \"pack\", \"id\"}\nSCALAR_KEYS = (\"source\", \"ref\", \"path\", \"id\") # every known key but the list-valued `pack`\n_TREE_URL_RE = re.compile(\n r\"^(?P\u003cbase>https?://github\\.com/[^/\\s]+/[^/\\s]+?)(?:\\.git)?/tree/(?P\u003cref>[^/\\s]+)(?:/(?P\u003cpath>[^\\s]*))?/?$\"\n)\n\n\ndef _is_git_url(source: str) -> bool:\n return bool(\n re.match(r\"^(https?|ssh|git|file)://\", source) or re.match(r\"^[\\w.-]+@[\\w.-]+:\", source)\n )\n\n\ndef _within(path: str, parent: str) -> bool:\n \"\"\"True when `path` is `parent` or lies below it (both already realpath'd).\"\"\"\n return path == parent or path.startswith(parent + os.sep)\n\n\n# --- scratch root (peer-job-runner shape: probe /tmp, fall back to TMPDIR) ---\n\ndef _owned_dir(path: str) -> bool:\n \"\"\"Directory, not a symlink, owned by the effective uid (POSIX).\"\"\"\n try:\n st = os.lstat(path)\n except OSError:\n return False\n if not stat.S_ISDIR(st.st_mode):\n return False\n if _EFFECTIVE_UID is not None and st.st_uid != _EFFECTIVE_UID:\n return False\n return True\n\n\ndef _private_root_usable(path: str) -> bool:\n \"\"\"Create or repair `path` as a private (0700) root owned by this user.\"\"\"\n try:\n os.mkdir(path, 0o700)\n except FileExistsError:\n pass\n except OSError:\n return False\n if not IS_WINDOWS:\n if not _owned_dir(path):\n return False\n try:\n os.chmod(path, 0o700)\n except OSError:\n return False\n return os.path.isdir(path) and os.access(path, os.W_OK)\n\n\ndef _trusted_checkout(path: str) -> bool:\n \"\"\"A real directory (never a symlink) that this user owns; ownership is POSIX-only.\"\"\"\n if os.path.islink(path) or not os.path.isdir(path):\n return False\n return IS_WINDOWS or _owned_dir(path)\n\n\[email protected]_cache(maxsize=None)\ndef cache_base() -> str | None:\n configured = os.environ.get(\"CE_PACKS_CACHE_ROOT\")\n if configured:\n root = os.path.abspath(configured)\n os.makedirs(root, exist_ok=True)\n return root\n if IS_WINDOWS:\n base = os.environ.get(\"LOCALAPPDATA\") or tempfile.gettempdir()\n root = os.path.join(base, \"compound-engineering-packs\")\n return root if _private_root_usable(root) else None\n if _EFFECTIVE_UID is None:\n return None\n for base in (\"/tmp\", os.environ.get(\"TMPDIR\") or \"/tmp\"):\n root = os.path.join(base, f\"compound-engineering-{_EFFECTIVE_UID}\")\n if _private_root_usable(root):\n packs = os.path.join(root, \"ce-packs\")\n if _private_root_usable(packs):\n return packs\n return None\n\n\n# --- minimal YAML reader for the documented packs: subset --------------------\n\ndef _strip_comment(line: str) -> str:\n \"\"\"Drop a trailing comment (a # preceded by whitespace, outside quotes).\n\n A quote toggles quoted state only when it opens a value (start of line or\n after `: `/`- `/`[`/`,`) or closes one it opened -- a mid-word apostrophe\n (``it's``) is ordinary content and must not absorb a later comment.\n \"\"\"\n out, quote = [], \"\"\n for i, ch in enumerate(line):\n prev = line[i - 1] if i else \" \"\n if quote:\n if ch == quote:\n quote = \"\"\n elif ch in \"'\\\"\" and prev in \" \\t[,:\":\n quote = ch\n elif ch == \"#\" and prev in \" \\t\":\n break\n out.append(ch)\n return \"\".join(out).rstrip()\n\n\ndef _scalar(raw: str):\n raw = raw.strip()\n if len(raw) >= 2 and raw[0] == raw[-1] and raw[0] in \"'\\\"\":\n return raw[1:-1]\n if raw.lower() in (\"true\", \"false\"):\n return raw.lower() == \"true\"\n return raw\n\n\ndef _parse_value(raw: str):\n raw = raw.strip()\n if raw.startswith(\"[\") and raw.endswith(\"]\"):\n inner = raw[1:-1].strip()\n return [] if not inner else [_scalar(part) for part in inner.split(\",\")]\n return _scalar(raw)\n\n\ndef parse_packs_block(path: str, errors: list) -> list:\n \"\"\"Return the entry dicts under this file's top-level `packs:` key.\"\"\"\n if not os.path.isfile(path):\n return []\n with open(path, encoding=\"utf-8-sig\", errors=\"replace\") as fh:\n lines = fh.read().splitlines()\n entries, in_packs, current, pending_list_key = [], False, None, None\n for lineno, raw in enumerate(lines, 1):\n line = _strip_comment(raw)\n if not line.strip():\n continue\n indent = len(line) - len(line.lstrip())\n if indent == 0 and not (in_packs and line.lstrip().startswith(\"-\")):\n # A new top-level key ends the packs block; a zero-indent list item\n # (`- source: ...`) is still part of it -- YAML allows both styles.\n key, sep, rest = line.partition(\":\")\n in_packs = bool(sep) and key.strip() == \"packs\" and rest.strip() in (\"\", \"[]\")\n if sep and key.strip() == \"packs\" and not in_packs:\n # `packs: \u003cinline value>` is malformed, not absent: say so rather\n # than letting a flow list or a bare path declare nothing.\n errors.append(\n f\"{os.path.basename(path)}:{lineno}: `packs:` must be a block list of\"\n f\" `- source: ...` entries (got `{line.strip()}`)\"\n )\n current, pending_list_key = None, None\n continue\n if not in_packs:\n continue\n stripped = line.strip()\n loc = f\"{os.path.basename(path)}:{lineno}\"\n if stripped.startswith(\"- \") or stripped == \"-\":\n body = stripped[1:].strip()\n if pending_list_key and current is not None and \":\" not in body:\n current[pending_list_key].append(_scalar(body))\n continue\n current, pending_list_key = {\"_origin\": os.path.basename(path), \"_line\": lineno}, None\n entries.append(current)\n if body:\n if \":\" not in body:\n errors.append(f\"{loc}: unrecognized packs entry `{stripped}` -- expected `key: value`\")\n continue\n key, _, val = body.partition(\":\")\n _set_key(current, key.strip(), val, loc, errors)\n continue\n if current is None:\n errors.append(f\"{loc}: unrecognized line under packs: `{stripped}` -- expected a `- source: ...` entry\")\n continue\n if \":\" not in stripped:\n errors.append(f\"{loc}: unrecognized line under packs: `{stripped}` -- expected `key: value`\")\n continue\n key, _, val = stripped.partition(\":\")\n key = key.strip()\n if val.strip() == \"\" and key == \"pack\":\n current[key] = []\n pending_list_key = key\n continue\n pending_list_key = None\n _set_key(current, key, val, loc, errors)\n return entries\n\n\ndef _set_key(entry: dict, key: str, raw_val: str, loc: str, errors: list) -> None:\n if key not in KNOWN_KEYS:\n errors.append(f\"{loc}: unknown packs entry key `{key}:` -- accepted keys: {', '.join(sorted(KNOWN_KEYS))}\")\n return\n entry[key] = _parse_value(raw_val)\n\n\n# --- git ---------------------------------------------------------------------\n\[email protected]_cache(maxsize=None)\ndef _git_env() -> dict:\n env = dict(os.environ)\n env[\"GIT_TERMINAL_PROMPT\"] = \"0\"\n env[\"GIT_ASKPASS\"] = env.get(\"GIT_ASKPASS\") or \"true\"\n ssh = env.get(\"GIT_SSH_COMMAND\") or \"ssh\"\n if \"BatchMode\" not in ssh:\n env[\"GIT_SSH_COMMAND\"] = ssh + \" -o BatchMode=yes\"\n return env\n\n\ndef _run_git(args: list, cwd: str | None = None):\n return subprocess.run(\n [\"git\", *args], cwd=cwd, env=_git_env(), timeout=GIT_TIMEOUT,\n capture_output=True, text=True,\n )\n\n\ndef resolve_git_source(url: str, ref: str, warnings: list, label: str) -> str | None:\n \"\"\"Return the cached checkout dir for url@ref, cloning on miss. None = warn+skip.\"\"\"\n if shutil.which(\"git\") is None:\n warnings.append(f\"{label}: git binary not found; source skipped\")\n return None\n base = cache_base()\n if base is None:\n warnings.append(f\"{label}: no writable cache root for git sources; source skipped\")\n return None\n key = hashlib.sha256(f\"{url}\\n{ref}\".encode()).hexdigest()\n dest = os.path.join(base, key)\n if os.path.lexists(dest):\n if _trusted_checkout(dest):\n return dest\n warnings.append(f\"{label}: cached checkout {dest} is a symlink or not owned by this user; refetching\")\n # rmtree refuses to follow a symlink (and ignores a plain file); unlink\n # those explicitly so a planted link is removed, never its target.\n shutil.rmtree(dest, ignore_errors=True)\n if os.path.islink(dest) or os.path.isfile(dest):\n try:\n os.unlink(dest)\n except OSError:\n pass\n if os.path.lexists(dest):\n warnings.append(f\"{label}: cannot replace untrusted cached checkout {dest}; source skipped\")\n return None\n tmp = tempfile.mkdtemp(prefix=f\"{key}.part-\", dir=base)\n try:\n try:\n proc = _run_git([\"clone\", \"--quiet\", \"--depth\", \"1\", \"--no-recurse-submodules\",\n \"--branch\", ref, \"--end-of-options\", url, tmp])\n except subprocess.TimeoutExpired:\n warnings.append(f\"{label}: git clone timed out after {int(GIT_TIMEOUT)}s; source skipped\")\n return None\n if proc.returncode != 0:\n # tag/branch clone failed -- retry treating ref as a commit sha\n try:\n fetched = (\n _run_git([\"init\", \"--quiet\", tmp]).returncode == 0\n and _run_git([\"fetch\", \"--quiet\", \"--depth\", \"1\", \"--end-of-options\", url, ref], cwd=tmp).returncode == 0\n and _run_git([\"checkout\", \"--quiet\", \"FETCH_HEAD\"], cwd=tmp).returncode == 0\n )\n if not fetched:\n warnings.append(f\"{label}: cannot fetch `{ref}` from {url}; source skipped\")\n return None\n except subprocess.TimeoutExpired:\n warnings.append(f\"{label}: git fetch timed out after {int(GIT_TIMEOUT)}s; source skipped\")\n return None\n if not os.path.lexists(dest):\n try:\n os.replace(tmp, dest)\n except OSError:\n # Expected when another resolver published the same key\n # concurrently; anything else leaves no checkout to return.\n if not os.path.lexists(dest):\n warnings.append(f\"{label}: could not publish the clone to {dest}; source skipped\")\n return None\n if _trusted_checkout(dest):\n return dest\n warnings.append(\n f\"{label}: cached checkout {dest} appeared during the clone and is a symlink or not owned by this user; source skipped\"\n )\n return None\n finally:\n if os.path.isdir(tmp):\n shutil.rmtree(tmp, ignore_errors=True)\n\n\n# --- pack enumeration --------------------------------------------------------\n\n_FRONTMATTER_KEYS = (\"title:\", \"applies_when:\")\n_FRONTMATTER_CAP = 64 * 1024 # frontmatter must close within this many characters\n# A pack's README is its description, never a rule, whatever frontmatter it\n# carries: it is not published, not reported as skipped, and not counted.\n_README = \"readme.md\"\n\n\[email protected]_cache(maxsize=None)\ndef _is_knowledge_file(path: str) -> bool:\n \"\"\"True when the file opens with a `---` frontmatter block, closed by a `---`\n line within the cap, that carries `title:` and `applies_when:`. A UTF-8 BOM\n is not part of the content.\"\"\"\n try:\n with open(path, encoding=\"utf-8-sig\", errors=\"replace\") as fh:\n lines = fh.read(_FRONTMATTER_CAP).splitlines()\n except OSError:\n return False\n if not lines or lines[0].strip() != \"---\":\n return False\n for end, line in enumerate(lines[1:], 1):\n if line.strip() == \"---\":\n fm = \"\\n\".join(lines[1:end])\n break\n else:\n return False\n return all(re.search(rf\"^\\s*{re.escape(k)}\", fm, re.MULTILINE) for k in _FRONTMATTER_KEYS)\n\n\ndef _contained_md_files(directory: str, boundary: str, escaped: list) -> list:\n \"\"\"Paths of the `.md` entries directly under `directory`, minus its README,\n whose real path stays within `boundary` (a realpath). An entry that links\n outside it is appended to `escaped` and never opened, so a pack cannot read\n files off the user's machine.\"\"\"\n try:\n names = sorted(os.listdir(directory))\n except OSError:\n return []\n files = []\n for name in names:\n if not name.endswith(\".md\") or name.lower() == _README:\n continue\n child = os.path.join(directory, name)\n if _within(os.path.realpath(child), boundary):\n files.append(child)\n else:\n escaped.append(child)\n return files\n\n\ndef _contained_child_dirs(directory: str, boundary: str, escaped: list) -> list:\n \"\"\"`(name, path)` of the non-hidden directories directly under `directory`\n whose real path stays within `boundary`; one that links outside it is\n appended to `escaped` and never entered.\"\"\"\n try:\n names = sorted(os.listdir(directory))\n except OSError:\n return []\n dirs = []\n for name in names:\n child = os.path.join(directory, name)\n if name.startswith(\".\") or not os.path.isdir(child):\n continue\n if _within(os.path.realpath(child), boundary):\n dirs.append((name, child))\n else:\n escaped.append(child)\n return dirs\n\n\ndef _has_knowledge_files(directory: str, boundary: str, escaped: list) -> bool:\n return any(_is_knowledge_file(f) for f in _contained_md_files(directory, boundary, escaped))\n\n\ndef _escaping_links(root: str, boundary: str) -> list:\n \"\"\"Symlinks anywhere under `root` (walked without following links) whose real\n path leaves `boundary`, sorted. Only a link can leave: every other entry sits\n under `root`, which is already inside the boundary.\"\"\"\n leaks = []\n for dirpath, dirnames, filenames in os.walk(root, followlinks=False):\n for name in dirnames + filenames:\n child = os.path.join(dirpath, name)\n if os.path.islink(child) and not _within(os.path.realpath(child), boundary):\n leaks.append(child)\n return sorted(leaks)\n\n\ndef enumerate_packs(source_root: str, boundary: str, escaped: list, self_name: str | None = None) -> dict:\n \"\"\"Map published pack id -> dir. Immediate children only; self = single pack.\n\n `boundary` is the realpath every child directory and rule file must stay\n within (the git checkout, the repository, or the source root itself); a\n child that links outside it is recorded in `escaped` and skipped. Symlinks\n that stay inside the boundary are ordinary content.\n \"\"\"\n if _has_knowledge_files(source_root, boundary, escaped):\n name = self_name or os.path.basename(os.path.abspath(source_root))\n return {name: source_root}\n return {\n name: child\n for name, child in _contained_child_dirs(source_root, boundary, escaped)\n if _has_knowledge_files(child, boundary, escaped)\n }\n\n\ndef nested_rule_shaped(pack_dir: str, boundary: str) -> tuple:\n \"\"\"`(total, subdirs)`: how many rule-shaped files sit in the immediate\n subdirectories of `pack_dir`, and which subdirectories hold them. Scans one\n level down only, inside `boundary`, so the pack's data is never walked.\n Subdirectories are storage, so these files are counted, never published.\"\"\"\n hits, total = [], 0\n for name, child in _contained_child_dirs(pack_dir, boundary, []):\n count = sum(1 for f in _contained_md_files(child, boundary, []) if _is_knowledge_file(f))\n if count:\n hits.append(name)\n total += count\n return total, hits\n\n\ndef nested_rules_warning(pack_id: str, pack_dir: str, boundary: str) -> str | None:\n \"\"\"The warning for a pack directory with no rule at its top level whose\n subdirectories hold rule-shaped files: it registers, yet nothing can ever be\n discovered. None when the subdirectories hold none. The caller establishes\n that the top level is empty of rules; a pack with top-level rules keeps its\n nested files as storage and gets `nested_rule_shaped` instead.\"\"\"\n total, hits = nested_rule_shaped(pack_dir, boundary)\n if not hits:\n return None\n where = \", \".join(f\"`{name}/`\" for name in hits)\n return (\n f\"pack `{pack_id}` has {total} rule-shaped file(s) under {where} that discovery\"\n \" never reads -- move rules to the pack's top level (see https://everyinc.github.io/compound-engineering-plugin/guides/packs/, Pack layout)\"\n )\n\n\n# --- entry resolution --------------------------------------------------------\n\ndef _entry_label(entry: dict) -> str:\n return f\"{entry.get('_origin', 'config')}:{entry.get('_line', '?')}\"\n\n\ndef _entry_shape_ok(entry: dict, label: str, errors: list) -> bool:\n \"\"\"I/O-free checks on one parsed entry: `source:` present, and the scalar\n keys actually scalar. False after appending the error.\"\"\"\n for key in SCALAR_KEYS:\n val = entry.get(key)\n if val is not None and not isinstance(val, str):\n errors.append(f\"{label}: `{key}:` must be a single string (got {val!r})\")\n return False\n if not entry.get(\"source\"):\n errors.append(f\"{label}: entry has no `source:`\")\n return False\n return True\n\n\ndef resolve_entry(entry: dict, repo_root: str, roots: list, warnings: list, errors: list) -> None:\n label = _entry_label(entry)\n if not _entry_shape_ok(entry, label, errors):\n return\n source, ref, sub_path = entry[\"source\"], entry.get(\"ref\"), entry.get(\"path\")\n\n tree = _TREE_URL_RE.match(source)\n if tree:\n t_ref, t_path = tree.group(\"ref\"), tree.group(\"path\") or \"\"\n if isinstance(ref, str) and ref != t_ref:\n errors.append(f\"{label}: tree URL pins ref `{t_ref}` but entry says `ref: {ref}` -- remove one\")\n return\n if isinstance(sub_path, str) and sub_path.strip(\"/\") != t_path.strip(\"/\"):\n errors.append(f\"{label}: tree URL path `{t_path}` conflicts with `path: {sub_path}` -- remove one\")\n return\n source, ref, sub_path = tree.group(\"base\"), t_ref, t_path or None\n\n if _is_git_url(source):\n if not isinstance(ref, str) or not ref:\n errors.append(f\"{label}: git source `{source}` requires `ref:` (tag, sha, or branch)\")\n return\n if ref.startswith(\"-\") or source.startswith(\"-\"):\n errors.append(f\"{label}: git source/ref may not begin with `-`\")\n return\n checkout = resolve_git_source(source, ref, warnings, label)\n if checkout is None:\n if tree:\n warnings.append(\n f\"{label}: if the branch name contains `/`, tree-URL parsing splits it wrong -- use explicit `ref:` and `path:` fields\"\n )\n return\n git_meta = {\"url\": source, \"ref\": ref}\n source_root = os.path.join(checkout, sub_path) if sub_path else checkout\n real_root, real_checkout = os.path.realpath(source_root), os.path.realpath(checkout)\n if not _within(real_root, real_checkout):\n errors.append(f\"{label}: path `{sub_path}` escapes the source checkout\")\n return\n source_root, boundary = real_root, real_checkout\n if not os.path.isdir(source_root):\n errors.append(f\"{label}: path `{sub_path}` does not exist in {source}@{ref}\")\n return\n else:\n git_meta = None\n if ref is not None:\n errors.append(f\"{label}: `ref:` is only valid on git sources; path sources are read live\")\n return\n if sub_path is not None:\n errors.append(f\"{label}: `path:` is only valid on git sources; point `source:` at the directory instead\")\n return\n expanded = os.path.expanduser(source)\n if os.path.isabs(expanded):\n source_root = os.path.realpath(expanded)\n boundary = source_root\n else:\n source_root = os.path.realpath(os.path.join(repo_root, expanded))\n repo_real = os.path.realpath(repo_root)\n if not _within(source_root, repo_real) or _within(source_root, os.path.join(repo_real, \".git\")):\n errors.append(f\"{label}: repo-relative source `{source}` resolves outside the repository\")\n return\n boundary = repo_real\n if not os.path.isdir(source_root):\n errors.append(f\"{label}: source directory `{source}` does not exist\")\n return\n\n if git_meta:\n # Display name for a single-pack git source: the path: subfolder's\n # basename, else the URL's last path segment (never the cache key).\n tail = (sub_path or source).rstrip(\"/\").rsplit(\"/\", 1)[-1]\n self_name = re.sub(r\"\\.git$\", \"\", tail.split(\":\")[-1]) or None\n else:\n self_name = None\n escaped = []\n published = enumerate_packs(source_root, boundary, escaped, self_name)\n for link in escaped:\n warnings.append(\n f\"{label}: skipped `{os.path.relpath(link, source_root)}` in `{source}` -- it links outside the source\"\n )\n if not published:\n warnings.append(f\"{label}: source `{source}` publishes no packs (no directories with valid knowledge files)\")\n # Each child directory is a would-be pack with no top-level rule (one\n # with a rule would have published); say when its rules sit one level\n # too deep, so the author learns why nothing published.\n for name, child in _contained_child_dirs(source_root, boundary, []):\n nested = nested_rules_warning(name, child, boundary)\n if nested:\n warnings.append(f\"{label}: {nested}\")\n return\n\n selection = entry.get(\"pack\")\n if selection is None:\n selected = dict(published)\n else:\n wanted = selection if isinstance(selection, list) else [selection]\n if not wanted:\n warnings.append(f\"{label}: `pack:` lists no ids; nothing installed from `{source}`\")\n return\n missing = [w for w in wanted if w not in published]\n if missing:\n errors.append(\n f\"{label}: pack id(s) {', '.join(map(str, missing))} not published by `{source}`\"\n f\" -- available: {', '.join(sorted(published)) or 'none'}\"\n )\n return\n selected = {w: published[w] for w in wanted}\n\n override = entry.get(\"id\")\n if override is not None:\n if len(selected) != 1:\n errors.append(f\"{label}: `id:` override requires the entry to install exactly one pack\")\n return\n selected = {str(override): next(iter(selected.values()))}\n\n for pack_id, pack_dir in selected.items():\n # Consumers list the pack directory themselves, so a link that leaves\n # the source anywhere inside it would let pack content read files off\n # the user's machine: refuse the whole pack. (Top-level escapes were\n # already named as skipped during enumeration.)\n leaks = _escaping_links(pack_dir, boundary)\n if leaks:\n names = \", \".join(f\"`{os.path.relpath(p, pack_dir)}`\" for p in leaks)\n errors.append(f\"{label}: pack `{pack_id}` not published -- {names} link(s) outside the source\")\n continue\n for child in _contained_md_files(pack_dir, boundary, []):\n name = os.path.basename(child)\n if os.path.isfile(child) and not _is_knowledge_file(child):\n warnings.append(\n f\"{label}: skipped pack file `{pack_id}/{name}` (missing `title`/`applies_when` frontmatter)\"\n )\n # A published pack has a rule at its top level, so rule-shaped files in\n # its subdirectories are storage: counted for the health report, not warned.\n nested_total, _ = nested_rule_shaped(pack_dir, boundary)\n root = {\"id\": pack_id, \"dir\": pack_dir, \"nested_rule_shaped\": nested_total, \"_label\": label}\n if git_meta:\n root.update(git_meta)\n roots.append(root)\n\n\ndef _emit(declared_only: bool, entries: list, roots: list, warnings: list, errors: list,\n declared: bool | None = None) -> int:\n if declared_only:\n print(json.dumps({\"declared\": declared, \"entries\": len(entries), \"errors\": errors}))\n else:\n print(json.dumps({\n \"roots\": [{k: v for k, v in r.items() if not k.startswith(\"_\")} for r in roots],\n \"warnings\": warnings,\n \"errors\": errors,\n \"entries\": len(entries),\n }))\n return 0\n\n\ndef _repo_root() -> str | None:\n \"\"\"The enclosing checkout's top level: git's answer when git is on PATH,\n otherwise the nearest ancestor of the working directory holding a `.git`\n entry (a directory, or the file a worktree or submodule leaves). None when\n the working directory is not inside a checkout.\"\"\"\n if shutil.which(\"git\") is not None:\n proc = subprocess.run([\"git\", \"rev-parse\", \"--show-toplevel\"],\n stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True)\n return proc.stdout.strip() if proc.returncode == 0 else None\n current = os.getcwd()\n while True:\n if os.path.lexists(os.path.join(current, \".git\")):\n return current\n parent = os.path.dirname(current)\n if parent == current:\n return None\n current = parent\n\n\ndef _main(argv: list) -> int:\n parser = argparse.ArgumentParser(description=\"Resolve the Compound Packs declared in CE config.\")\n parser.add_argument(\n \"--declared-only\", action=\"store_true\",\n help=\"parse and shape-check the packs: entries only; no git or cache work\",\n )\n args = parser.parse_args(argv)\n warnings, errors, roots, entries = [], [], [], []\n\n repo_root = _repo_root()\n if repo_root is None:\n warnings.append(\"not inside a git repository; no CE config to read\")\n return _emit(args.declared_only, entries, roots, warnings, errors)\n cfg_dir = os.path.join(repo_root, \".compound-engineering\")\n for name in CONFIG_FILES:\n entries.extend(parse_packs_block(os.path.join(cfg_dir, name), errors))\n\n if args.declared_only:\n for entry in entries:\n _entry_shape_ok(entry, _entry_label(entry), errors)\n return _emit(True, entries, roots, warnings, errors, declared=bool(entries or errors))\n\n for entry in entries:\n try:\n resolve_entry(entry, repo_root, roots, warnings, errors)\n except Exception as exc: # one entry's surprise is that entry's error, not everyone's\n errors.append(f\"{_entry_label(entry)}: unexpected error resolving entry: {exc}\")\n\n # First declaration wins: config.yaml entries precede config.local.yaml in\n # CONFIG_FILES, so a personal pack can never displace the team's.\n by_id = {}\n final = []\n for root in roots:\n prev = by_id.get(root[\"id\"])\n if prev is not None:\n errors.append(\n f\"duplicate pack id `{root['id']}`: {root['_label']} ignored, {prev['_label']} kept\"\n \" -- rename one with `id:`\"\n )\n continue\n by_id[root[\"id\"]] = root\n final.append(root)\n return _emit(False, entries, final, warnings, errors)\n\n\ndef main() -> int:\n try:\n return _main(sys.argv[1:])\n except Exception as exc: # never a traceback: consumers need valid JSON\n print(json.dumps({\n \"roots\": [], \"warnings\": [], \"errors\": [f\"packs resolver failed unexpectedly: {exc}\"], \"entries\": 0,\n }))\n return 0\n\n\nif __name__ == \"__main__\":\n sys.exit(main())\n","content_type":"text/x-python; charset=utf-8","language":"python","size":31561,"content_sha256":"39c36946cd1a59ca564e60e4e049934831eb4bfb5cfce9de32ab6b4b522ef228"},{"filename":"scripts/session-history/discover-sessions.sh","content":"#!/usr/bin/env bash\n# Discover session files across Claude Code, Codex, Cursor, Pi, and oh-my-pi (omp).\n#\n# Usage: discover-sessions.sh \u003crepo-name> \u003cdays> [--cwd /abs/repo/root] [--platform claude|codex|cursor|pi|omp]\n#\n# Outputs one file path per line. Safe in both bash and zsh (all globs guarded).\n# Pass output to extract-metadata.py:\n# python3 extract-metadata.py --cwd-filter \u003crepo-name> $(bash discover-sessions.sh \u003crepo-name> 7)\n#\n# Arguments:\n# repo-name Folder name of the repo (e.g., \"my-repo\"). Used for directory matching.\n# days Scan window in days (e.g., 7). Files older than this are skipped.\n# --cwd Absolute repo root. Used for exact Pi encoded-CWD discovery\n# and the omp raw-bucket probe. Claude listing is unfiltered;\n# extract-metadata.py --cwd-filter matches recorded cwd.\n# --platform Restrict to a single platform. Omit to search all.\n\nset -euo pipefail\n\nREPO_NAME=\"${1:?Usage: discover-sessions.sh \u003crepo-name> \u003cdays> [--cwd /abs/repo/root] [--platform claude|codex|cursor|pi|omp]}\"\nDAYS=\"${2:?Usage: discover-sessions.sh \u003crepo-name> \u003cdays> [--cwd /abs/repo/root] [--platform claude|codex|cursor|pi|omp]}\"\nPLATFORM=\"all\"\nREPO_CWD=\"\"\n\n# Parse optional --platform flag\nshift 2\nwhile [ $# -gt 0 ]; do\n case \"$1\" in\n --cwd) REPO_CWD=\"$2\"; shift 2 ;;\n --platform) PLATFORM=\"$2\"; shift 2 ;;\n *) shift ;;\n esac\ndone\n\nencode_pi_cwd() {\n local cwd=\"${1%/}\"\n local encoded=\"${cwd//\\//-}\"\n encoded=\"${encoded#-}\"\n printf -- \"--%s--\" \"$encoded\"\n}\n\n# --- Claude Code ---\n# List every recent jsonl under \u003cconfig-dir>/projects. Folder names are an\n# undocumented encoder of session CWD; do not invert them. Repo attribution\n# is the recorded `cwd` field, applied by extract-metadata.py --cwd-filter.\n# CLAUDE_CONFIG_DIR relocates the whole config tree (official); unset -> ~/.claude.\ndiscover_claude() {\n local base=\"${CLAUDE_CONFIG_DIR:-$HOME/.claude}/projects\"\n [ -d \"$base\" ] || return 0\n find \"$base\" -mindepth 2 -maxdepth 2 -type f -name \"*.jsonl\" -mtime \"-${DAYS}\" 2>/dev/null\n}\n\n# --- Codex ---\ndiscover_codex() {\n local codex_home=\"${CODEX_HOME:-$HOME/.codex}\"\n for base in \"$codex_home/sessions\" \"$HOME/.agents/sessions\"; do\n [ -d \"$base\" ] || continue\n\n # Use mtime-based discovery (consistent with Claude/Cursor) so that\n # sessions started before the scan window but still active within it\n # are not missed.\n find \"$base\" -name \"*.jsonl\" -mtime \"-${DAYS}\" 2>/dev/null\n done\n}\n\n# --- Cursor ---\ndiscover_cursor() {\n local base=\"$HOME/.cursor/projects\"\n [ -d \"$base\" ] || return 0\n\n for dir in \"$base\"/*\"$REPO_NAME\"*/; do\n [ -d \"$dir\" ] || continue\n local transcripts=\"$dir/agent-transcripts\"\n [ -d \"$transcripts\" ] || continue\n find \"$transcripts\" -name \"*.jsonl\" -mtime \"-${DAYS}\" 2>/dev/null\n done\n}\n\n# --- Pi ---\ndiscover_pi() {\n local agent_dir=\"${PI_CODING_AGENT_DIR:-$HOME/.pi/agent}\"\n local base=\"${PI_CODING_AGENT_SESSION_DIR:-$agent_dir/sessions}\"\n [ -d \"$base\" ] || return 0\n\n # Pi's explicit session-dir override stores session files directly in the\n # supplied directory. The cwd filter later reads each header and keeps only\n # sessions for the active repo.\n if [ -n \"${PI_CODING_AGENT_SESSION_DIR:-}\" ]; then\n find \"$base\" -maxdepth 1 -name \"*.jsonl\" -mtime \"-${DAYS}\" 2>/dev/null\n if [ -z \"$REPO_CWD\" ]; then\n for dir in \"$base\"/*\"$REPO_NAME\"*/; do\n [ -d \"$dir\" ] || continue\n find \"$dir\" -maxdepth 1 -name \"*.jsonl\" -mtime \"-${DAYS}\" 2>/dev/null\n done\n fi\n return 0\n fi\n\n # Pi stores sessions under --\u003cabsolute-cwd-with-slashes-as-hyphens>--.\n # When the caller supplies an exact repo root, probe only that encoded\n # directory so sibling repos like my-repo-old never enter the pipeline.\n if [ -n \"$REPO_CWD\" ]; then\n local dir=\"$base/$(encode_pi_cwd \"$REPO_CWD\")\"\n [ -d \"$dir\" ] || return 0\n find \"$dir\" -maxdepth 1 -name \"*.jsonl\" -mtime \"-${DAYS}\" 2>/dev/null\n return 0\n fi\n\n # Fallback for direct script use without --cwd.\n for dir in \"$base\"/*\"$REPO_NAME\"*/; do\n [ -d \"$dir\" ] || continue\n find \"$dir\" -maxdepth 1 -name \"*.jsonl\" -mtime \"-${DAYS}\" 2>/dev/null\n done\n}\n\n# --- oh-my-pi (omp) ---\n# Encode omp's raw bucket name for a cwd: home-relative \"-\u003crel>\",\n# tmp-relative \"-tmp-\u003crel>\", and otherwise \"--\u003cabs>--\", with path separators\n# and \":\" encoded as \"-\" (session-paths.ts getDefaultSessionDirName /\n# encodeLegacyAbsoluteSessionDirName). This raw scheme predates the hashed\n# scheme and is current again since omp 17.2.9 (#7646 restored it and removed\n# automatic migration), so buckets in the wild use both shapes. Canonicalize\n# with physical paths so symlinked cwds resolve to the same bucket, mirroring\n# omp's resolveEquivalentPath. Prints nothing when the cwd cannot be resolved.\nencode_omp_raw_cwd() {\n local cwd canon_home canon_tmp rel\n cwd=\"$(cd \"$1\" 2>/dev/null && pwd -P)\" || return 0\n canon_home=\"$(cd \"$HOME\" 2>/dev/null && pwd -P)\" || canon_home=\"$HOME\"\n case \"$cwd\" in\n \"$canon_home\")\n printf -- '-'\n ;;\n \"$canon_home\"/*)\n rel=\"$(printf '%s' \"${cwd#\"$canon_home\"/}\" | sed 's/[/\\\\:]/-/g')\"\n printf -- '-%s' \"$rel\"\n ;;\n *)\n canon_tmp=\"$(cd \"${TMPDIR:-/tmp}\" 2>/dev/null && pwd -P)\" || canon_tmp=\"\"\n case \"$cwd\" in\n \"$canon_tmp\")\n printf -- '-tmp'\n ;;\n \"$canon_tmp\"/*)\n rel=\"$(printf '%s' \"${cwd#\"$canon_tmp\"/}\" | sed 's/[/\\\\:]/-/g')\"\n printf -- '-tmp-%s' \"$rel\"\n ;;\n *)\n rel=\"$(printf '%s' \"${cwd#/}\" | sed 's/[/\\\\:]/-/g')\"\n printf -- '--%s--' \"$rel\"\n ;;\n esac\n ;;\n esac\n}\n\ndiscover_omp() {\n local config_dir=\"${PI_CONFIG_DIR:-.omp}\"\n\n # omp's explicit session-dir override stores session files directly in the\n # supplied directory (flat), mirroring Pi's override branch. The cwd filter\n # later reads each header and keeps only sessions for the active repo.\n if [ -n \"${PI_CODING_AGENT_SESSION_DIR:-}\" ]; then\n local base=\"$PI_CODING_AGENT_SESSION_DIR\"\n [ -d \"$base\" ] || return 0\n find \"$base\" -maxdepth 1 -name \"*.jsonl\" -mtime \"-${DAYS}\" 2>/dev/null\n if [ -z \"$REPO_CWD\" ]; then\n for dir in \"$base\"/*\"$REPO_NAME\"*/; do\n [ -d \"$dir\" ] || continue\n find \"$dir\" -maxdepth 1 -name \"*.jsonl\" -mtime \"-${DAYS}\" 2>/dev/null\n done\n fi\n return 0\n fi\n\n # omp has two bucket-naming schemes in the wild, and both keep the raw\n # repo basename (spaces and all) inside the bucket name:\n # - raw: \"-\u003chome-rel>\", \"-tmp-\u003ctmp-rel>\", \"--\u003cabs>--\" (legacy relative to\n # the hashed scheme; restored as current in omp 17.2.9, #7646)\n # - hashed: \"\u003cscope>-\u003csanitized-basename>-\u003csha256-of-canonical-cwd>\"\n # (intermediate releases; basename runs of [^a-zA-Z0-9._-] collapse to\n # \"-\", edge dashes stripped, capped at the last 80 chars, empty falls\n # back to \"project\" — session-paths.ts getDefaultSessionDirName)\n # Scan basename-matching buckets in the default-profile sessions root and\n # in every named-profile root; exact repo attribution comes from the\n # downstream header `cwd` filter (extract-metadata.py --cwd-filter reads\n # the type:\"session\" header). Glob the sanitized form so repos whose\n # basename contains characters the hashed scheme normalizes (e.g. spaces)\n # still match, and glob the raw form so raw-scheme buckets whose basename\n # sanitizes differently (e.g. \"my repo\" in \"--Users-test-Code-my repo--\")\n # are found too. When --cwd is supplied, also probe the exact raw bucket\n # name: it catches buckets the basename globs miss when the bucket's path\n # segments no longer resemble the repo name as typed.\n local sanitized\n sanitized=\"$(printf '%s' \"$REPO_NAME\" | sed -E 's/[^a-zA-Z0-9._-]+/-/g; s/^-+//; s/-+$//' | tail -c 80)\"\n [ -n \"$sanitized\" ] || sanitized=\"project\"\n local agent_dir=\"${PI_CODING_AGENT_DIR:-$HOME/$config_dir/agent}\"\n # omp getSessionsDir() flattens the agent/ prefix under XDG: ~/.omp/agent/sessions\n # becomes $XDG_DATA_HOME/omp/sessions (and $XDG_DATA_HOME/omp/profiles/\u003cname>/sessions).\n # Only linux/darwin, only when XDG_DATA_HOME is set and the omp app root exists,\n # and only when PI_CODING_AGENT_DIR did not already take over the agent root.\n local xdg_omp=\"\"\n if [ -z \"${PI_CODING_AGENT_DIR:-}\" ] && [ -n \"${XDG_DATA_HOME:-}\" ]; then\n case \"$(uname -s)\" in\n Linux|Darwin)\n if [ -d \"${XDG_DATA_HOME%/}/omp\" ]; then\n xdg_omp=\"${XDG_DATA_HOME%/}/omp\"\n fi\n ;;\n esac\n fi\n {\n local root dir encoded\n if [ -n \"$REPO_CWD\" ]; then\n encoded=\"$(encode_omp_raw_cwd \"$REPO_CWD\")\"\n if [ -n \"$encoded\" ]; then\n for root in \"$agent_dir/sessions\" \\\n \"$HOME/$config_dir\"/profiles/*/agent/sessions \\\n ${xdg_omp:+\"$xdg_omp/sessions\"} \\\n ${xdg_omp:+\"$xdg_omp\"/profiles/*/sessions}; do\n [ -d \"$root/$encoded\" ] || continue\n find \"$root/$encoded\" -maxdepth 1 -name \"*.jsonl\" -mtime \"-${DAYS}\" 2>/dev/null\n done\n fi\n fi\n for root in \"$agent_dir/sessions\" \\\n \"$HOME/$config_dir\"/profiles/*/agent/sessions \\\n ${xdg_omp:+\"$xdg_omp/sessions\"} \\\n ${xdg_omp:+\"$xdg_omp\"/profiles/*/sessions}; do\n [ -d \"$root\" ] || continue\n for dir in \"$root\"/*\"$sanitized\"*/; do\n [ -d \"$dir\" ] || continue\n find \"$dir\" -maxdepth 1 -name \"*.jsonl\" -mtime \"-${DAYS}\" 2>/dev/null\n done\n if [ \"$REPO_NAME\" != \"$sanitized\" ]; then\n for dir in \"$root\"/*\"$REPO_NAME\"*/; do\n [ -d \"$dir\" ] || continue\n find \"$dir\" -maxdepth 1 -name \"*.jsonl\" -mtime \"-${DAYS}\" 2>/dev/null\n done\n fi\n done\n # The probe and the globs can hit the same bucket; emit each path once.\n } | awk '!seen[$0]++'\n}\n\n# --- Dispatch ---\ncase \"$PLATFORM\" in\n claude) discover_claude ;;\n codex) discover_codex ;;\n cursor) discover_cursor ;;\n pi) discover_pi ;;\n omp) discover_omp ;;\n all)\n # Pi and omp share the PI_CODING_AGENT_SESSION_DIR override: when it\n # is set, both discover functions emit the same flat-dir files, and\n # the downstream xargs call does not deduplicate. Emit each path once;\n # platform attribution is unaffected because extract-metadata.py\n # detects the file shape (title slot => omp, otherwise pi).\n { discover_claude; discover_codex; discover_cursor; discover_pi; discover_omp; } | awk '!seen[$0]++'\n ;;\n *)\n echo \"Unknown platform: $PLATFORM\" >&2\n exit 1\n ;;\nesac\n","content_type":"application/x-sh; charset=utf-8","language":"bash","size":11345,"content_sha256":"f12ac7d98600d1116cfeed84503553df2fb412ef665caa20eff449b286efba24"},{"filename":"scripts/session-history/extract-errors.py","content":"#!/usr/bin/env python3\n\"\"\"Extract error signals from a Claude Code, Codex, Cursor, Pi, or oh-my-pi (omp) JSONL session file.\n\nUsage:\n cat \u003csession.jsonl> | python3 extract-errors.py\n cat \u003csession.jsonl> | python3 extract-errors.py --output PATH\n\nAuto-detects platform from the JSONL structure.\nNote: Cursor agent transcripts do not log tool results, so no errors can be extracted.\nFinds failed tool calls / commands and outputs them with timestamps.\n\nWhen --output PATH is given, the extracted error log is written to PATH and\nstdout receives only a one-line JSON status (_meta with wrote/bytes/stats).\nThis lets callers route bulk content to a scratch file without round-tripping\nextraction bytes through orchestrator tool results.\n\nWithout --output, extracted content goes to stdout and ends with a _meta line.\n\"\"\"\nimport argparse\nimport io\nimport os\nimport sys\nimport json\n\nparser = argparse.ArgumentParser(add_help=True)\nparser.add_argument(\n \"--output\",\n metavar=\"PATH\",\n help=\"Write extracted errors to PATH instead of stdout. Stdout receives a one-line _meta status.\",\n)\nargs = parser.parse_args()\n\n# Session logs are UTF-8; force UTF-8 I/O so Windows' cp1252 default can't crash\n# on non-Latin-1 content like emoji, smart quotes, or box-drawing (#1258).\nsys.stdin.reconfigure(encoding=\"utf-8\", errors=\"replace\")\nsys.stdout.reconfigure(encoding=\"utf-8\", errors=\"replace\")\n\n_original_stdout = sys.stdout\nif args.output:\n sys.stdout = io.StringIO()\n\nstats = {\"lines\": 0, \"parse_errors\": 0, \"errors_found\": 0}\n\n\ndef summarize_error(raw):\n \"\"\"Extract a short error summary instead of dumping the full payload.\"\"\"\n text = str(raw).strip()\n # Take the first non-empty line as the error message\n for line in text.split(\"\\n\"):\n line = line.strip()\n if line:\n return line[:200]\n return text[:200]\n\n\ndef handle_claude(obj):\n if obj.get(\"type\") == \"user\":\n content = obj.get(\"message\", {}).get(\"content\", [])\n if isinstance(content, list):\n for block in content:\n if block.get(\"type\") == \"tool_result\" and block.get(\"is_error\"):\n ts = obj.get(\"timestamp\", \"\")[:19]\n summary = summarize_error(block.get(\"content\", \"\"))\n print(f\"[{ts}] [error] {summary}\")\n print(\"---\")\n stats[\"errors_found\"] += 1\n\n\ndef handle_codex(obj):\n if obj.get(\"type\") == \"event_msg\":\n p = obj.get(\"payload\", {})\n if p.get(\"type\") == \"exec_command_end\":\n output = p.get(\"aggregated_output\", \"\")\n stderr = p.get(\"stderr\", \"\")\n command = p.get(\"command\", [])\n cmd_str = command[-1] if command else \"\"\n\n exit_match = None\n if \"Process exited with code \" in output:\n try:\n code_str = output.split(\"Process exited with code \")[1].split(\"\\n\")[0]\n exit_code = int(code_str)\n if exit_code != 0:\n exit_match = exit_code\n except (IndexError, ValueError):\n pass\n\n if exit_match is not None or stderr:\n ts = obj.get(\"timestamp\", \"\")[:19]\n error_summary = summarize_error(stderr if stderr else output)\n print(f\"[{ts}] [error] exit={exit_match} cmd={cmd_str[:120]}: {error_summary}\")\n print(\"---\")\n stats[\"errors_found\"] += 1\n\n\ndef _pi_content_summary(content):\n if isinstance(content, str):\n return summarize_error(content)\n if isinstance(content, list):\n text = \"\\n\".join(\n block.get(\"text\", \"\")\n for block in content\n if isinstance(block, dict) and block.get(\"type\") in (\"text\", \"toolError\")\n )\n return summarize_error(text)\n return summarize_error(content)\n\n\ndef _pi_active_path_objects(objects):\n \"\"\"Return only entries on Pi's active leaf-to-root path.\"\"\"\n by_id = {\n obj.get(\"id\"): obj\n for obj in objects\n if isinstance(obj.get(\"id\"), str) and obj.get(\"type\") != \"session\"\n }\n leaf_id = None\n for obj in objects:\n if obj.get(\"type\") != \"session\" and isinstance(obj.get(\"id\"), str):\n leaf_id = obj[\"id\"]\n if not leaf_id:\n return objects\n\n active_ids = set()\n current = leaf_id\n while isinstance(current, str) and current and current not in active_ids:\n active_ids.add(current)\n parent = by_id.get(current, {}).get(\"parentId\")\n current = parent if isinstance(parent, str) else None\n return [\n obj\n for obj in objects\n if obj.get(\"type\") == \"session\" or obj.get(\"id\") in active_ids\n ]\n\n\ndef _pi_context_objects(objects):\n \"\"\"Return Pi entries that participate in active LLM context.\"\"\"\n active = _pi_active_path_objects(objects)\n compactions = [obj for obj in active if obj.get(\"type\") == \"compaction\"]\n if not compactions:\n return active\n\n first_kept = compactions[-1].get(\"firstKeptEntryId\")\n if not isinstance(first_kept, str):\n return active\n\n latest_compaction_id = compactions[-1].get(\"id\")\n started = False\n found_first_kept = False\n context = [obj for obj in active if obj.get(\"type\") == \"session\"]\n context.append(compactions[-1])\n for obj in active:\n if obj.get(\"type\") == \"session\":\n continue\n if obj.get(\"id\") == first_kept:\n started = True\n found_first_kept = True\n if obj.get(\"id\") == latest_compaction_id:\n continue\n if started:\n context.append(obj)\n return context if found_first_kept and len(context) > 1 else active\n\n\ndef handle_pi(obj):\n # omp's physical type:'title' slot line returns here like any non-message\n # entry. type:'title_change' entries are a different pi entry type — they\n # return the same way; do not conflate the two when filtering.\n if obj.get(\"type\") != \"message\":\n return\n msg = obj.get(\"message\", {})\n if msg.get(\"role\") == \"bashExecution\":\n exit_code = msg.get(\"exitCode\")\n if exit_code in (None, 0) and not msg.get(\"cancelled\"):\n return\n ts = obj.get(\"timestamp\", \"\")[:19]\n command = msg.get(\"command\", \"\")\n output = msg.get(\"output\", \"\")\n summary = summarize_error(output)\n status = \"cancelled\" if msg.get(\"cancelled\") else f\"exit={exit_code}\"\n print(f\"[{ts}] [error] {status} cmd={command[:120]}: {summary}\")\n print(\"---\")\n stats[\"errors_found\"] += 1\n return\n\n if msg.get(\"role\") != \"toolResult\":\n return\n content = msg.get(\"content\", [])\n is_error = bool(msg.get(\"isError\"))\n if isinstance(content, list):\n is_error = is_error or any(\n isinstance(block, dict) and block.get(\"type\") == \"toolError\"\n for block in content\n )\n if not is_error:\n return\n\n ts = obj.get(\"timestamp\", \"\")[:19]\n tool = msg.get(\"toolName\", \"unknown\")\n summary = _pi_content_summary(content)\n print(f\"[{ts}] [error] tool={tool}: {summary}\")\n print(\"---\")\n stats[\"errors_found\"] += 1\n\n\n# Auto-detect platform from first few lines, then process all\ndetected = None\nbuffer = []\n# omp files physically begin with a fixed-width type:'title' slot line before\n# the pi-shaped type:'session' header; bare pi files start with the header.\nseen_title_slot = False\n\nfor line in sys.stdin:\n line = line.strip()\n if not line:\n continue\n buffer.append(line)\n stats[\"lines\"] += 1\n\n if not detected and len(buffer) \u003c= 10:\n try:\n obj = json.loads(line)\n if obj.get(\"type\") == \"session\" and \"cwd\" in obj:\n detected = \"omp\" if seen_title_slot else \"pi\"\n elif obj.get(\"type\") == \"title\" and len(buffer) == 1:\n seen_title_slot = True\n elif obj.get(\"type\") in (\"user\", \"assistant\"):\n detected = \"claude\"\n elif obj.get(\"type\") in (\"session_meta\", \"turn_context\", \"response_item\", \"event_msg\"):\n detected = \"codex\"\n elif obj.get(\"role\") in (\"user\", \"assistant\") and \"type\" not in obj:\n detected = \"cursor\"\n except (json.JSONDecodeError, KeyError):\n pass\n\n# Cursor transcripts don't log tool results — no errors to extract\ndef handle_noop(obj):\n pass\n\nhandlers = {\"claude\": handle_claude, \"codex\": handle_codex, \"cursor\": handle_noop, \"pi\": handle_pi, \"omp\": handle_pi}\nhandler = handlers.get(detected, handle_noop)\n\nobjects = []\nfor line in buffer:\n try:\n objects.append(json.loads(line))\n except (json.JSONDecodeError, KeyError):\n stats[\"parse_errors\"] += 1\n\nif detected in (\"pi\", \"omp\"):\n objects = _pi_context_objects(objects)\n\nfor obj in objects:\n try:\n handler(obj)\n except KeyError:\n stats[\"parse_errors\"] += 1\n\nprint(json.dumps({\"_meta\": True, **stats}))\n\nif args.output:\n body = sys.stdout.getvalue()\n sys.stdout = _original_stdout\n with open(args.output, \"w\", encoding=\"utf-8\") as f:\n f.write(body)\n bytes_written = os.path.getsize(args.output)\n print(json.dumps({\"_meta\": True, \"wrote\": args.output, \"bytes\": bytes_written, **stats}))\n","content_type":"text/x-python; charset=utf-8","language":"python","size":9270,"content_sha256":"1232b5f148b30cdca0d9753ea5b054e55f766befa399c64a78a698b804ecc086"},{"filename":"scripts/session-history/extract-metadata.py","content":"#!/usr/bin/env python3\n\"\"\"Extract session metadata from Claude Code, Codex, Cursor, Pi, and oh-my-pi (omp) JSONL files.\n\nBatch mode (preferred — one invocation for all files):\n python3 extract-metadata.py /path/to/dir/*.jsonl\n python3 extract-metadata.py file1.jsonl file2.jsonl file3.jsonl\n\nSingle-file mode (stdin):\n head -20 \u003csession.jsonl> | python3 extract-metadata.py\n\nAuto-detects platform from the JSONL structure.\nOutputs one JSON object per file, one per line.\nIncludes a final _meta line with processing stats.\n\"\"\"\nimport sys\nimport json\nimport os\n\nMAX_LINES = 200 # Resumed Claude sessions front-load dozens of non-message records before the first user record\n\n\ndef try_claude(lines):\n result = None\n cwd = \"\"\n for line in lines:\n try:\n obj = json.loads(line.strip())\n except (json.JSONDecodeError, KeyError):\n continue\n if not cwd and obj.get(\"cwd\"):\n cwd = obj[\"cwd\"]\n if result is None and obj.get(\"type\") == \"user\" and \"gitBranch\" in obj:\n result = {\n \"platform\": \"claude\",\n \"branch\": obj[\"gitBranch\"],\n \"ts\": obj.get(\"timestamp\", \"\"),\n \"session\": obj.get(\"sessionId\", \"\"),\n }\n if obj.get(\"cwd\"):\n cwd = obj[\"cwd\"]\n if result is not None and cwd:\n break\n if result is not None and cwd:\n result[\"cwd\"] = cwd\n return result\n\n\ndef try_codex(lines):\n meta = {}\n for line in lines:\n try:\n obj = json.loads(line.strip())\n if obj.get(\"type\") == \"session_meta\":\n p = obj.get(\"payload\", {})\n meta[\"platform\"] = \"codex\"\n meta[\"cwd\"] = p.get(\"cwd\", \"\")\n meta[\"session\"] = p.get(\"id\", \"\")\n meta[\"ts\"] = p.get(\"timestamp\", obj.get(\"timestamp\", \"\"))\n meta[\"source\"] = p.get(\"source\", \"\")\n meta[\"cli_version\"] = p.get(\"cli_version\", \"\")\n elif obj.get(\"type\") == \"turn_context\":\n p = obj.get(\"payload\", {})\n meta[\"model\"] = p.get(\"model\", \"\")\n meta[\"cwd\"] = meta.get(\"cwd\") or p.get(\"cwd\", \"\")\n except (json.JSONDecodeError, KeyError):\n pass\n return meta if meta else None\n\n\ndef try_omp(lines):\n \"\"\"oh-my-pi (omp) sessions: a fixed-width type='title' slot line physically\n first, then a pi-shaped type='session' header with cwd. Checked before Pi:\n a bare pi file has no title slot and must still detect as pi.\"\"\"\n seen_first = False\n for line in lines:\n stripped = line.strip()\n if not stripped:\n continue\n try:\n obj = json.loads(stripped)\n except (json.JSONDecodeError, KeyError):\n if not seen_first:\n return None\n continue\n if not seen_first:\n seen_first = True\n if obj.get(\"type\") != \"title\":\n return None\n continue\n if obj.get(\"type\") == \"session\" and \"cwd\" in obj:\n return {\n \"platform\": \"omp\",\n \"cwd\": obj.get(\"cwd\", \"\"),\n \"session\": obj.get(\"id\", \"\"),\n \"ts\": obj.get(\"timestamp\", \"\"),\n }\n return None\n\n\ndef try_pi(lines):\n \"\"\"Pi sessions: type='session' header with cwd, followed by message entries.\"\"\"\n for line in lines:\n try:\n obj = json.loads(line.strip())\n if obj.get(\"type\") == \"session\" and \"cwd\" in obj:\n return {\n \"platform\": \"pi\",\n \"cwd\": obj.get(\"cwd\", \"\"),\n \"session\": obj.get(\"id\", \"\"),\n \"ts\": obj.get(\"timestamp\", \"\"),\n }\n except (json.JSONDecodeError, KeyError):\n pass\n return None\n\n\ndef try_cursor(lines):\n \"\"\"Cursor agent transcripts: role-based entries, no timestamps or metadata fields.\"\"\"\n for line in lines:\n try:\n obj = json.loads(line.strip())\n # Cursor entries have 'role' at top level but no 'type'\n if obj.get(\"role\") in (\"user\", \"assistant\") and \"type\" not in obj:\n return {\"platform\": \"cursor\"}\n except (json.JSONDecodeError, KeyError):\n pass\n return None\n\n\ndef extract_from_lines(lines):\n return try_claude(lines) or try_codex(lines) or try_omp(lines) or try_pi(lines) or try_cursor(lines)\n\n\nTAIL_BYTES = 16384 # Read last 16KB to find final timestamp past trailing metadata\n\n\ndef get_last_timestamp(filepath, size):\n \"\"\"Read the tail of a file to find the last message with a timestamp.\"\"\"\n try:\n with open(filepath, \"rb\") as f:\n f.seek(max(0, size - TAIL_BYTES))\n tail = f.read().decode(\"utf-8\", errors=\"ignore\")\n lines = tail.strip().split(\"\\n\")\n for line in reversed(lines):\n try:\n obj = json.loads(line.strip())\n if \"timestamp\" in obj:\n return obj[\"timestamp\"]\n except (json.JSONDecodeError, KeyError):\n pass\n except (OSError, IOError):\n pass\n return None\n\n\ndef _pi_active_path_objects(objects):\n \"\"\"Return only entries on Pi's active leaf-to-root path.\n\n Pi session files are append-only trees. The final non-session entry is the\n active leaf; abandoned branches remain in the file but are not in context.\n \"\"\"\n by_id = {\n obj.get(\"id\"): obj\n for obj in objects\n if isinstance(obj.get(\"id\"), str) and obj.get(\"type\") != \"session\"\n }\n leaf_id = None\n for obj in objects:\n if obj.get(\"type\") != \"session\" and isinstance(obj.get(\"id\"), str):\n leaf_id = obj[\"id\"]\n if not leaf_id:\n return objects\n\n active_ids = set()\n current = leaf_id\n while isinstance(current, str) and current and current not in active_ids:\n active_ids.add(current)\n parent = by_id.get(current, {}).get(\"parentId\")\n current = parent if isinstance(parent, str) else None\n return [\n obj\n for obj in objects\n if obj.get(\"type\") == \"session\" or obj.get(\"id\") in active_ids\n ]\n\n\ndef _pi_context_objects(objects):\n \"\"\"Return Pi entries that participate in active LLM context.\"\"\"\n active = _pi_active_path_objects(objects)\n compactions = [obj for obj in active if obj.get(\"type\") == \"compaction\"]\n if not compactions:\n return active\n\n # Pi emits compaction summary first, then entries from firstKeptEntryId\n # onward. Exclude older ancestors so keyword search mirrors context.\n first_kept = compactions[-1].get(\"firstKeptEntryId\")\n if not isinstance(first_kept, str):\n return active\n\n latest_compaction_id = compactions[-1].get(\"id\")\n started = False\n found_first_kept = False\n context = [obj for obj in active if obj.get(\"type\") == \"session\"]\n context.append(compactions[-1])\n for obj in active:\n if obj.get(\"type\") == \"session\":\n continue\n if obj.get(\"id\") == first_kept:\n started = True\n found_first_kept = True\n if obj.get(\"id\") == latest_compaction_id:\n continue\n if started:\n context.append(obj)\n return context if found_first_kept and len(context) > 1 else active\n\n\ndef _append_pi_content_text(chunks, content):\n if isinstance(content, str):\n chunks.append(content)\n elif isinstance(content, list):\n for block in content:\n if isinstance(block, dict) and block.get(\"type\") == \"text\":\n chunks.append(block.get(\"text\", \"\"))\n\n\ndef _append_pi_tool_call_targets(chunks, content):\n \"\"\"Append searchable Pi toolCall targets without indexing tool output.\"\"\"\n if not isinstance(content, list):\n return\n for block in content:\n if not isinstance(block, dict) or block.get(\"type\") != \"toolCall\":\n continue\n args = block.get(\"arguments\", {})\n if not isinstance(args, dict):\n continue\n for key in (\"path\", \"file_path\", \"command\", \"pattern\", \"query\", \"prompt\"):\n value = args.get(key)\n if isinstance(value, str):\n chunks.append(value)\n\n\ndef _extract_user_assistant_text(filepath):\n \"\"\"Return concatenated user + assistant text content from a session JSONL.\n\n Skips JSONL metadata field names and values (sessionId, gitBranch, uuid,\n timestamps, type tags), tool_use blocks (tool names + tool inputs),\n tool_result blocks (tool outputs), and thinking/reasoning blocks. Only\n content the user or assistant actually said is included.\n\n Without this filtering, common topic words like \"session\" would match every\n JSONL file via the sessionId field, drowning out real content matches.\n \"\"\"\n chunks = []\n try:\n objects = []\n with open(filepath, \"r\", encoding=\"utf-8\", errors=\"replace\") as f:\n for line in f:\n try:\n objects.append(json.loads(line.strip()))\n except (json.JSONDecodeError, ValueError):\n continue\n\n # omp files share the pi-shaped session header, so this covers both.\n is_pi = any(\n obj.get(\"type\") == \"session\" and \"cwd\" in obj for obj in objects\n )\n if is_pi:\n objects = _pi_context_objects(objects)\n\n for obj in objects:\n # Claude Code: type-tagged top-level\n t = obj.get(\"type\")\n if t == \"user\":\n msg = obj.get(\"message\", {})\n content = msg.get(\"content\")\n if isinstance(content, str):\n chunks.append(content)\n elif isinstance(content, list):\n for block in content:\n if isinstance(block, dict) and block.get(\"type\") == \"text\":\n chunks.append(block.get(\"text\", \"\"))\n # Skip tool_result blocks — tool outputs are not user content.\n continue\n if t == \"assistant\":\n msg = obj.get(\"message\", {})\n content = msg.get(\"content\", [])\n if isinstance(content, list):\n for block in content:\n if isinstance(block, dict) and block.get(\"type\") == \"text\":\n chunks.append(block.get(\"text\", \"\"))\n # Skip tool_use and thinking blocks.\n continue\n\n # Codex: payload-typed events\n if t == \"event_msg\":\n p = obj.get(\"payload\", {})\n if p.get(\"type\") == \"user_message\":\n # Strip Codex/Conductor `\u003csystem_instruction>...\u003c/system_instruction>`\n # wrapper before counting. Without this, generic wrapper terms\n # (e.g., \"Conductor\", environment labels) false-match against\n # boilerplate the user did not author. Mirrors the same split\n # used in ce-session-extract/scripts/extract-skeleton.py.\n msg = p.get(\"message\", \"\")\n if isinstance(msg, str):\n parts = msg.split(\"\u003c/system_instruction>\")\n chunks.append(parts[-1] if parts else msg)\n continue\n if t == \"response_item\":\n p = obj.get(\"payload\", {})\n if p.get(\"type\") == \"message\" and p.get(\"role\") == \"assistant\":\n for block in p.get(\"content\", []):\n if isinstance(block, dict) and block.get(\"type\") == \"output_text\":\n chunks.append(block.get(\"text\", \"\"))\n continue\n\n # Pi: type='message' envelope with AgentMessage under message.\n if t == \"message\" and \"message\" in obj:\n msg = obj.get(\"message\", {})\n role = msg.get(\"role\", \"\")\n if role == \"bashExecution\":\n command = msg.get(\"command\", \"\")\n if isinstance(command, str):\n chunks.append(command)\n # Search command text only. Output is tool output and can\n # be large/noisy in the same way as toolResult content.\n continue\n content = msg.get(\"content\", [])\n if role == \"custom\":\n _append_pi_content_text(chunks, content)\n continue\n if role not in (\"user\", \"assistant\"):\n continue\n _append_pi_content_text(chunks, content)\n if role == \"assistant\":\n _append_pi_tool_call_targets(chunks, content)\n continue\n\n if t in (\"compaction\", \"branch_summary\"):\n summary = obj.get(\"summary\", \"\")\n if isinstance(summary, str):\n chunks.append(summary)\n continue\n\n if t == \"custom_message\":\n _append_pi_content_text(chunks, obj.get(\"content\", []))\n continue\n\n # Cursor: role-tagged with no top-level type\n if obj.get(\"role\") in (\"user\", \"assistant\") and \"type\" not in obj:\n msg = obj.get(\"message\", {})\n for block in msg.get(\"content\", []) if isinstance(msg.get(\"content\"), list) else []:\n if isinstance(block, dict) and block.get(\"type\") == \"text\":\n chunks.append(block.get(\"text\", \"\"))\n continue\n except (OSError, IOError):\n pass\n return \"\\n\".join(chunks)\n\n\ndef count_keyword_matches(filepath, keywords):\n \"\"\"Case-insensitive substring count for each keyword in user/assistant text.\n\n Returns a dict {original_keyword: count}. Scans only content the user or\n assistant said — not JSONL metadata, tool calls, tool outputs, or thinking\n blocks — so common topic words like \"session\" do not false-match against\n the sessionId field.\n \"\"\"\n text_lower = _extract_user_assistant_text(filepath).lower()\n return {kw: text_lower.count(kw.lower()) for kw in keywords}\n\n\ndef process_file(filepath):\n \"\"\"Extract metadata only. Keyword scanning is done separately so callers\n can apply cheap filters (e.g. --cwd-filter) before paying the full-file\n content scan cost.\n\n Returns (result, error): (dict, None) on success, (None, None) when the\n file is readable but matches no platform, (None, filepath) on read error.\"\"\"\n try:\n size = os.path.getsize(filepath)\n with open(filepath, \"r\", encoding=\"utf-8\", errors=\"replace\") as f:\n lines = []\n for i, line in enumerate(f):\n if i >= MAX_LINES:\n break\n lines.append(line)\n result = extract_from_lines(lines)\n if result:\n result[\"file\"] = filepath\n result[\"size\"] = size\n return result, None\n else:\n return None, None\n except (OSError, IOError) as e:\n return None, filepath\n\n\ndef _normalize_cwd(path):\n path = path.replace(\"\\\\\", \"/\").strip()\n if len(path) > 1 and path.endswith(\"/\") and not (\n len(path) == 3 and path[1] == \":\"\n ):\n path = path.rstrip(\"/\")\n # Windows drive paths are case-insensitive. Do not rewrite POSIX /d/...\n # to D:/ — those are different filesystems.\n if len(path) >= 2 and path[1] == \":\":\n path = path[0].upper() + \":\" + path[2:].casefold()\n return path\n\n\ndef _is_abs_cwd(path):\n if os.path.isabs(path):\n return True\n return len(path) >= 2 and path[1] == \":\"\n\n\ndef _cwd_paths_related(session_cwd, cwd_filter):\n if session_cwd == cwd_filter:\n return True\n prefix = cwd_filter if cwd_filter.endswith(\"/\") else cwd_filter + \"/\"\n if session_cwd.startswith(prefix):\n return True\n prefix = session_cwd if session_cwd.endswith(\"/\") else session_cwd + \"/\"\n return cwd_filter.startswith(prefix)\n\n\ndef _attach_timestamps(result, filepath):\n if result[\"platform\"] == \"cursor\":\n # Cursor transcripts have no timestamps in JSONL.\n # Use file modification time as the best available signal.\n # Derive session ID from the parent directory name (UUID).\n mtime = os.path.getmtime(filepath)\n from datetime import datetime, timezone\n\n result[\"ts\"] = datetime.fromtimestamp(mtime, tz=timezone.utc).isoformat()\n result[\"session\"] = os.path.basename(os.path.dirname(filepath))\n return\n last_ts = get_last_timestamp(filepath, result[\"size\"])\n if last_ts:\n result[\"last_ts\"] = last_ts\n\n\ndef cwd_matches_filter(session_cwd, cwd_filter):\n if not session_cwd or not cwd_filter:\n return True\n session = _normalize_cwd(session_cwd)\n filt = _normalize_cwd(cwd_filter)\n if not session or not filt:\n return True\n if _is_abs_cwd(filt):\n return _cwd_paths_related(session, filt)\n return filt in session.split(\"/\")\n\n\n# Parse arguments: files and optional --cwd-filter / --keyword\nfiles = []\ncwd_filter = None\nkeywords = None\nargs = sys.argv[1:]\ni = 0\nwhile i \u003c len(args):\n if args[i] == \"--cwd-filter\" and i + 1 \u003c len(args):\n cwd_filter = args[i + 1]\n i += 2\n elif args[i] == \"--keyword\" and i + 1 \u003c len(args):\n keywords = [k for k in args[i + 1].split(\",\") if k]\n i += 2\n elif not args[i].startswith(\"-\"):\n files.append(args[i])\n i += 1\n else:\n i += 1\n\nif files:\n # Batch mode: process all files\n processed = 0\n parse_errors = 0\n no_metadata = 0\n filtered = 0\n matched = 0\n for filepath in files:\n if not filepath.endswith(\".jsonl\"):\n continue\n result, error = process_file(filepath)\n processed += 1\n if result:\n # Apply CWD filter first: cheap metadata-only check. Skip\n # sessions from other repos before paying the full-file keyword\n # scan cost — Claude and Codex discovery list across projects,\n # so without this ordering --keyword would scan files that are\n # immediately discarded.\n if cwd_filter:\n session_cwd = result.get(\"cwd\")\n if session_cwd:\n if not cwd_matches_filter(session_cwd, cwd_filter):\n filtered += 1\n continue\n elif result.get(\"platform\") in (\"claude\", \"codex\", \"pi\", \"omp\"):\n filtered += 1\n continue\n _attach_timestamps(result, filepath)\n # Apply keyword scan only after cheap filters pass.\n if keywords:\n matches = count_keyword_matches(filepath, keywords)\n result[\"keyword_matches\"] = matches\n result[\"match_count\"] = sum(matches.values())\n if result[\"match_count\"] == 0:\n continue\n matched += 1\n print(json.dumps(result))\n elif error:\n parse_errors += 1\n else:\n no_metadata += 1\n\n meta = {\"_meta\": True, \"files_processed\": processed, \"parse_errors\": parse_errors}\n if no_metadata:\n meta[\"files_without_metadata\"] = no_metadata\n if filtered:\n meta[\"filtered_by_cwd\"] = filtered\n if keywords:\n meta[\"files_matched\"] = matched\n print(json.dumps(meta))\nelse:\n # No file arguments: either single-file stdin mode or empty xargs invocation.\n # When xargs runs us with no input (e.g., discover found no files), stdin is\n # empty or a TTY — emit a clean zero-file result instead of a false parse error.\n if sys.stdin.isatty():\n lines = []\n else:\n lines = list(sys.stdin)\n\n if not lines:\n # No input at all — zero-file result (clean exit for empty pipelines).\n # When --keyword was supplied, emit files_matched: 0 so callers relying\n # on its presence to terminate quickly in zero-match scans see a\n # consistent shape with the batch-mode no-match case.\n meta = {\"_meta\": True, \"files_processed\": 0, \"parse_errors\": 0}\n if keywords:\n meta[\"files_matched\"] = 0\n print(json.dumps(meta))\n else:\n # Genuine single-file stdin mode (backward compatible)\n result = extract_from_lines(lines)\n meta = {\"_meta\": True, \"files_processed\": 1, \"parse_errors\": 0}\n if result:\n print(json.dumps(result))\n else:\n meta[\"files_without_metadata\"] = 1\n print(json.dumps(meta))\n","content_type":"text/x-python; charset=utf-8","language":"python","size":20599,"content_sha256":"33fd414ddff18562fd661c1d34b739c6640dfcd7ec5050e424bc11a269742201"},{"filename":"scripts/session-history/extract-skeleton.py","content":"#!/usr/bin/env python3\n\"\"\"Extract the conversation skeleton from a Claude Code, Codex, Cursor, Pi, or oh-my-pi (omp) JSONL session file.\n\nUsage:\n cat \u003csession.jsonl> | python3 extract-skeleton.py\n cat \u003csession.jsonl> | python3 extract-skeleton.py --output PATH\n\nAuto-detects platform (Claude Code, Codex, Cursor, Pi, or oh-my-pi (omp)) from the JSONL structure.\nExtracts:\n - User messages (text only, no tool results)\n - Assistant text (no thinking/reasoning blocks)\n - Collapsed tool call summaries (consecutive same-tool calls grouped)\n\nConsecutive tool calls of the same type are collapsed:\n 3+ Read calls -> \"[tools] 3x Read (file1, file2, +1 more) -> all ok\"\nCodex call/result pairs are deduplicated (only the result with status is kept).\n\nWhen --output PATH is given, the extracted skeleton is written to PATH and\nstdout receives only a one-line JSON status (_meta with wrote/bytes/stats).\nThis lets callers route bulk content to a scratch file without round-tripping\nextraction bytes through orchestrator tool results.\n\nWithout --output, extracted content goes to stdout and ends with a _meta line.\n\"\"\"\nimport argparse\nimport io\nimport os\nimport sys\nimport json\nimport re\n\nparser = argparse.ArgumentParser(add_help=True)\nparser.add_argument(\n \"--output\",\n metavar=\"PATH\",\n help=\"Write extracted skeleton to PATH instead of stdout. Stdout receives a one-line _meta status.\",\n)\nargs = parser.parse_args()\n\n# Session logs are UTF-8; force UTF-8 I/O so Windows' cp1252 default can't crash\n# on non-Latin-1 content like emoji, smart quotes, or box-drawing (#1258).\nsys.stdin.reconfigure(encoding=\"utf-8\", errors=\"replace\")\nsys.stdout.reconfigure(encoding=\"utf-8\", errors=\"replace\")\n\n# Capture-and-redirect when --output is set: prints in the rest of the script\n# go to the buffer; at the end the buffer is written to PATH and a status\n# line is emitted to the real stdout.\n_original_stdout = sys.stdout\nif args.output:\n sys.stdout = io.StringIO()\n\nstats = {\"lines\": 0, \"parse_errors\": 0, \"user\": 0, \"assistant\": 0, \"tool\": 0, \"summary\": 0}\n\n# Claude Code wrapper tags to strip from user message content.\n# Strip entirely (tag + content): framework noise and raw command output.\n# Strip tags only (keep content): command-message, command-name, command-args, user_query.\n_STRIP_BLOCK = re.compile(\n r\"\u003c(?:task-notification|local-command-caveat|local-command-stdout|local-command-stderr|system-reminder)[^>]*>.*?\u003c/(?:task-notification|local-command-caveat|local-command-stdout|local-command-stderr|system-reminder)>\",\n re.DOTALL,\n)\n_STRIP_TAG = re.compile(\n r\"\u003c/?(?:command-message|command-name|command-args|user_query)[^>]*>\"\n)\n\n\ndef clean_text(text):\n \"\"\"Strip framework wrapper tags from message text (Claude and Cursor).\"\"\"\n text = _STRIP_BLOCK.sub(\"\", text)\n text = _STRIP_TAG.sub(\"\", text)\n text = re.sub(r\"\\n{3,}\", \"\\n\\n\", text).strip()\n return text\n\n# Buffer for pending tool entries: [{\"ts\", \"name\", \"target\", \"status\"}]\npending_tools = []\n\n\ndef flush_tools():\n \"\"\"Print buffered tool entries, collapsing consecutive same-name groups.\"\"\"\n if not pending_tools:\n return\n\n # Group consecutive entries by tool name\n groups = []\n for entry in pending_tools:\n if groups and groups[-1][0][\"name\"] == entry[\"name\"]:\n groups[-1].append(entry)\n else:\n groups.append([entry])\n\n for group in groups:\n name = group[0][\"name\"]\n if len(group) \u003c= 2:\n # Print individually\n for e in group:\n status = f\" -> {e['status']}\" if e.get(\"status\") else \"\"\n ts_prefix = f\"[{e['ts']}] \" if e.get(\"ts\") else \"\"\n print(f\"{ts_prefix}[tool] {name} {e['target']}{status}\")\n stats[\"tool\"] += 1\n else:\n # Collapse\n ts = group[0].get(\"ts\", \"\")\n targets = [e[\"target\"] for e in group if e.get(\"target\")]\n ok = sum(1 for e in group if e.get(\"status\") == \"ok\")\n err = sum(1 for e in group if e.get(\"status\") and e[\"status\"] != \"ok\")\n no_status = len(group) - ok - err\n\n # Show first 2 targets, then \"+N more\"\n if len(targets) > 2:\n target_str = \", \".join(targets[:2]) + f\", +{len(targets) - 2} more\"\n elif targets:\n target_str = \", \".join(targets)\n else:\n target_str = \"\"\n\n if no_status == len(group):\n status_str = \"\"\n elif err == 0:\n status_str = \" -> all ok\"\n else:\n status_str = f\" -> {ok} ok, {err} error\"\n\n ts_prefix = f\"[{ts}] \" if ts else \"\"\n print(f\"{ts_prefix}[tools] {len(group)}x {name} ({target_str}){status_str}\")\n stats[\"tool\"] += len(group)\n\n pending_tools.clear()\n\n\ndef _safe_slice(value, n):\n \"\"\"Slice value if it is a string; otherwise return ''.\n\n Some Claude Code / MCP tool inputs put structured data (dicts, lists) in\n fields like `query` or `prompt`. `dict[:N]` raises TypeError, so guard\n every slice with an isinstance check.\n \"\"\"\n return value[:n] if isinstance(value, str) else \"\"\n\n\ndef summarize_claude_tool(block):\n \"\"\"Extract name and target from a Claude Code tool_use block.\"\"\"\n name = block.get(\"name\", \"unknown\")\n inp = block.get(\"input\", {})\n fp = inp.get(\"file_path\")\n p = inp.get(\"path\")\n target = (\n (fp if isinstance(fp, str) else None)\n or (p if isinstance(p, str) else None)\n or _safe_slice(inp.get(\"command\"), 120)\n or _safe_slice(inp.get(\"pattern\"), 200)\n or _safe_slice(inp.get(\"query\"), 80)\n or _safe_slice(inp.get(\"prompt\"), 80)\n or \"\"\n )\n if isinstance(target, str) and len(target) > 120:\n target = target[:120]\n return name, target\n\n\ndef handle_claude(obj):\n msg_type = obj.get(\"type\")\n ts = obj.get(\"timestamp\", \"\")[:19]\n\n if msg_type == \"user\":\n msg = obj.get(\"message\", {})\n content = msg.get(\"content\", \"\")\n\n if isinstance(content, list):\n for block in content:\n if block.get(\"type\") == \"tool_result\":\n is_error = block.get(\"is_error\", False)\n status = \"error\" if is_error else \"ok\"\n tool_use_id = block.get(\"tool_use_id\")\n matched = False\n if tool_use_id:\n for entry in pending_tools:\n if entry.get(\"id\") == tool_use_id:\n entry[\"status\"] = status\n matched = True\n break\n if not matched:\n # Fallback: assign to earliest pending entry without a status\n for entry in pending_tools:\n if not entry.get(\"status\"):\n entry[\"status\"] = status\n break\n\n texts = [\n c.get(\"text\", \"\")\n for c in content\n if c.get(\"type\") == \"text\" and len(c.get(\"text\", \"\")) > 10\n ]\n content = \" \".join(texts)\n\n if isinstance(content, str):\n content = clean_text(content)\n if len(content) > 15:\n flush_tools()\n print(f\"[{ts}] [user] {content[:800]}\")\n print(\"---\")\n stats[\"user\"] += 1\n\n elif msg_type == \"assistant\":\n msg = obj.get(\"message\", {})\n content = msg.get(\"content\", [])\n if isinstance(content, list):\n has_text = False\n for block in content:\n if block.get(\"type\") == \"text\":\n text = clean_text(block.get(\"text\", \"\"))\n if len(text) > 20:\n if not has_text:\n flush_tools()\n has_text = True\n print(f\"[{ts}] [assistant] {text[:800]}\")\n print(\"---\")\n stats[\"assistant\"] += 1\n elif block.get(\"type\") == \"tool_use\":\n name, target = summarize_claude_tool(block)\n entry = {\"ts\": ts, \"name\": name, \"target\": target}\n tool_id = block.get(\"id\")\n if tool_id:\n entry[\"id\"] = tool_id\n pending_tools.append(entry)\n\n\ndef handle_codex(obj):\n msg_type = obj.get(\"type\")\n ts = obj.get(\"timestamp\", \"\")[:19]\n\n if msg_type == \"event_msg\":\n p = obj.get(\"payload\", {})\n if p.get(\"type\") == \"user_message\":\n text = p.get(\"message\", \"\")\n if isinstance(text, str) and len(text) > 15:\n parts = text.split(\"\u003c/system_instruction>\")\n user_text = parts[-1].strip() if parts else text\n if len(user_text) > 15:\n flush_tools()\n print(f\"[{ts}] [user] {user_text[:800]}\")\n print(\"---\")\n stats[\"user\"] += 1\n\n elif p.get(\"type\") == \"exec_command_end\":\n # This is the deduplicated result — has status info\n command = p.get(\"command\", [])\n cmd_str = command[-1] if command else \"\"\n output = p.get(\"aggregated_output\", \"\")\n\n status = \"ok\"\n if \"Process exited with code \" in output:\n try:\n code = int(output.split(\"Process exited with code \")[1].split(\"\\n\")[0])\n if code != 0:\n status = f\"error(exit {code})\"\n except (IndexError, ValueError):\n pass\n\n if cmd_str:\n # Shorten common patterns for readability\n short_cmd = cmd_str[:120]\n pending_tools.append({\"ts\": ts, \"name\": \"exec\", \"target\": short_cmd, \"status\": status})\n\n elif msg_type == \"response_item\":\n p = obj.get(\"payload\", {})\n if p.get(\"type\") == \"message\" and p.get(\"role\") == \"assistant\":\n for block in p.get(\"content\", []):\n if block.get(\"type\") == \"output_text\" and len(block.get(\"text\", \"\")) > 20:\n flush_tools()\n print(f\"[{ts}] [assistant] {block['text'][:800]}\")\n print(\"---\")\n stats[\"assistant\"] += 1\n\n # Skip function_call — exec_command_end is the deduplicated version with status\n\n\ndef _pi_text_content(content):\n if isinstance(content, str):\n return [content]\n if not isinstance(content, list):\n return []\n return [\n block.get(\"text\", \"\")\n for block in content\n if isinstance(block, dict) and block.get(\"type\") == \"text\"\n ]\n\n\ndef _pi_active_path_objects(objects):\n \"\"\"Return only entries on Pi's active leaf-to-root path.\"\"\"\n by_id = {\n obj.get(\"id\"): obj\n for obj in objects\n if isinstance(obj.get(\"id\"), str) and obj.get(\"type\") != \"session\"\n }\n leaf_id = None\n for obj in objects:\n if obj.get(\"type\") != \"session\" and isinstance(obj.get(\"id\"), str):\n leaf_id = obj[\"id\"]\n if not leaf_id:\n return objects\n\n active_ids = set()\n current = leaf_id\n while isinstance(current, str) and current and current not in active_ids:\n active_ids.add(current)\n parent = by_id.get(current, {}).get(\"parentId\")\n current = parent if isinstance(parent, str) else None\n return [\n obj\n for obj in objects\n if obj.get(\"type\") == \"session\" or obj.get(\"id\") in active_ids\n ]\n\n\ndef _pi_context_objects(objects):\n \"\"\"Return Pi entries that participate in active LLM context.\"\"\"\n active = _pi_active_path_objects(objects)\n compactions = [obj for obj in active if obj.get(\"type\") == \"compaction\"]\n if not compactions:\n return active\n\n first_kept = compactions[-1].get(\"firstKeptEntryId\")\n if not isinstance(first_kept, str):\n return active\n\n latest_compaction_id = compactions[-1].get(\"id\")\n started = False\n found_first_kept = False\n context = [obj for obj in active if obj.get(\"type\") == \"session\"]\n context.append(compactions[-1])\n for obj in active:\n if obj.get(\"type\") == \"session\":\n continue\n if obj.get(\"id\") == first_kept:\n started = True\n found_first_kept = True\n if obj.get(\"id\") == latest_compaction_id:\n continue\n if started:\n context.append(obj)\n return context if found_first_kept and len(context) > 1 else active\n\n\ndef handle_pi(obj):\n \"\"\"Pi/omp sessions: type='message' with message.role and content blocks.\n\n omp's physical type:'title' slot line lands here too and returns below\n like any non-message entry. type:'title_change' entries are a different\n pi entry type — they also carry no message payload and return the same\n way; do not conflate the two when filtering.\"\"\"\n entry_type = obj.get(\"type\")\n ts = obj.get(\"timestamp\", \"\")[:19]\n\n if entry_type in (\"compaction\", \"branch_summary\"):\n text = clean_text(obj.get(\"summary\", \"\"))\n if len(text) > 15:\n flush_tools()\n print(f\"[{ts}] [summary] {text[:800]}\")\n print(\"---\")\n stats[\"summary\"] += 1\n return\n\n if entry_type == \"custom_message\":\n text = clean_text(\" \".join(_pi_text_content(obj.get(\"content\", []))))\n if len(text) > 15:\n flush_tools()\n print(f\"[{ts}] [summary] {text[:800]}\")\n print(\"---\")\n stats[\"summary\"] += 1\n return\n\n if entry_type != \"message\":\n return\n\n msg = obj.get(\"message\", {})\n role = msg.get(\"role\", \"\")\n content = msg.get(\"content\", [])\n\n if role == \"bashExecution\":\n exit_code = msg.get(\"exitCode\")\n if msg.get(\"cancelled\"):\n status = \"cancelled\"\n elif exit_code in (None, 0):\n status = \"ok\"\n else:\n status = f\"error(exit {exit_code})\"\n command = _safe_slice(msg.get(\"command\"), 120)\n pending_tools.append({\"ts\": ts, \"name\": \"bash\", \"target\": command, \"status\": status})\n return\n\n if role == \"custom\":\n text = clean_text(\" \".join(_pi_text_content(content)))\n if len(text) > 15:\n flush_tools()\n print(f\"[{ts}] [summary] {text[:800]}\")\n print(\"---\")\n stats[\"summary\"] += 1\n return\n\n if role == \"user\":\n text = clean_text(\" \".join(_pi_text_content(content)))\n if len(text) > 15:\n flush_tools()\n print(f\"[{ts}] [user] {text[:800]}\")\n print(\"---\")\n stats[\"user\"] += 1\n\n elif role == \"assistant\":\n if isinstance(content, str):\n text = clean_text(content)\n if len(text) > 20:\n flush_tools()\n print(f\"[{ts}] [assistant] {text[:800]}\")\n print(\"---\")\n stats[\"assistant\"] += 1\n return\n\n has_text = False\n for block in (content if isinstance(content, list) else []):\n if not isinstance(block, dict):\n continue\n if block.get(\"type\") == \"text\":\n text = clean_text(block.get(\"text\", \"\"))\n if len(text) > 20:\n if not has_text:\n flush_tools()\n has_text = True\n print(f\"[{ts}] [assistant] {text[:800]}\")\n print(\"---\")\n stats[\"assistant\"] += 1\n elif block.get(\"type\") == \"toolCall\":\n name = block.get(\"name\", \"unknown\")\n args = block.get(\"arguments\", {})\n if not isinstance(args, dict):\n args = {}\n target = (\n _safe_slice(args.get(\"path\"), 200)\n or _safe_slice(args.get(\"file_path\"), 200)\n or _safe_slice(args.get(\"command\"), 120)\n or _safe_slice(args.get(\"pattern\"), 200)\n or _safe_slice(args.get(\"query\"), 80)\n or _safe_slice(args.get(\"prompt\"), 80)\n or \"\"\n )\n if isinstance(target, str) and len(target) > 120:\n target = target[:120]\n entry = {\"ts\": ts, \"name\": name, \"target\": target}\n tool_id = block.get(\"id\")\n if tool_id:\n entry[\"id\"] = tool_id\n pending_tools.append(entry)\n\n elif role == \"toolResult\":\n tool_call_id = msg.get(\"toolCallId\")\n is_error = bool(msg.get(\"isError\"))\n if isinstance(content, list):\n is_error = is_error or any(\n isinstance(block, dict) and block.get(\"type\") == \"toolError\"\n for block in content\n )\n status = \"error\" if is_error else \"ok\"\n if tool_call_id:\n for entry in pending_tools:\n if entry.get(\"id\") == tool_call_id:\n entry[\"status\"] = status\n break\n else:\n for entry in pending_tools:\n if not entry.get(\"status\"):\n entry[\"status\"] = status\n break\n\n\ndef handle_cursor(obj):\n \"\"\"Cursor agent transcripts: role-based, no timestamps, same content structure as Claude.\"\"\"\n role = obj.get(\"role\")\n content = obj.get(\"message\", {}).get(\"content\", [])\n\n if role == \"user\":\n texts = []\n for block in (content if isinstance(content, list) else []):\n if block.get(\"type\") == \"text\":\n texts.append(block.get(\"text\", \"\"))\n text = clean_text(\" \".join(texts))\n if len(text) > 15:\n flush_tools()\n # No timestamps available in Cursor transcripts\n print(f\"[user] {text[:800]}\")\n print(\"---\")\n stats[\"user\"] += 1\n\n elif role == \"assistant\":\n has_text = False\n for block in (content if isinstance(content, list) else []):\n if block.get(\"type\") == \"text\":\n text = block.get(\"text\", \"\")\n # Skip [REDACTED] placeholder blocks\n if len(text) > 20 and text.strip() != \"[REDACTED]\":\n if not has_text:\n flush_tools()\n has_text = True\n print(f\"[assistant] {text[:800]}\")\n print(\"---\")\n stats[\"assistant\"] += 1\n elif block.get(\"type\") == \"tool_use\":\n name = block.get(\"name\", \"unknown\")\n inp = block.get(\"input\", {})\n p = inp.get(\"path\")\n fp = inp.get(\"file_path\")\n target = (\n (p if isinstance(p, str) else None)\n or (fp if isinstance(fp, str) else None)\n or _safe_slice(inp.get(\"command\"), 120)\n or _safe_slice(inp.get(\"pattern\"), 200)\n or _safe_slice(inp.get(\"glob_pattern\"), 200)\n or _safe_slice(inp.get(\"target_directory\"), 200)\n or \"\"\n )\n if isinstance(target, str) and len(target) > 120:\n target = target[:120]\n # No status info available — Cursor doesn't log tool results\n pending_tools.append({\"ts\": \"\", \"name\": name, \"target\": target})\n\n\n# Auto-detect platform from first few lines, then process all\ndetected = None\nbuffer = []\n# omp files physically begin with a fixed-width type:'title' slot line before\n# the pi-shaped type:'session' header; bare pi files start with the header.\nseen_title_slot = False\n\nfor line in sys.stdin:\n line = line.strip()\n if not line:\n continue\n buffer.append(line)\n stats[\"lines\"] += 1\n\n # Scan until a decisive record appears (short-circuits once detected).\n # Claude Code sessions front-load 10+ non-message metadata records\n # (last-prompt, custom-title, attachments, ...), so capping detection to\n # the first N lines silently misroutes them to the codex handler and\n # produces an empty skeleton with parse_errors: 0 (issue #923).\n if not detected:\n try:\n obj = json.loads(line)\n if obj.get(\"type\") == \"session\" and \"cwd\" in obj:\n detected = \"omp\" if seen_title_slot else \"pi\"\n elif obj.get(\"type\") == \"title\" and len(buffer) == 1:\n seen_title_slot = True\n elif obj.get(\"type\") in (\"user\", \"assistant\"):\n detected = \"claude\"\n elif obj.get(\"type\") in (\"session_meta\", \"turn_context\", \"response_item\", \"event_msg\"):\n detected = \"codex\"\n elif obj.get(\"role\") in (\"user\", \"assistant\") and \"type\" not in obj:\n detected = \"cursor\"\n except (json.JSONDecodeError, KeyError):\n pass\n\nhandlers = {\"claude\": handle_claude, \"codex\": handle_codex, \"cursor\": handle_cursor, \"pi\": handle_pi, \"omp\": handle_pi}\nhandler = handlers.get(detected, handle_codex)\n\nobjects = []\nfor line in buffer:\n try:\n objects.append(json.loads(line))\n except (json.JSONDecodeError, KeyError):\n stats[\"parse_errors\"] += 1\n\nif detected in (\"pi\", \"omp\"):\n objects = _pi_context_objects(objects)\n\nfor obj in objects:\n try:\n handler(obj)\n except KeyError:\n stats[\"parse_errors\"] += 1\n\n# Flush any remaining buffered tools\nflush_tools()\n\nprint(json.dumps({\"_meta\": True, **stats}))\n\nif args.output:\n body = sys.stdout.getvalue()\n sys.stdout = _original_stdout\n with open(args.output, \"w\", encoding=\"utf-8\") as f:\n f.write(body)\n bytes_written = os.path.getsize(args.output)\n print(json.dumps({\"_meta\": True, \"wrote\": args.output, \"bytes\": bytes_written, **stats}))\n","content_type":"text/x-python; charset=utf-8","language":"python","size":21932,"content_sha256":"0ce683e6e97506816deb25fcbf428a8559de48cf81e25e6abbc0ad76933b0ed1"},{"filename":"scripts/validate-doc-claims.py","content":"#!/usr/bin/env python3\n\"\"\"Validate cited claims in a solution doc against the git tree.\n\nUsage:\n python3 validate-doc-claims.py \u003cdoc-path>\n\nExit codes:\n 0 — nothing flagged\n 1 — one or more flags need adjudication (report on stdout)\n 2 — usage error (bad arguments, missing file)\n\nScope: mechanical grounding checks on a written doc's *body*. Complements\nvalidate-frontmatter.py (parser-safety) — this script checks the body's\ncitations against the repository:\n\n 1. Cited repo-relative paths (backticked, containing at least one '/')\n exist in the working tree, including already-absolute citations that\n fall inside the repo (rewritten to repo-relative before candidacy).\n Tokens containing '../' resolve from the doc's directory (those\n escaping the repo are skipped). Misses tracked at HEAD or the\n upstream default branch still count as real paths and are classified\n (deleted/uncommitted vs stale checkout). Tokens missing everywhere\n are flagged only when path-shaped; slash-delimited identifiers\n (branch names, git refs, provider/model IDs) and slash-prefixed\n URL routes are skipped.\n 2. Cited commit SHAs (7-40 hex chars with at least one digit and one\n a-f letter) resolve to commits, classified by reachability from\n HEAD and the upstream default branch. Session ids, content hashes\n and blob hashes are hex too, so an unresolvable hex word is\n reported in one of two tiers rather than asserted to be fabricated:\n FLAG when the text right before it presents it as a commit (a\n likely fabricated citation), NOTE otherwise (an identifier this\n script cannot classify). Only FLAG affects the exit code. The\n cue vocabulary therefore ranks confidence; it does not decide\n whether an item is surfaced, so a phrasing it misses is reported\n one tier down instead of disappearing.\n 3. Relative markdown link targets resolve from the doc's location.\n 4. Dangling drafting scaffold: \"Learning(s) N\" numbering and\n unresolved {{...}} placeholder tokens. Inline code spans and fenced\n code blocks are masked first, so a {{...}} shown as documented syntax\n (Handlebars, a CI variable, a GitHub ruleset placeholder) is not\n mistaken for a leaked scaffold; only a bare token in prose is flagged.\n\nFlags are adjudication input, NOT hard failures — a doc may legitimately\ncite a path deleted by the very fix it documents. The calling agent\ndecides per flag: fix, annotate as historical, or confirm intentional.\nOnly the summary exit code distinguishes \"clean\" from \"needs a look\".\n\nThe script never touches the network (no fetch); classification uses\nwhatever refs exist locally. Run a best-effort `git fetch --quiet` first\nwhen freshness matters. Pure stdlib (no third-party deps).\n\"\"\"\nimport os\nimport re\nimport subprocess\nimport sys\n\n# Tokens containing these are placeholders/examples, not real citations.\nPLACEHOLDER_CHARS = set(\"\u003c>{}*$\")\nPLACEHOLDER_SUBSTRINGS = (\"path/to\", \"...\", \"…\")\n\nSHA_RE = re.compile(r\"\\b[0-9a-f]{7,40}\\b\")\n# Words that name a commit or a commit operation, so a hex token right after\n# one is a commit citation. Words naming some other git object (\"blob\", \"tree\"),\n# any hash (\"sha\"), and the bare tool name (\"git\", which precedes every object\n# kind equally) are deliberately absent — they are what the flag used to mistake\n# for commits.\nCOMMIT_WORDS = frozenset(\n \"commit commits committed committing revision revisions rev revs \"\n \"revert reverts reverted cherry-pick cherry-picked rebase rebased \"\n \"bisect bisected\".split()\n)\n# A hex token is also a citation when the sentence attributes a change landing\n# in this repository to it: \"landed in \u003csha>\", \"resolved by \u003csha>\". Both halves\n# are needed. The preposition alone attributes without saying what to, so it\n# would read \"recorded at \u003cdigest>\" as a commit; the verb alone does not point\n# at the token. Membership below is that condition, not a tally of phrasings\n# seen so far: a verb belongs when it says a change landed, and does not when\n# it says an identifier was assigned or a value stored.\nCITATION_VERBS = frozenset(\n \"fixed fix fixes landed lands land introduced introduces introduce \"\n \"shipped ships ship merged merges merge resolved resolves resolve \"\n \"reverted reverts broke breaks broken caused causes regressed \"\n \"added adds removed removes released releases\".split()\n)\nCITATION_PREPS = frozenset((\"in\", \"by\", \"at\", \"with\"))\n# The pin form that names a commit is owner/repo@\u003csha>. A bare \"@\" is not it:\n# it also prefixes account names and image tags, whose identifiers are hex too.\nREPO_PIN_RE = re.compile(r\"(?\u003c![\\w./@-])[\\w.-]+/[\\w.-]+@$\")\nBACKTICK_RE = re.compile(r\"`([^`\\n]+)`\")\nMD_LINK_RE = re.compile(r\"\\[[^\\]]*\\]\$([^)\\s]+)\$\")\nFENCE_RE = re.compile(r\"^\\s*(`{3,}|~{3,})(.*)$\")\nSCAFFOLD_RES = (\n re.compile(r\"\\bLearnings?\\s+#?\\d\"),\n re.compile(r\"\\{\\{[^}\\n]*\\}\\}\"),\n)\n\n\ndef usage_fail(msg: str) -> \"NoReturn\":\n sys.stderr.write(f\"validate-doc-claims: {msg}\\n\")\n sys.exit(2)\n\n\ndef git(args: list[str], cwd: str) -> tuple[int, str]:\n try:\n result = subprocess.run(\n [\"git\", *args],\n cwd=cwd,\n capture_output=True,\n text=True,\n timeout=30,\n )\n return result.returncode, result.stdout.strip()\n except (OSError, subprocess.TimeoutExpired):\n return 1, \"\"\n\n\ndef split_body(text: str) -> tuple[str, int]:\n \"\"\"Return (body, 1-indexed line number the body starts on).\n\n Skips YAML frontmatter when present so frontmatter fields are not\n scanned as body citations.\n \"\"\"\n lines = text.split(\"\\n\")\n if lines and lines[0].rstrip() == \"---\":\n for i in range(1, len(lines)):\n if lines[i].rstrip() == \"---\":\n return \"\\n\".join(lines[i + 1 :]), i + 2\n return text, 1\n\n\ndef is_path_candidate(token: str, *, known_path: bool = False) -> bool:\n if any(ch.isspace() for ch in token):\n return False\n if not known_path and \"/\" not in token:\n return False\n if \"://\" in token or token.startswith((\"http\", \"#\", \"/\", \"~\")):\n return False\n if token.startswith((\"origin/\", \"upstream/\", \"refs/\")):\n return False # git refs, not repo paths\n if PLACEHOLDER_CHARS & set(token):\n return False\n if any(sub in token for sub in PLACEHOLDER_SUBSTRINGS):\n return False\n return True\n\n\ndef is_path_shaped(token: str, base: str) -> bool:\n \"\"\"Distinguish a path citation from a slash-delimited identifier\n (branch name, provider/model ID) among tokens found nowhere in git.\"\"\"\n segments = token.split(\"/\")\n if re.search(r\"\\.[A-Za-z0-9]{1,8}$\", segments[-1]):\n return True\n if token.endswith(\"/\"):\n return True\n return os.path.isdir(os.path.join(base, segments[0]))\n\n\ndef mask_code(lines: list[str]) -> list[str]:\n \"\"\"Blank out fenced code blocks and inline code spans, preserving line\n count and length. Illustrative {{...}} in quoted code must not read as a\n leaked drafting scaffold; only bare tokens in prose should.\"\"\"\n masked: list[str] = []\n fence: str | None = None # active fence run (e.g. \"```\"), or None\n for line in lines:\n m = FENCE_RE.match(line)\n if fence is None and m:\n fence = m.group(1)\n masked.append(\" \" * len(line))\n continue\n if fence is not None:\n # CommonMark: a closing fence is the same char, at least as long,\n # and followed only by whitespace — an info string (```json) opens\n # but never closes, so it stays block content.\n if (\n m\n and m.group(1)[0] == fence[0]\n and len(m.group(1)) >= len(fence)\n and not m.group(2).strip()\n ):\n fence = None\n masked.append(\" \" * len(line))\n continue\n masked.append(BACKTICK_RE.sub(lambda x: \" \" * len(x.group(0)), line))\n return masked\n\n\ndef cites_a_commit(prefix: str) -> bool:\n \"\"\"True when the text just before a hex word presents it as a commit.\n\n Only the last few words on the same line count: a cue further away\n (\"the git history shows session 7e6861b4\") describes the surroundings,\n not the token. Digits are word characters so a hash named after its\n algorithm (\"SHA256\") stays one non-cue word, and command flags drop out\n first so `git show --format=%H \u003csha>` reads like `git show \u003csha>`.\n \"\"\"\n if REPO_PIN_RE.search(prefix):\n return True # owner/repo@\u003csha> pins a commit\n unflagged = \" \".join(w for w in prefix.split() if not w.startswith(\"-\"))\n words = [\n w.lower() for w in re.findall(r\"[A-Za-z][A-Za-z0-9-]*\", unflagged)[-3:]\n ]\n if any(word in COMMIT_WORDS for word in words):\n return True\n return (\n len(words) >= 2\n and words[-1] in CITATION_PREPS\n and words[-2] in CITATION_VERBS\n )\n\n\ndef normalize_path(token: str) -> str:\n token = token.strip().rstrip(\".,;\")\n token = re.sub(r\":\\d+(-\\d+)?$\", \"\", token) # strip `:line` / `:a-b` refs\n if token.startswith(\"./\"):\n token = token[2:]\n return token\n\n\ndef strip_repo_prefix(token: str, base: str) -> str:\n \"\"\"Rewrite an already-absolute path inside the repo to repo-relative.\n\n Relative tokens, URL routes, and out-of-repo absolute paths are\n unchanged so the existing candidacy guard still drops them. Realpath\n both sides so a host where /tmp is a symlink still matches. A\n successful rewrite is slash-normalized so Windows relpath output\n stays a candidate.\n \"\"\"\n if not os.path.isabs(token):\n return token\n try:\n rel = os.path.relpath(os.path.realpath(token), os.path.realpath(base))\n except ValueError:\n return token\n if rel == \"..\" or rel.startswith(\"..\" + os.sep):\n return token\n return rel.replace(\"\\\\\", \"/\")\n\n\ndef main(argv: list[str]) -> int:\n if len(argv) != 2:\n usage_fail(f\"usage: {os.path.basename(argv[0])} \u003cdoc-path>\")\n\n doc_path = argv[1]\n if not os.path.isfile(doc_path):\n usage_fail(f\"file not found: {doc_path}\")\n\n with open(doc_path) as f:\n text = f.read()\n\n doc_dir = os.path.dirname(os.path.abspath(doc_path))\n body, body_start = split_body(text)\n body_lines = body.split(\"\\n\")\n\n def loc_suffix(needle: str) -> str:\n for i, line in enumerate(body_lines):\n if needle in line:\n return f\" (line {body_start + i})\"\n return \"\"\n\n infos: list[str] = []\n notes: list[str] = []\n flags: list[str] = []\n\n # --- Repo context -----------------------------------------------------\n code, repo_root = git([\"rev-parse\", \"--show-toplevel\"], doc_dir)\n in_git = code == 0 and bool(repo_root)\n upstream: str | None = None\n if in_git:\n code, ref = git([\"rev-parse\", \"--abbrev-ref\", \"origin/HEAD\"], repo_root)\n if code == 0 and ref:\n upstream = ref\n else:\n for candidate in (\"origin/main\", \"origin/master\"):\n code, _ = git(\n [\"rev-parse\", \"--verify\", \"--quiet\", candidate], repo_root\n )\n if code == 0:\n upstream = candidate\n break\n if upstream:\n code, behind = git(\n [\"rev-list\", \"--count\", f\"HEAD..{upstream}\"], repo_root\n )\n if code == 0 and behind.isdigit() and int(behind) > 0:\n infos.append(\n f\"INFO: worktree is {behind} commits behind {upstream} — \"\n \"verify merge-state claims against remote truth (gh pr view), \"\n \"not this checkout\"\n )\n else:\n infos.append(\n \"INFO: no upstream default branch found — \"\n \"path/SHA classification limited to HEAD\"\n )\n else:\n infos.append(\n \"INFO: not a git repository — path and SHA classification skipped \"\n \"(scaffold and link checks still apply)\"\n )\n\n def upstream_has_path(path: str) -> bool:\n if not (in_git and upstream):\n return False\n code, _ = git([\"cat-file\", \"-e\", f\"{upstream}:{path}\"], repo_root)\n return code == 0\n\n def head_has_path(path: str) -> bool:\n if not in_git:\n return False\n code, _ = git([\"cat-file\", \"-e\", f\"HEAD:{path}\"], repo_root)\n return code == 0\n\n # --- 1. Cited repo paths ----------------------------------------------\n checked_paths = 0\n seen_paths: set[str] = set()\n base = repo_root if in_git else os.getcwd()\n for raw in BACKTICK_RE.findall(body):\n token = normalize_path(raw)\n rewritten_abs = False\n if in_git:\n before = token\n token = strip_repo_prefix(token, base)\n rewritten_abs = token != before\n if not is_path_candidate(token, known_path=rewritten_abs):\n continue\n check = token\n if token.startswith(\"../\") or \"/../\" in token:\n # A `../` citation is doc-relative (matching how markdown links\n # resolve), so map it to a repo-root path before checking.\n if not in_git:\n continue\n resolved = os.path.realpath(os.path.join(doc_dir, token))\n check = os.path.relpath(resolved, os.path.realpath(base))\n if check.startswith(\"..\"):\n continue # escapes the repo — not checkable as a repo path\n if check in seen_paths:\n continue\n seen_paths.add(check)\n if os.path.exists(os.path.join(base, check)):\n checked_paths += 1\n continue\n tracked_head = head_has_path(check)\n tracked_upstream = upstream_has_path(check)\n if not (tracked_head or tracked_upstream) and not is_path_shaped(\n check, base\n ):\n continue # branch name / provider ID, not a path citation\n checked_paths += 1\n loc = loc_suffix(raw)\n if tracked_head:\n flags.append(\n f\"FLAG path `{token}`{loc} — tracked at HEAD but missing from \"\n \"the working tree: deleted or uncommitted removal? Annotate as \"\n \"historical (e.g. removed by this fix) or restore it.\"\n )\n elif tracked_upstream:\n flags.append(\n f\"FLAG path `{token}`{loc} — not in working tree but exists at \"\n f\"{upstream}: stale checkout? Annotate or verify against upstream.\"\n )\n else:\n flags.append(\n f\"FLAG path `{token}`{loc} — not found under {base}\"\n + (f\" or {upstream}\" if upstream else \"\")\n + \". Only that location was searched; check other repos or \"\n \"stores before treating the citation as wrong. Fix the \"\n \"citation, or annotate it as historical (e.g. removed by \"\n \"this fix).\"\n )\n\n # --- 2. Cited commit SHAs ----------------------------------------------\n checked_shas = 0\n if in_git:\n # One entry per distinct hex word: the line to report it at, and\n # whether any occurrence of it is presented as a commit. A doc often\n # quotes a token in a transcript before citing it, so a later citing\n # occurrence upgrades the tier and supplies the line.\n seen_shas: dict[str, tuple[int, bool]] = {}\n order: list[str] = []\n for m in SHA_RE.finditer(body):\n sha = m.group(0)\n if not (any(c.isdigit() for c in sha) and any(c in \"abcdef\" for c in sha)):\n continue # dates and decimal ids are not SHAs\n line_start = body.rfind(\"\\n\", 0, m.start()) + 1\n cited = cites_a_commit(body[line_start : m.start()])\n line_no = body_start + body.count(\"\\n\", 0, m.start())\n if sha not in seen_shas:\n seen_shas[sha] = (line_no, cited)\n order.append(sha)\n elif cited and not seen_shas[sha][1]:\n seen_shas[sha] = (line_no, True)\n for sha in order:\n line_no, cited = seen_shas[sha]\n code, _ = git([\"cat-file\", \"-e\", f\"{sha}^{{commit}}\"], repo_root)\n resolved = code == 0\n loc = f\" (line {line_no})\"\n if not resolved:\n if not cited:\n # Nothing here says \"commit\", and hex is also how session\n # ids and content hashes are written. Surface it without\n # claiming to know which it is; the reader adjudicates.\n notes.append(\n f\"NOTE sha {sha}{loc} — an unresolved hex identifier \"\n \"with no commit reference around it. This script cannot \"\n \"tell a session id or content hash from a commit; verify \"\n \"it if it was meant as one.\"\n )\n continue\n checked_shas += 1\n flags.append(\n f\"FLAG sha {sha}{loc} — does not resolve to a commit in this \"\n \"repository. Replace with the PR number, or drop it.\"\n )\n continue\n checked_shas += 1\n in_head = (\n git([\"merge-base\", \"--is-ancestor\", sha, \"HEAD\"], repo_root)[0] == 0\n )\n in_up = (\n upstream is not None\n and git([\"merge-base\", \"--is-ancestor\", sha, upstream], repo_root)[0]\n == 0\n )\n if in_head and (in_up or upstream is None):\n continue\n if in_head and not in_up:\n flags.append(\n f\"FLAG sha {sha}{loc} — reachable from HEAD but not {upstream}: \"\n \"local-only commit whose SHA may be rewritten on merge \"\n \"(rebase/squash). Prefer citing the PR number.\"\n )\n elif in_up:\n flags.append(\n f\"FLAG sha {sha}{loc} — not reachable from HEAD but reachable \"\n f\"from {upstream}: this checkout predates the merge. Add a \"\n \"temporal qualifier or verify the claim via gh.\"\n )\n else:\n flags.append(\n f\"FLAG sha {sha}{loc} — exists but unreachable from HEAD\"\n + (f\" or {upstream}\" if upstream else \"\")\n + \": likely a rebased-away commit. Prefer citing the PR number.\"\n )\n\n # --- 3. Relative markdown links -----------------------------------------\n checked_links = 0\n seen_links: set[str] = set()\n for target in MD_LINK_RE.findall(body):\n if re.match(r\"^[a-z][a-z0-9+.-]*:\", target, re.IGNORECASE):\n continue # URL scheme\n if target.startswith(\"#\"):\n continue # intra-doc anchor\n bare = target.split(\"#\", 1)[0]\n if not bare or bare in seen_links:\n continue\n seen_links.add(bare)\n checked_links += 1\n if not os.path.exists(os.path.normpath(os.path.join(doc_dir, bare))):\n loc = loc_suffix(target)\n flags.append(\n f\"FLAG link ({target}){loc} — relative target does not resolve \"\n \"from the doc's location. Fix the path.\"\n )\n\n # --- 4. Dangling drafting scaffold ---------------------------------------\n for i, line_text in enumerate(mask_code(body_lines)):\n for pattern in SCAFFOLD_RES:\n m = pattern.search(line_text)\n if m:\n flags.append(\n f'FLAG scaffold \"{m.group(0)}\" (line {body_start + i}) — '\n \"drafting-context reference leaked into the doc. Rewrite it \"\n \"as a real path or link.\"\n )\n\n # --- Report ---------------------------------------------------------------\n for info in infos:\n print(info)\n for note in notes:\n print(note)\n for flag in flags:\n print(flag)\n summary = (\n f\"checked {checked_paths} paths, {checked_shas} SHAs, \"\n f\"{checked_links} links; {len(flags)} flags\"\n )\n if notes:\n summary += f\", {len(notes)} notes\"\n print(summary)\n if flags:\n return 1\n print(f\"OK: {doc_path}\")\n return 0\n\n\nif __name__ == \"__main__\":\n sys.exit(main(sys.argv))\n","content_type":"text/x-python; charset=utf-8","language":"python","size":20561,"content_sha256":"24fae245bade400f7cb7a085ee68b08289f4d8b24481155a226ca976a2acf501"},{"filename":"scripts/validate-frontmatter.py","content":"#!/usr/bin/env python3\n\"\"\"Validate ce-compound learning-doc frontmatter for parser-safety issues.\n\nUsage:\n python3 validate-frontmatter.py \u003cdoc-path>\n\nExit codes:\n 0 — frontmatter passes all checks\n 1 — validation failure (diagnostics on stderr)\n 2 — usage error (bad arguments, missing file)\n\nScope: this script catches *parser-safety* issues — frontmatter that strict\nYAML parsers will silently misread. It does NOT validate against the\nschema's required-field or enum-value rules; that's a separate concern. The\nintent is to prevent the silent-data-loss bug class where YAML's quoting\nrules truncate or reframe scalar values without raising.\n\nChecks (regex-based, no YAML parser dependency):\n 1. File starts and ends frontmatter with `---` lines (matched as full\n lines, not substrings — `----` and `---extra` are rejected)\n 2. No top-level scalar value contains ` #` unquoted (silent comment\n truncation — what Codex caught on PR #695)\n 3. No top-level scalar value contains `: ` unquoted (mapping confusion —\n what surfaced in a 2026-04-16 plan doc's `title:` field)\n\nThe script does NOT flag values starting with YAML reserved indicators\n(`` ` ``, `*`, `&`, `!`, etc.) because those produce loud parser errors\ndownstream rather than silent corruption — they're already caught by\nwhatever consumes the doc. This validator's purpose is silent-corruption\nprevention, not lint.\n\nPure-stdlib (no PyYAML or other third-party deps). Runs in \u003c50ms typical.\nDesigned to produce concrete, actionable error messages so the calling\nagent can fix and retry without ambiguity.\n\"\"\"\nimport os\nimport re\nimport sys\n\n\ndef usage_fail(msg: str) -> \"NoReturn\":\n sys.stderr.write(f\"validate-frontmatter: {msg}\\n\")\n sys.exit(2)\n\n\ndef main(argv: list[str]) -> int:\n if len(argv) != 2:\n usage_fail(f\"usage: {os.path.basename(argv[0])} \u003cdoc-path>\")\n\n doc_path = argv[1]\n if not os.path.isfile(doc_path):\n usage_fail(f\"file not found: {doc_path}\")\n\n with open(doc_path) as f:\n text = f.read()\n\n issues: list[str] = []\n\n # Check 1: frontmatter delimiters. Match the delimiter as a complete\n # line whose stripped content is exactly `---` — substring matching\n # (e.g. `text.find(\"\\n---\", 4)`) would falsely accept `----` or\n # `---extra` as a terminator and let malformed docs slip through to\n # downstream parsers that require a strict `---` line.\n lines = text.split(\"\\n\")\n if not lines or lines[0].rstrip() != \"---\":\n sys.stderr.write(\n f\"FAIL: {doc_path}\\n\"\n f\" file does not start with '---' frontmatter delimiter line\\n\"\n )\n return 1\n\n end_idx: int | None = None\n for i in range(1, len(lines)):\n if lines[i].rstrip() == \"---\":\n end_idx = i\n break\n\n if end_idx is None:\n sys.stderr.write(\n f\"FAIL: {doc_path}\\n\"\n f\" frontmatter not closed (no '---' line after the opening delimiter)\\n\"\n )\n return 1\n\n fm_text = \"\\n\".join(lines[1:end_idx])\n\n # Checks 2 & 3: silent-corruption quoting risks on top-level scalar\n # fields. We scan line-by-line and only flag top-level mapping entries\n # (no leading whitespace) whose value isn't already quoted/structured.\n for lineno, line in enumerate(fm_text.split(\"\\n\"), start=2):\n stripped = line.lstrip()\n if not stripped or stripped.startswith(\"#\"):\n continue\n if \":\" not in line:\n continue\n # Top-level mapping keys only — skip nested values, array items\n if line.startswith((\" \", \"\\t\")):\n continue\n # Skip pure list-marker lines like \"- item\" (these can't be top-level\n # in our frontmatter convention, but be defensive)\n if stripped.startswith(\"- \"):\n continue\n\n key, _, val = line.partition(\":\")\n val_stripped = val.strip()\n if not val_stripped:\n # Key with no value on this line — likely a parent of a nested\n # block (`tags:` followed by `- foo`). Nothing to validate here.\n continue\n # Already quoted or structured (block scalar, flow collection)\n if val_stripped[0] in '\"\\'[{|>':\n continue\n\n if re.search(r\"\\s#\", val_stripped):\n issues.append(\n f\"line {lineno}: '{key.strip()}' value contains ' #' — quote it. \"\n \"YAML treats space-then-# as a comment delimiter and silently \"\n \"drops the rest of the value.\"\n )\n if re.search(r\":\\s\", val_stripped):\n issues.append(\n f\"line {lineno}: '{key.strip()}' value contains ': ' — quote it. \"\n \"Strict YAML parsers may treat this as a nested mapping.\"\n )\n\n if issues:\n sys.stderr.write(f\"FAIL: {doc_path}\\n\")\n for issue in issues:\n sys.stderr.write(f\" {issue}\\n\")\n return 1\n\n print(f\"OK: {doc_path}\")\n return 0\n\n\nif __name__ == \"__main__\":\n sys.exit(main(sys.argv))\n","content_type":"text/x-python; charset=utf-8","language":"python","size":5060,"content_sha256":"580815f6b6030f5f3383972283927ff7354649b7b599b008941a9440d5a5e674"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"/ce-compound","type":"text"}]},{"type":"paragraph","content":[{"text":"Outcome:","type":"text","marks":[{"type":"strong"}]},{"text":" the lesson of one qualifying solved problem is held in the cheapest form that stops the repeat, grounded against the current tree, discoverable by the next agent.","type":"text"}]},{"type":"paragraph","content":[{"text":"Done:","type":"text","marks":[{"type":"strong"}]},{"text":" the chosen form is written or updated and validated, vocabulary capture recorded, and the mode's completion report emitted; when no lesson qualifies, nothing is written and the report says why.","type":"text"}]},{"type":"paragraph","content":[{"text":"One learning per run.","type":"text","marks":[{"type":"strong"}]},{"text":" A session that produced several gets several sequential runs, never one batched run. Read ","type":"text"},{"text":"references/research.md","type":"text","marks":[{"type":"code_inline"}]},{"text":"; it explains what batching breaks.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Preconditions","type":"text"}]},{"type":"paragraph","content":[{"text":"Document only a problem that is solved and verified.","type":"text"}]},{"type":"paragraph","content":[{"text":"\u003c!-- ce-durable-bar:start --> A learning earns its place only when it holds durable project reasoning that is not readily recoverable from the final code, tests, types, comments, or existing documentation, and losing it would plausibly cause recurrence, material risk, or substantial rediscovery. Apply this counterfactual: if the learning document disappeared, would a future engineer reading the final implementation still be likely to repeat the mistake or redo substantial investigation? Completion, effort, and diff size do not establish eligibility. \u003c!-- ce-durable-bar:end -->","type":"text"}]},{"type":"paragraph","content":[{"text":"If the counterfactual fails, write nothing and report why. Judge this from the session rather than asking. An explicit invocation requests the judgment now but does not lower the bar.","type":"text"}]},{"type":"paragraph","content":[{"text":"An existing learning that became materially inaccurate or incomplete qualifies because leaving it would mislead. Update that learning instead of creating a duplicate.","type":"text"}]},{"type":"paragraph","content":[{"text":"ce-compound","type":"text","marks":[{"type":"code_inline"}]},{"text":" does not bootstrap ","type":"text"},{"text":"CONCEPTS.md","type":"text","marks":[{"type":"code_inline"}]},{"text":". It seeds the learning's own area as a side effect, never the whole repo. Send a standalone request to create or bootstrap that file to ","type":"text"},{"text":"ce-compound-refresh","type":"text","marks":[{"type":"code_inline"}]},{"text":", then exit.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Mode Detection","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"/ce-compound [brief context]\n/ce-compound mode:non-interactive depth:lightweight [context]\n/ce-compound mode:non-interactive depth:full [context]","type":"text"}]},{"type":"paragraph","content":[{"text":"Enter non-interactive mode when ","type":"text"},{"text":"either","type":"text","marks":[{"type":"strong"}]},{"text":" holds: the arguments you were invoked with contain the ","type":"text"},{"text":"mode:non-interactive","type":"text","marks":[{"type":"code_inline"}]},{"text":" token or its deprecated alias ","type":"text"},{"text":"mode:headless","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"or","type":"text","marks":[{"type":"strong"}]},{"text":" the invocation makes non-interactive intent unmistakable, such as a caller or standing instruction asking to run ","type":"text"},{"text":"ce-compound","type":"text","marks":[{"type":"code_inline"}]},{"text":" \"headless\", \"non-interactively\", \"unattended\", or \"without prompts/questions\". Both tokens together is not a conflict. Bare \"automatically\" or \"auto-run\" is ","type":"text"},{"text":"not","type":"text","marks":[{"type":"strong"}]},{"text":" on its own a non-interactive signal: it speaks to ","type":"text"},{"text":"invoking","type":"text","marks":[{"type":"em"}]},{"text":" the skill, not to suppressing its prompts. An ambiguous or absent signal defaults to interactive. Tokens starting with ","type":"text"},{"text":"mode:","type":"text","marks":[{"type":"code_inline"}]},{"text":" or ","type":"text"},{"text":"depth:","type":"text","marks":[{"type":"code_inline"}]},{"text":" are flags, not context: strip them before treating the remainder as the brief context hint. Once detected, non-interactive mode applies for the entire run.","type":"text"}]},{"type":"paragraph","content":[{"text":"Depth is chosen only by an explicit token, only in non-interactive mode, and at most one depth token is accepted. ","type":"text"},{"text":"depth:lightweight","type":"text","marks":[{"type":"code_inline"}]},{"text":" routes directly to Lightweight Mode. ","type":"text"},{"text":"depth:full","type":"text","marks":[{"type":"code_inline"}]},{"text":" or no depth token enters Full Mode, including its automatic session-history probe. A non-interactive call carrying no depth token therefore behaves as it always has. Non-interactive lightweight asks no blocking questions and launches no subagents. If the invocation carries an unknown ","type":"text"},{"text":"depth:","type":"text","marks":[{"type":"code_inline"}]},{"text":" token, multiple ","type":"text"},{"text":"depth:","type":"text","marks":[{"type":"code_inline"}]},{"text":" tokens, or a ","type":"text"},{"text":"depth:","type":"text","marks":[{"type":"code_inline"}]},{"text":" token without non-interactive intent, do not guess: emit the non-interactive failure report with the reason and end with ","type":"text"},{"text":"Documentation skipped","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]},{"type":"paragraph","content":[{"text":"Non-interactive mode asks nothing.","type":"text","marks":[{"type":"strong"}]},{"text":" It asks no blocking question of any kind, in any phase, because a caller reaching this path has no human to answer one. Every non-interactive exit, including one taken before any phase runs, ends on a terminal signal a caller parses: ","type":"text"},{"text":"Documentation complete","type":"text","marks":[{"type":"code_inline"}]},{"text":", or ","type":"text"},{"text":"Documentation skipped","type":"text","marks":[{"type":"code_inline"}]},{"text":" with the reason when the run wrote nothing. Interactive mode asks only where a step's own reference says to.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Artifact Root","type":"text"}]},{"type":"paragraph","content":[{"text":"Resolve ","type":"text"},{"text":"\u003croot>","type":"text","marks":[{"type":"code_inline"}]},{"text":" when you first compose a ","type":"text"},{"text":"\u003croot>/solutions/","type":"text","marks":[{"type":"code_inline"}]},{"text":" path, and pass a subagent the resolved path rather than the config.","type":"text"}]},{"type":"paragraph","content":[{"text":"\u003c!-- ce-docs-root:start --> ","type":"text"},{"text":"Resolve the CE artifact root ","type":"text","marks":[{"type":"strong"}]},{"text":"\u003croot>","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" before composing any artifact path.","type":"text","marks":[{"type":"strong"}]}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Read","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"docs_root","type":"text","marks":[{"type":"code_inline"}]},{"text":" from ","type":"text"},{"text":"\u003crepo-root>/.compound-engineering/config.yaml","type":"text","marks":[{"type":"code_inline"}]},{"text":" only (","type":"text"},{"text":"\u003crepo-root>","type":"text","marks":[{"type":"code_inline"}]},{"text":" = ","type":"text"},{"text":"git rev-parse --show-toplevel","type":"text","marks":[{"type":"code_inline"}]},{"text":"). Do not read it from ","type":"text"},{"text":"config.local.yaml","type":"text","marks":[{"type":"code_inline"}]},{"text":". Unset -> ","type":"text"},{"text":"\u003croot>","type":"text","marks":[{"type":"code_inline"}]},{"text":" is ","type":"text"},{"text":"docs","type":"text","marks":[{"type":"code_inline"}]},{"text":", exactly as before.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Validate","type":"text","marks":[{"type":"strong"}]},{"text":" a set value: a repo-relative directory whose real, symlink-resolved path stays inside the repo and is neither the repo root nor under ","type":"text"},{"text":".git/","type":"text","marks":[{"type":"code_inline"}]},{"text":". Otherwise stop with an error naming ","type":"text"},{"text":"docs_root","type":"text","marks":[{"type":"code_inline"}]},{"text":" and the value -- never fall back to ","type":"text"},{"text":"docs","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"\u003croot>","type":"text","marks":[{"type":"code_inline"}]},{"text":" as the sole artifact location: create it if absent, compose each path as ","type":"text"},{"text":"\u003croot>/\u003csubdir>","type":"text","marks":[{"type":"code_inline"}]},{"text":" with this skill's own subdirectory, and never also read ","type":"text"},{"text":"docs","type":"text","marks":[{"type":"code_inline"}]},{"text":". \u003c!-- ce-docs-root:end -->","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Write boundary","type":"text"}]},{"type":"paragraph","content":[{"text":"Only the orchestrator writes product files.","type":"text","marks":[{"type":"strong"}]},{"text":" Phase 1 subagents write to per-run scratch only, and never touch ","type":"text"},{"text":"\u003croot>/","type":"text","marks":[{"type":"code_inline"}]},{"text":", project instruction files, or any other tracked path.","type":"text"}]},{"type":"paragraph","content":[{"text":"The orchestrator writes what ","type":"text"},{"text":"references/form.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" chooses, in every mode and without asking: at most one added test, lint rule, or condition in an existing ","type":"text"},{"text":"CODING_STANDARDS.md","type":"text","marks":[{"type":"code_inline"}]},{"text":", and at most one learning under ","type":"text"},{"text":"\u003croot>/solutions/","type":"text","marks":[{"type":"code_inline"}]},{"text":". It never changes product code. It also writes two maintenance side effects that their own step describes: ","type":"text"},{"text":"CONCEPTS.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" during vocabulary capture, and — ","type":"text"},{"text":"only in interactive Full mode after consent","type":"text","marks":[{"type":"strong"}]},{"text":" — a small discoverability line in a project instruction file. Two further writes exist ","type":"text"},{"text":"only in interactive Full mode when the user selects them at the assembly destination step","type":"text","marks":[{"type":"strong"}]},{"text":": a rule file inside a writable declared Compound Pack, and the ","type":"text"},{"text":"packs:","type":"text","marks":[{"type":"code_inline"}]},{"text":" entry appended to ","type":"text"},{"text":".compound-engineering/config.yaml","type":"text","marks":[{"type":"code_inline"}]},{"text":". An instruction file is only ever edited, never created. Nothing else in the tree is written. Edits to ","type":"text"},{"text":"other","type":"text","marks":[{"type":"em"}]},{"text":" docs belong to ","type":"text"},{"text":"ce-compound-refresh","type":"text","marks":[{"type":"code_inline"}]},{"text":", which this skill recommends or invokes with a narrow scope but never stands in for.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Choosing the path","type":"text"}]},{"type":"paragraph","content":[{"text":"Read ","type":"text","marks":[{"type":"strong"}]},{"text":"references/modes.md","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" before step 1.","type":"text","marks":[{"type":"strong"}]},{"text":" An interactive run picks its own depth rather than asking the user. Default to ","type":"text"},{"text":"Full","type":"text","marks":[{"type":"strong"}]},{"text":". Choose ","type":"text"},{"text":"Lightweight","type":"text","marks":[{"type":"strong"}]},{"text":" only when the session is near its context limit.","type":"text"}]},{"type":"paragraph","content":[{"text":"Lightweight mode skips session history entirely; non-interactive Full runs the same automatic probe, which asks nothing and so preserves the non-interactive contract.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Full Mode","type":"text"}]},{"type":"paragraph","content":[{"text":"Run these in order. Each reference is a required read at the step that names it.","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Research","type":"text","marks":[{"type":"strong"}]},{"text":" — read ","type":"text"},{"text":"references/research.md","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Session history","type":"text","marks":[{"type":"strong"}]},{"text":" — read ","type":"text"},{"text":"references/session-history.md","type":"text","marks":[{"type":"code_inline"}]},{"text":", and start it ","type":"text"},{"text":"after","type":"text","marks":[{"type":"em"}]},{"text":" dispatching the Related Docs Finder so it overlaps your drafting rather than serializing.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Form, assembly, and write","type":"text","marks":[{"type":"strong"}]},{"text":" — wait for every Phase 1 input, read ","type":"text"},{"text":"references/form.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" and write what it chooses, then read ","type":"text"},{"text":"references/assembly.md","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Refresh check and discoverability","type":"text","marks":[{"type":"strong"}]},{"text":" — read ","type":"text"},{"text":"references/refresh-and-discoverability.md","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Optional enhancement","type":"text","marks":[{"type":"strong"}]},{"text":" — read ","type":"text"},{"text":"references/enhancement.md","type":"text","marks":[{"type":"code_inline"}]},{"text":". Interactive only.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Report","type":"text","marks":[{"type":"strong"}]},{"text":" — read ","type":"text"},{"text":"references/report.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" for the report shape your mode must emit and how the skill ends.","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Lightweight Mode","type":"text","marks":[{"type":"strong"}]},{"text":" replaces steps 1-6 with a single pass; read ","type":"text"},{"text":"references/lightweight.md","type":"text","marks":[{"type":"code_inline"}]},{"text":", which defines its own completion output for both modes.","type":"text"}]}]},"metadata":{"date":"2026-11-02","name":"ce-compound","author":"@skillopedia","source":{"stars":25863,"repo_name":"compound-engineering-plugin","origin_url":"https://github.com/EveryInc/compound-engineering-plugin/blob/HEAD/skills/ce-compound/SKILL.md","repo_owner":"EveryInc","body_sha256":"e14c0bdaef3c43e21e339192c7f6dcca8f79f8377647daf8411bf2852d586f57","cluster_key":"089746d58cd4e0bfeadffa778e87da333aa0df9c1999cc0e74d0fc21d3f3f2fa","clean_bundle":{"format":"clean-skill-bundle-v1","source":"EveryInc/compound-engineering-plugin/skills/ce-compound/SKILL.md","text_bytes":269378,"source_commit":"77cc431daaebea7ee9f33ac530a66eec5404572b","skills_sh_rank":5507,"attachment_count":28,"text_attachments":28,"attachment_storage":"skill_attachments","binary_attachments":0},"cluster_size":1,"skill_md_path":"skills/ce-compound/SKILL.md","import_metadata":{"date":"2026-11-02","author":"@skillopedia","version":"v1","category":"general","category_label":"General"},"exact_dupes_collapsed_into_this":0},"version":"v1","category":"general","import_tag":"skills-sh-top20k-2026-11","description":"Capture a solved problem's lesson in its cheapest durable form, a test or lint rule, a coding standard, or a repo learning. Use when verified work produced non-obvious reasoning absent from its final code, tests, or existing docs; avoid routine fixes whose artifacts already explain the lesson.","argument-hint":"[optional: brief context] [mode:non-interactive] [depth:lightweight|full]"}},"renderedAt":1793661530410}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.