bmad-agent-po

---
id: [the entry's id in tickets.toml; the folder's `next_id` if no entry]   # publish to a tracker writes tracker_id and remote
type: bug
title: "[What is wrong, from the user's view]"
parent: [folder name of the epic, or of the initiative when there are no epics; none for a standalone bug in backlog/]
covers: []
after: []   # prerequisites only: a sibling's id; "<epic id>.<entry id>" in another epic; epic-<slug> for that whole epic
assignee: ""   # written by `tickets.py mirror` from a tracker; otherwise assignee, blocked_at, blocked_reason live in the plan
# status lives in the plan beside this file; tracker_status by a tracker sync
refined: false   # true once the user approves its criteria
hitl: false
risk: [low|medium|high]
severity: [P0|P1|P2|P3]   # P0 outage, data loss, security exposure; P1 core function broken, no workaround; P2 impaired, workaround exists; P3 cosmetic
---

<!-- Refined shape. Leave a done or dropped ticket as written. -->

# [Title]

## Description

[What the user sees go wrong and what should happen instead. 2–4 sentences.]

## Reproduction

[Exact steps, environment, actual versus expected. Logs or errors verbatim.]

## Cause Hypothesis

[Where the defect likely lives and why. Never a prescribed fix.]

## Acceptance Criteria

1. **[The expected behavior holds]**
   **Given** [the state from the reproduction]
   **When** [the reproduction steps are followed]
   **Then** [what should now happen]
2. **Tests cover the condition found and fixed**
   **Given** the test suite
   **When** it runs
   **Then** a test that fails on the defect and passes on the fix covers the reproduction, and one covers each related case the fix touched
3. **Or: no change is needed, with proof**
   **Given** the reproduction
   **When** it is run on the current code
   **Then** the expected behavior already holds, or the report was mistaken, with the evidence recorded in Notes — this supersedes 1 and 2

## References

- parent — [path or url, or none]
- [type — path, section]
- [logs, screenshots, or sample data — location]

## Notes

[Only what is not in the repo or the source. Cut if empty.]

- Decision: [a choice the user made, dated]
- Assumption: [a choice made while drafting that the user has not confirmed]
- Open question: [what is not settled]

---
id: [the entry's id in tickets.toml; the folder's `next_id` for a ticket with no entry]   # tracker_id and remote are written at publish on a tracker
type: spike
title: "[The question this answers]"
parent: [folder name of the epic, or of the initiative when there are no epics; none for a standalone ticket in backlog/]
covers: []
after: []   # prerequisites only: a sibling's id; "<epic id>.<entry id>" in another epic; epic-<slug> for that whole epic
assignee: ""   # a tracker's assignee, written by `tickets.py mirror`; otherwise assignee, blocked_at, and blocked_reason live in the plan
# status lives in the plan beside this file; tracker_status is written by a tracker sync
refined: false   # true once the user approves its full criteria; a pulled spike has this line only when its entry says `refine = true`
hitl: true
risk: [low|medium|high]
---

<!-- The refined shape. Leave a done or dropped ticket as written. Fill each slot concretely: named people, file paths, numbers. -->

# [Title]

## Description

[The question, why it is open, and which tickets wait on the answer.]

## Approach

[Numbered steps: what to research, design and prototype, and who signs off. The time box. What good enough looks like.]

## Acceptance Criteria

1. **[The candidates are compared on the criteria that decide]**
   Verify: [where the comparison is recorded]
2. **[The prototype shows the chosen answer works for the cases that matter]**
   Verify: [where the prototype and its results live]
3. **[The answer is recorded, with its consequences, where the waiting tickets can find it]**
   Verify: [where]
4. **[The people who must accept it have]**
   Verify: [who, recorded where]

## References

- parent — [path or url]
- [type — path, section]

## Notes

[Assumptions and open questions about the spike itself, each marked. Cut if empty.]

---
tracker_id: ""   # with remote (url): set at publish on a tracker, on a story by `tickets.py mirror`; cut on the repo store
id: [story: its tickets.toml entry id, or the folder's `next_id` if none; cut on a container]
type: [initiative | epic | story]
title: "[Container: its outcome. Story: what works when done]"
parent: [the parent container's folder name; none for an initiative or a backlog/ story]
covers: [epic: parent ids it owns; story: epic ids it serves; cut on an initiative. Constraints go in References, not here]
sources: []   # every input SPEC derived Requirements from, paths from {project-root}; each is re-read on update; never the memlog
companions: []   # files downstream reads beside this ticket, written or adopted by SPEC, paths from {project-root}
after: []   # container: epics it waits on; epic order is in tickets.toml. Story: prerequisites only: sibling id, "<epic id>.<entry id>", or epic-<slug>
assignee: ""   # story: only a tracker's, via `tickets.py mirror`; cut on a container
# container: add status: in-progress | done | dropped when its first ticket starts
# a story's status, assignee, blocked_at, blocked_reason live in the plan beside this file, not here; tracker_status from tracker sync
refined: false   # story: true once the user approves its criteria; on a pulled story only if its entry has `refine = true`. Epic: false only when slicing writes it thin, until refined; else cut. Cut on an initiative
hitl: false   # true when a person does part of the work; story only, cut on a container
risk: [low|medium|high; container: a best guess until sliced, then its children's highest]
---

<!-- At create a container gets all but Requirements and Constraints, which SPEC writes from an input. An epic written at slicing gets Description, Outcome, Boundaries, Done when as far as known, and References to the parent; Requirements and Constraints come when it is refined. A pulled or reviewed story: Description, Verify: line, References, local Notes; refining adds numbered criteria and Boundaries. Done and dropped tickets stay as written. An initiative is the business outcome its epics serve, usually spanning quarters and more than one boundary; for a solo developer or small team a whole product is a fine initiative. -->

# [Title]

## Description

[Container: a paragraph, what is true when done and why, pointing at the inputs or parent. Story: one sentence; with numbered criteria, 2–4 on how it advances the epic, not restating them.]

## Outcome

[Container: one sentence: for whom, what changes, the source's success signal named not restated. Story: cut.]

## Requirements

[Container: SPEC writes this from the ticket's `.memlog.md`; slicing never touches it. Initiative: `R<n> (<source ids>): <checkable result>`, ids prefixed with the source's References type when ambiguous (`prd FR-2`), its section when it has none. Epic: its owned parent lines, `R<n> (parent R<m>): <the line>`, deepened only as inputs require; split or added lines name a parent id or source section, all lines a source section when `covers` is empty. Story: only `Verify: how it will be checked`. No epic, `refine = true`, or on request: numbered criteria, each naming the epic ids it proves; a criterion states a rule, not an instance, with its failure path.]

1. **[Behavior name]** (R<m>)
   **Given** [state before]
   **When** [the one action]
   **Then** [what is observable]
   **And** [another observation, if any]

## Constraints

[SPEC writes this. Initiative: each design-bending non-negotiable, citing its source id or section. Epic: the parent's that bind it. Story: only ones beyond its epic's; behavior to keep goes under Boundaries.]

## Done when

[Container: three to six end checks from the source's measures and limits: an initiative's business outcomes, an epic's deliverables at its Requirements' altitude, production delivery among them. Story: cut.]

## Boundaries

[Container: its boundary and what it is not, from the source's non-goals. Initiative: plus a line per touched unit with no epic, and a one-sentence tracer path across epics. Epic: what belongs to siblings. Story: with numbered criteria, adjacent behavior an implementation could break, not files; else cut.]

- Touch point: [unit] — [what is used there]; owner: epic-[slug]
- Must not change: [behavior]

## References

[`type — location, section`, location a path from {project-root} or a url. A line per companion SPEC writes or adopts, typed by content (glossary, design, architecture, prd). Initiative: section pointers into its sources and companions. Epic: the parent's Requirements, then the architecture spine section and design it adopts. Story: the parent, plus what the epic's lack.]

- parent — [path or url, section Requirements]
- [companion — a file beside this one, named for its content]
- [design — an adopted companion, such as DESIGN.md in the ux-<slug>/ run beside this file, or a url]

## Notes

[Only what is unsettled here and not in the repo or source, one marked line each; cut if empty. SPEC writes Assumption: and Open question: from the memlog; slicing and refining, the rest.]

- Assumption: [an unconfirmed choice; once confirmed, a dated Decision]
- Open question: [container: what the source leaves open, and who waits; story: only its own, answered in its work]
- Unknown: [the gap and the entries waiting on it; gates an entry only once copied to its `unknown` key]
- Parked: [an unplaced requirement id, why, with the user's knowledge]
- Decision: [the user's choice, dated; a declined suggestion too]
- Source conflict: [id or section — source vs code or another source]
- Waits on [epic] because: [reason, one per after entry]
- Dropped: [why, before status goes to dropped]

# A container's agreed children, in its tickets/ folder; tickets.py ignores unknown keys.
# Table order is build order. `id` is permanent, never reused.
# One kind of table per file: an initiative with no epics uses [[entry]] tables, a parent initiative [[initiative]] tables.
# No status here: the plan carries `status`; `tickets.py mirror` writes `tracker_status` to the leaf.
# next_id = 7 # written by `tickets.py move`; the folder's counter never goes below it

# ---- In an epic's tickets/ folder: one table per planned story, spike, or bug ----

[[entry]]
id = 1 # leaf file's `id:`
type = "story" # story | spike | bug
title = "Cart service scaffold"
description = "Stands up the cart service with one add-item path through UI, API, and store."
verify = "A shopper adds one item on the deployed site and sees it in the cart."
covers = ["R1"] # epic Requirements ids
after = [] # prerequisites only, not sequence
hitl = true # a person runs the starter
risk = "low" # low | medium | high
# estimate = 2 # only if the user gives one
# unknown = "..." # one-sentence user question; `tickets.py next` blocks the entry until removed
# references = ["architecture-checkout/architecture-checkout.md#ad-8"] # beyond the epic's References; pull copies to the ticket
# notes = ["..."] # user's words only; pull copies to Notes
# refine = true # refine to full acceptance criteria first
# plan_checkpoint = true # a person reviews the ticket or plan first
# done_checkpoint = true # pause after it until a person says continue

[[entry]]
id = 2
type = "story"
title = "Apply and refuse discount codes"
description = "Adds code validation to the pricing call and shows the refusal reason."
verify = "A valid code lowers the total; an expired one is refused with its reason."
covers = ["R2", "R3"]
after = [1, "1.3"] # sibling id; "<epic id>.<entry id>"; "epic-<slug>"
hitl = false
risk = "medium"

# ---- In an initiative's tickets/ folder: one table per epic ----

[[epic]]
id = 1
slug = "epic-pricing-rules" # the epic's folder, inside this tickets/ folder
title = "Pricing rules"
description = "Every price a shopper sees comes from one pricing function, deployed." # what is true when it is done; seeds the envelope's Outcome
covers = ["R1", "R2"]

[[epic]]
id = 2
slug = "epic-cart-rules"
title = "Shoppers manage their cart"
description = "A shopper adds, changes and removes items on the deployed site and the total follows."
covers = ["R3", "R4", "R5"]
after = [{ epic = 1, needs = "the one-function pricing contract" }] # entries carry their own `after`

# ---- In a parent initiative's tickets/ folder: one table per child initiative ----

[[initiative]]
id = 1
slug = "initiative-storefront" # the child's folder, beside the parent's
title = "Shoppers buy online"
description = "A shopper finds, prices and pays for an order on the live site."
covers = ["R1", "R2"]

[[initiative]]
id = 2
slug = "initiative-fulfillment"
title = "Orders ship themselves"
description = "A paid order is picked, packed and tracked without manual steps."
covers = ["R3"]
after = [{ initiative = 1, needs = "the order-paid event" }]

[skill]
bmod = "bmod-method"
source = "github:bmad-code-org/BMAD-METHOD/skills"
scripts = ["scripts/tickets.py"]

# Store starter: GitHub Issues. `scripts/read_store.py` reads this file as the base layer; the project's
# _bmad/custom/ticketing-store-config.toml names the store, fills the empty fields below, and replaces
# any other key here, key by key.
# It updates with the skill and holds only what is specific to this store: the rules every store shares
# are in references/board.md. Comments and `description` are never loaded; `description` is for choosing a store.
#
# One tool: the gh CLI, 2.94 or later (sub-issues and dependencies). Labels carry type and status,
# sub-issues carry hierarchy, blocked-by relations carry blocking, milestones put initiatives on the roadmap.
# Using the defaults below: the repo needs the type, status, and hitl labels; `setup` creates them.

description = "GitHub Issues through the gh CLI: labels, sub-issues, blocked-by, milestones. For teams whose code and planning already live on GitHub."

[tickets]
store = "github"
key = ""    # ids are the repo's issue numbers; leave empty
repo = ""   # OWNER/REPO; empty = the repo of the current checkout (origin)
access = """
Use the gh CLI against `repo`, passing `-R <repo>` on every call when it is set. Needs gh 2.94+; check `--help` when a flag is doubted.
"""

fields = """
risk is a label risk:<value>; severity is a label P0-P3. `setup` creates them with the other
labels; publish and update set them like any label. estimate is a label est:<value> (`setup` creates none; add on first use). Empty field: no label.
"""

reference = """
GitHub renders urls as links; nothing more to do.
"""

[verbs]
setup = """
Connect, unless `gh auth status` already shows a login: `gh auth login`.
Labels must exist before they are used. List the repo's labels (`gh label list --limit 1000 --json name`)
and run `gh label create <name>` for each of these it lacks, comparing names ignoring case: the
[tickets.types] values except initiative (a milestone), the [tickets.status] values except done and
dropped (closed states), `hitl`, risk:low, risk:medium, risk:high, and P0-P3. A create that fails with
"already exists" found the label after all: leave it. Never `--force`: it recolours a label the repo
already has.
"""

write = """
One issue per new ticket. tracker_id is the issue number and remote its url.
Create with content only: `gh issue create --title <t> --body-file <b> --label <type>,<[tickets.status].<state>>[,hitl,risk:<r>,P<n>]
[--milestone <initiative>]`, and mirror tracker_id and remote from the url it prints before anything else. Then add the
relations with `gh issue edit <n> --parent <n> --add-blocked-by <n,n>`, leaving out any that `gh issue view <n> --json
parent,blockedBy` already shows, since adding one twice fails. Keep --parent and --blocked-by off the create: gh adds them
after the issue exists, and when that step fails it exits non-zero without printing the url, so a rerun files the ticket
twice. A create that fails with no url may still have made the issue: before creating again, list the newest issues
(`gh issue list --state all --limit 100 --json number,title,body,createdAt`) twice, a few seconds apart since the list can
trail a create, and take the one whose title and body are what was sent. None matches and the list does not reach back to
the failed create: ask before creating again. Changes go through `gh issue edit <n>` (--body-file, --add-assignee,
--add-blocked-by, label swaps; --add-sub-issue on the parent). No parent: omit --parent; no initiative: omit --milestone.
Status is a label swap on an open issue. done: remove the status label and `gh issue close <n>`;
dropped: the same with `--reason "not planned"`. An initiative is a
milestone (`gh api repos/{owner}/{repo}/milestones -f title=<name>`) holding its tree's issues.
"""

query = """
One issue: `gh issue view <n> --json
number,title,labels,assignees,state,stateReason,milestone,parent,blockedBy,subIssuesSummary,body`.
Type, hitl, risk, severity, and status are labels — except done and dropped, the closed states per
[tickets.status]; after is blockedBy; the body's covers holds the source requirement ids. Children: `gh issue view <n>
--json subIssues`. Words: `gh issue list --search "<words>" --state all`, narrowed by
--label. An initiative's issues are the ones in its milestone; a standalone ticket has no milestone.
"""

# BMad type -> this store's type.
[tickets.types]
initiative = "milestone"   # an initiative is a milestone, not an issue; its epics carry it
epic = "epic"
story = "story"
spike = "spike"
bug = "bug"

# BMad state -> this store's status.
[tickets.status]
backlog = "backlog"
in-progress = "in-progress"
review = "review"
done = "closed"   # state, not a label
dropped = "closed: not planned"   # state, not a label

# Store starter: Jira Cloud. `scripts/read_store.py` reads this file as the base layer; the project's
# _bmad/custom/ticketing-store-config.toml names the store, fills the empty fields below, and replaces
# any other key here, key by key.
# It updates with the skill and holds only what is specific to this store: the rules every store shares
# are in references/board.md. Comments and `description` are never loaded; `description` is for choosing a store.
#
# One tool: acli, Atlassian's official CLI (`acli jira workitem ...`): create with --parent, transition,
# links, comments, assign, JQL. The Atlassian Rovo MCP server (https://mcp.atlassian.com/v1/mcp/authv2)
# is the alternative for an org that prefers MCP; it has no link-creation tool.

description = "Jira Cloud through acli: work types, transitions, links, JQL. For teams already on Jira, usually with an admin who can adjust the project workflow."

[tickets]
store = "jira"
key = ""    # the default Jira project key, e.g. "SHOP"; Jira assigns the number. A container's own `key` overrides it for everything under it
site = ""   # https://yourorg.atlassian.net
access = """
Use acli against `site`. Custom fields go through `--from-json`; check a subcommand's spelling
with `acli jira workitem --help` before first use.
"""

fields = """
severity is Jira's priority: P0 = Highest, P1 = High, P2 = Medium, P3 = Low; set it at create
(--priority, or --from-json when the flag is absent) and change it the same way. risk is a label
risk-low, risk-medium, or risk-high. estimate is the Story Points field on a story and a label est:<value> on an epic, through --from-json. Empty field: default priority, no label.
"""

reference = """
acli creates no web links; the description line is the reference.
"""

[verbs]
setup = """
Connect, unless acli is already installed and signed in to `site`: install acli (on macOS,
`brew tap atlassian/homebrew-acli && brew install acli`; elsewhere, per Atlassian's acli install
guide), then `acli jira auth login --web --site <site>`.
acli cannot create work types or statuses; those live in the project's settings. At first publish,
read an existing work item (`acli jira workitem view <key> --json --fields issuetype,status`) and
the project's work types, compare with [tickets.types] and [tickets.status], and tell the user
exactly what to add in Jira or which map entry to set in the project's store config. Jira's defaults (To Do, In
Progress, Done; resolution Won't Do) need no setup; [tickets.status] also expects an In Review status.
Do not ask for a level above Epic: by default the initiative stays local and everything below it
publishes. A team-managed project's admin adds a status as a board column under
Project settings, Board; a company-managed project needs a Jira admin to add it to the workflow.
"""

write = """
The project is the ticket's own `key`, else that of the nearest container above it, else `key` here.
tracker_id is the work item's key and remote its url.
Create:
`acli jira workitem create --project <key> --type <per [tickets.types]> --summary <t>
--description-file <b> --parent <key> --label <type>[,hitl,risk-<risk>]`, priority per `fields` on
a bug. Changes: `edit` (body, labels), `assign`, `transition --status <[tickets.status].<state>>` —
status never moves by edit — `link create --out <blocker> --in <blocked> --type "Blocks"`,
`comment create`. No parent: omit --parent — the item sits loose in the project. Custom fields go through --from-json; when dropped shares a status with done,
set the resolution the same way so the two stay distinct.
Epics sit loose in the project while [tickets.types].initiative is empty. Set that type only when the
project has a level above Epic; then epics take it as --parent.
"""

query = """
One work item: `acli jira workitem view <key> --json --fields
key,issuetype,summary,status,assignee,parent,labels,issuelinks,resolution,description`. after is the
"is blocked by" links; Done with resolution Won't Do is dropped; the description's covers holds the source requirement ids. Children:
`search --jql "parent = <key>"`. Words: `search --jql 'project = <key> AND text ~ "<words>"'`,
narrowed by issuetype or status per the maps.
"""

# BMad type -> this store's type.
[tickets.types]
initiative = ""   # empty: the initiative stays local. Jira's default hierarchy is Epic > Story > Subtask; a level above Epic needs Plans (Premium or Enterprise). Set to "Initiative" only when the project has that level.
epic = "Epic"
story = "Story"
spike = "Task"   # plus label spike
bug = "Bug"

# BMad state -> this store's status.
[tickets.status]
backlog = "To Do"
in-progress = "In Progress"
review = "In Review"   # not a default Jira status; `setup` asks for it. With no review status, set it to the in-progress one
done = "Done"
dropped = "Done"   # Jira default has no separate status; resolution Won't Do distinguishes it

# Store starter: Linear. `scripts/read_store.py` reads this file as the base layer; the project's
# _bmad/custom/ticketing-store-config.toml names the store, fills the empty fields below, and replaces
# any other key here, key by key.
# It updates with the skill and holds only what is specific to this store: the rules every store shares
# are in references/board.md. Comments and `description` are never loaded; `description` is for choosing a store.
#
# One tool: Linear's official MCP server. Initiatives hold projects, parent issues are epics, labels carry
# type, relations carry blocking. Linear ships no CLI; community CLIs are not used here.

description = "Linear through its MCP server: projects, parent issues, relations, labels. For product teams on Linear."

[tickets]
store = "linear"
key = ""    # Linear team key, e.g. "SHOP"; Linear assigns the number
team = ""   # the default team, key or name. A container's own `key` overrides it for everything under it
access = """
Use the Linear MCP (create_issue, update_issue, create_project, update_project, get_issue,
list_issues, list_issue_statuses, create_issue_label, create_comment) against `team`. Confirm from the tool schema whether parent
and blockedBy take identifiers (SHOP-12) or ids, and the exact tool names against the server's list at first use.
"""

fields = """
severity is Linear's priority, set on the issue: P0 = Urgent, P1 = High, P2 = Medium, P3 = Low.
risk is a label risk:<value>; `setup` creates the three. estimate is Linear's estimate field on a story; a label est:<value> on an epic. Empty field: no priority, no label.
"""

reference = """
Also pass the url in the issue's links so it shows in the sidebar.
"""

[verbs]
setup = """
Connect, unless the Linear tools already answer: add the MCP server https://mcp.linear.app/mcp to
the agent tool in use (in Claude Code, `claude mcp add --transport http linear-server https://mcp.linear.app/mcp`),
then sign in over OAuth. Do not use the deprecated /sse endpoint.
`create_issue_label` for every non-empty value in [tickets.types], `hitl`, and risk:low, risk:medium,
risk:high that the team lacks. States
cannot be created from the MCP: `list_issue_statuses`, compare with [tickets.status], and tell the
user what to add under Settings, Teams, Issue statuses, or which map entry to set in the project's
store config. Linear's defaults (Backlog, Todo, In Progress, Done, Canceled) need no setup; [tickets.status]
also expects an In Review state in the Started group.
"""

write = """
The team is the ticket's own `key`, else that of the nearest container above it, else `team` here.
tracker_id is the issue's identifier and remote its url.
`create_issue` creates, `update_issue` changes — title, description, state, assignee, parent,
blockedBy, labels, priority, project. The issue's state is [tickets.status].<state>. Labels carry type, hitl, and
risk:<value>; priority carries severity per `fields`. A standalone ticket: omit parent and project.
Epics are parent issues. Set [tickets.types].initiative to map onto a Linear Initiative only when the
server exposes initiative tools.
"""

query = """
One issue: `get_issue` with includeRelations — labels, priority, state, assignee, parent,
project, relations, description; after is the "blocked by" relations, and the description's
covers holds the source requirement ids. Many: `list_issues` by project, parent, label, state, or
query words.
"""

# BMad type -> this store's type.
[tickets.types]
initiative = ""   # empty: the initiative stays local. Linear has no Epic object and Projects cannot nest; set to "initiative" only to map onto Linear Initiatives, which nest five deep and allow several parents, so the tree cannot round-trip.
epic = "epic"   # label on a parent issue
story = "story"
spike = "spike"
bug = "bug"

# BMad state -> this store's status.
[tickets.status]
backlog = "Backlog"   # or "Todo" — the team's choice
in-progress = "In Progress"
review = "In Review"   # not a default Linear state; `setup` asks for it. With no review state, set it to the in-progress one
done = "Done"
dropped = "Canceled"

# Store starter: Notion. `scripts/read_store.py` reads this file as the base layer; the project's
# _bmad/custom/ticketing-store-config.toml names the store, fills the empty fields below, and replaces
# any other key here, key by key.
# It updates with the skill and holds only what is specific to this store: the rules every store shares
# are in references/board.md. Comments and `description` are never loaded; `description` is for choosing a store.
#
# One tool: Notion's official MCP server. One database holds every ticket; properties carry type,
# status, assignee, and hitl; two self-relations carry hierarchy (Parent) and blocking (Blocked by);
# a board view grouped by Status is the kanban. `setup` creates all of it.

description = "Notion through its MCP server: one tickets database with a board view, relations for hierarchy and blocking. For teams that plan in Notion and want tickets beside their docs."

[tickets]
store = "notion"
key = ""        # prefix of the database's ID property, e.g. "SHOP" gives SHOP-12; empty = page ids only
database = ""   # url or id of the tickets database; empty = `setup` creates it
access = """
Use the Notion MCP (notion-fetch, notion-create-pages, notion-update-page,
notion-query-data-sources, notion-create-comment, notion-get-users) against `database`. Confirm
from the tool schema how relation and person properties are passed; `notion-get-users` resolves a name.
An empty `database` means `setup` has not run.
"""

fields = """
risk and severity are select properties: Risk (low, medium, high) and Severity (P0-P3); `setup`
creates them, publish and update set them. estimate is a text property Estimate, created with the rest. Empty field: property left empty.
"""

reference = """
Notion renders urls as links; a Notion page url becomes a mention.
"""

[verbs]
setup = """
Connect, unless the Notion tools already answer: add the MCP server https://mcp.notion.com/mcp to
the agent tool in use (in Claude Code, `claude mcp add --transport http notion https://mcp.notion.com/mcp`),
then sign in over OAuth. Either way, share the database, or the page it will be created under, with
the connection.
When `database` is empty: ask which page to create under, then `notion-create-database` named
Tickets with properties Title (title), Type (select, options per [tickets.types]), Status (status,
options per [tickets.status]), Parent (relation to this database), Blocked by (relation to this
database), Assignee (person), Covers (text), Risk (select: low, medium, high), Severity (select:
P0, P1, P2, P3), Estimate (text), hitl (checkbox), and, when `key` is set, ID (unique id
with that prefix). Then `notion-create-view`: a board grouped by Status. Give the user the url and
put it in `database` in the project's store config.
When `database` is set: `notion-fetch` it and compare its properties and options with the maps;
add a missing option with `notion-update-data-source`, and tell the user about anything the
schema needs a person to change.
"""

write = """
tracker_id is the ID property when `key` is set, else the page id; remote is the page url.
`notion-create-pages` into `database` creates; `notion-update-page` changes any property or the
content — Status, Assignee, Parent, Blocked by, Covers, Risk, Severity, hitl. Status is [tickets.status].<state>. No parent: leave the Parent relation empty. An initiative
is a page of Type initiative with no Parent. The body is the page content.
"""

query = """
One page: `notion-fetch` — Title, Type, Status, Assignee, Parent, Blocked by, Covers, Risk,
Severity, hitl, and the content; after is Blocked by. Many: `notion-query-data-sources` on `database` filtered by any
property; `notion-search` scoped to `database` for a looser match.
"""

# BMad type -> this store's type (Type select options; `setup` creates them).
[tickets.types]
initiative = "initiative"
epic = "epic"
story = "story"
spike = "spike"
bug = "bug"

# BMad state -> this store's status (Status property options; `setup` creates them).
[tickets.status]
backlog = "Backlog"
in-progress = "In Progress"
review = "Review"
done = "Done"
dropped = "Dropped"

# Store starter: tickets as files in the repo, the default. `scripts/read_store.py` reads this file as the
# base layer; the project's _bmad/custom/ticketing-store-config.toml names the store and replaces any key
# here, key by key. With no project file this starter is the whole config.
# It updates with the skill and holds only what is specific to this store: the rules every store shares
# are in references/board.md. Comments and `description` are never loaded; `description` is for choosing a store.
#
# Tickets and artifacts live together as markdown files under `output_folder` (layout in workflow.md,
# The ticket tree), committed in the monorepo or in a repo of their own. Status and assignee change by commit, and the skill never pushes,
# so others see them only after a push and pull.

description = "Quick setup, and the default when nothing else is chosen. Tickets are markdown files under version control: in the monorepo, or in their own repo (recommended for mono and poly repos alike). No tracker or account needed. The simplest store — right for a solo developer or a very small team with no other tracking."

[tickets]
store = "repo"

access = """
Read and write the files under `output_folder`.
"""

fields = """
risk and severity live in the frontmatter (risk: low|medium|high on any ticket; severity: P0-P3
on bugs); estimate the same way. Nothing else carries them.
"""

[verbs]
setup = """
Tickets live in `output_folder` beside the documents. To keep them outside the project, set `output_folder`
under `[core]` in `_bmad/custom/config.toml` to a folder beside it — the session then launches one level
up — or in the workspace that already holds the repos. `git init` a new folder so tickets have
history; skip that when the user picks a folder already under git, the monorepo included. Then
create `backlog/` and the active initiative's folder there when they do not exist.
"""

write = """
Every change is an edit to the ticket's file, its `tickets.toml` entry, or, for a leaf's status and assignee, its plan through `tickets.py mark`.
Commit only the files you wrote on the current branch. Root not in a git repo: skip the commit and say so.
"""

query = """
The files are the store: `tickets.py find`, `next`, and `status` answer from them. By a frontmatter field: `grep -rl "^type: bug" <folder>`.
"""

# BMad type -> this store's type.
[tickets.types]
initiative = "initiative"
epic = "epic"
story = "story"
spike = "spike"
bug = "bug"

# BMad state -> this store's status.
[tickets.status]
backlog = "backlog"
in-progress = "in-progress"
review = "review"
done = "done"
dropped = "dropped"

# Store starter: Trello. `scripts/read_store.py` reads this file as the base layer; the project's
# _bmad/custom/ticketing-store-config.toml names the store, fills the empty fields below, and replaces
# any other key here, key by key.
# It updates with the skill and holds only what is specific to this store: the rules every store shares
# are in references/board.md. Comments and `description` are never loaded; `description` is for choosing a store.
#
# One tool: the official Trello MCP server. Trello is flat: lists carry status, labels carry type, card
# urls in descriptions carry hierarchy and blocking. Using the defaults below: the board needs the lists
# Backlog, In Progress, Review, Done, Initiatives, Epics and a label per type plus hitl; `setup` says how. Members, comments, attachments, and label creation
# are not in the MCP yet. Weakest fit of the shipped configs; expect to tune it.

description = "Trello through its MCP server: lists as status, labels as type, card links as hierarchy. For small teams that run a Trello board and want to keep it."

[tickets]
store = "trello"
key = ""     # ids are card short links (from the card url); leave empty
board = ""   # the board url
access = """
Use the Trello MCP (trelloReadBoard, trelloReadCard, trelloWriteCard, trelloWriteChecklist,
trelloSearch) against `board`. Write tools take the server's ids, not urls: resolve a url with
`trelloReadCard get` first.
"""

fields = """
risk is a label risk:<value>; severity a label P0-P3. The MCP cannot create labels: `setup` asks
the user to add them with the type labels. estimate is an `estimate:` line in the description. Empty field: no label.
"""

reference = """
Attachments are not in the MCP yet; the description line is the reference.
"""

[verbs]
setup = """
Connect, unless the Trello tools already answer: add the MCP server https://mcp.trello.com/v1 to
the agent tool in use (in Claude Code, `claude mcp add --transport http trello https://mcp.trello.com/v1`),
then sign in over OAuth with write permission.
The MCP can create a board but not lists or labels. When `board` is empty, offer to create one;
then ask the user to add the lists in [tickets.status] (Backlog, In Progress, Review, Done) plus
Initiatives and Epics, and labels for every value in [tickets.types] plus hitl, risk:low, risk:medium, risk:high, and
P0-P3. Check with
`trelloReadBoard` before first publish and name what is still missing.
"""

write = """
tracker_id is the card's short link and remote its url.
`trelloWriteCard create` makes a card, `update` changes description or labels, `move` changes
list, `archive` is dropped. Status is the list [tickets.status].<state>; initiative cards live in the Initiatives list,
epics in Epics, both carrying a `status:` line at the top of the description. Labels carry type,
hitl, risk, severity. Trello has no fields for the rest — conventions in the description:
`parent: <card url>` as its first line, an `assignee:` line (no member tools yet), after as
card urls; the parent card keeps a checklist with each child's url. A standalone ticket: no parent line.
The body is the description, and the `assignee:` line stays in it.
"""

query = """
One card: `trelloReadCard get` by url — name, list, labels, description (assignee, parent,
covers, after live there), checklists; a container's children are its checklist items. Many:
`trelloSearch` scoped to `board`, or `trelloReadBoard` filtered by the maps.
"""

# BMad type -> this store's type.
[tickets.types]
initiative = "initiative"
epic = "epic"
story = "story"
spike = "spike"
bug = "bug"

# BMad state -> this store's status.
[tickets.status]
backlog = "Backlog"
in-progress = "In Progress"
review = "Review"
done = "Done"
dropped = "archived"

# DO NOT EDIT -- overwritten on every update. Override files:
# {project-root}/_bmad/custom/bmad-agent-po.toml (team)
# {project-root}/_bmad/custom/bmad-agent-po.user.toml (personal)
# One run: `--set workflow.<key>=<value>`.
# Merge: strings replace, lists append, tables merge key by key, arrays of tables by `code`.

[agent]
# Set in the skill frontmatter; renaming needs a custom agent.
name = "Sofia"
title = "Product Owner"

icon = "🎯"

# Steps before activation.
activation_steps_prepend = []

# Steps after facts, before dispatch.
activation_steps_append = []

# Session context: sentences, or `file:{project-root}/...` paths and globs.
persistent_facts = []

role = "Own the backlog: what each ticket requires, how the work is split and ordered, what is accepted as done, and what changes when the plan meets reality."
identity = "The one person who can say what is in and what is out. Every line on the backlog is one I can defend to a stakeholder and explain to whoever builds it, and every line names where it came from."
communication_style = "Plain and short: what is in, what is out, and why. I say no early and kindly, with the reason, and I ask only what is unsettled, with the answer I would pick."

principles = [
  "What is required is written in the ticket, once, with its source beside it; a requirement that lives only in a conversation does not exist.",
  "The backlog is ordered, not listed: the first ticket proves the riskiest path, and every ticket says what is true when it is done.",
  "Scope is decided, not discovered: in or out, said once, dated.",
  "Done is a check the owner runs, not a count of closed children.",
  "I propose the split and the order with my reasons; the user makes the call, and the decision is recorded so nobody is asked twice.",
]

# Each item: `skill` (installed skill name) or `prompt`, not both; a matching `code` replaces, else appends.

[[agent.menu]]
code = "SPEC"
description = "requirements from an input (PRD, brief, transcript, idea, notes) into a ticket; updated or validated"
prompt = "`{skill-root}/references/spec.md`"

[[agent.menu]]
code = "TICKETS"
description = "initiative started, authored, split into epics, or put under a parent; an epic refined; an epic's stories planned or re-sliced"
prompt = "`{skill-root}/references/slice.md`"

[[agent.menu]]
code = "REFINE"
description = "one ticket written, reviewed, or refined"
prompt = "`{skill-root}/references/ticket.md`"

[[agent.menu]]
code = "BOARD"
description = "tickets published, started, moved, assigned, blocked, closed, dropped; a tree cancelled"
prompt = "`{skill-root}/references/board.md`"

[[agent.menu]]
code = "VALIDATE"
description = "a ticket, set, or tree validated"
prompt = "`{skill-root}/references/validate.md`"

[[agent.menu]]
code = "REPLAN"
description = "a mid-flight change: impact, stale work, re-slice or drop"
prompt = "`{skill-root}/references/replan.md`"

[[agent.menu]]
code = "STORE"
description = "tracker connected; store reconfigured or switched"
prompt = "`{skill-root}/references/store-setup.md`"

[workflow]

# Team rules on the method go in `persistent_facts`, or AGENTS.md for build skills too.

# Before an input is read. Top of references/spec.md.
spec_before = ""
# After requirements are presented. End of Output in references/spec.md.
spec_after = ""
# Before slicing starts. Top of references/slice.md.
tickets_before = ""
# After an approved slice, story plan, or re-slice. End of references/slice.md.
tickets_after = ""
# Before ticket work. Top of references/ticket.md.
refine_before = ""
# After the approved ticket is saved. End of references/ticket.md.
refine_after = ""
# Before a board operation. Top of references/board.md.
board_before = ""
# After a board operation. End of references/board.md.
board_after = ""
# Before check agents launch. Top of references/validate.md.
validate_before = ""
# After findings merge. End of references/validate.md.
validate_after = ""
# Before assessing a change. Top of references/replan.md.
replan_before = ""
# After the reason is recorded. End of references/replan.md.
replan_after = ""
# Before choosing or changing the store. Top of references/store-setup.md.
store_before = ""
# After the store is proven. End of references/store-setup.md.
store_after = ""

# Initiative, epic, story template; keep a home for Description, Outcome, Requirements, Constraints, Done when, Boundaries, References, Notes (`Decision:`, `Dropped:` lines in Notes). `tickets.py pull` writes a fixed layout, not this template. Lands in references/ticket.md, slice.md and spec.md.
ticket_template = "{skill-root}/assets/ticket-template.md"
# Bug template: Reproduction, Cause Hypothesis, fix or no-fix criteria. Lands in references/ticket.md.
bug_template = "{skill-root}/assets/bug-template.md"
# Spike template: Approach, `Verify:` lines instead of Given/When/Then (deliberate). Lands in references/ticket.md.
spike_template = "{skill-root}/assets/spike-template.md"
# `tickets.toml` entry, epic and initiative table keys; `tickets.py` ignores unknown ones. Lands in references/slice.md.
tickets_template = "{skill-root}/assets/tickets-template.toml"
# Tracker publishing: at_inception (whole breakdown) or on_start (each ticket); user can override; other values halt the render; on the repo store, approval commits and this key does not apply. Lands in references/board.md, Publish.
publication = "at_inception"
# Epic boundary rule (team text or a readable pointer) replacing the one in references/slice.md. Lands in references/slice.md, Initiative into epics.
slice_to_epics = ""
# Team checks, appended to their section in references/validate-checks.md.
checks.ticket = []
checks.set = []  # every entry in an epic's breakdown
checks.tree = []  # the initiative and its epic envelopes
checks.dependencies = []  # missing prerequisites; with set and tree
checks.closure = []  # a container before it closes

# Self-validation. Lands in references/spec.md, Self-validate.
spec_self_validate = """
Before presenting any create or update, run two passes.

**Pass 1 — Coherence.** Check against Spec Law rules 1–6 and 8; fix failures and weak spots without inventing content. Unconfirmed calls become `Assumption:` lines; unfilled gaps, `Open question:` lines.

**Pass 2 — Preservation.** Every load-bearing source claim lands in the ticket or a companion. Log each wrapper-only drop to the memlog as an event.

Record each pass's verdict to `.memlog.md` (`append --type event`); mention a pass to the user only when it changed something they approved.\
"""

# Epic's first entry; the Set check asserts it as rendered and slicing records it as a `Decision:` line in the epic's Notes. Lands in references/slice.md, Epic into stories.
tickets_tracer_bullet = """
The first entry is the tracer bullet: the thinnest path through every layer the epic touches. Its setup, including any starter the user runs, is a hitl step. It and any demoable entry say what someone sees running when done. Next is the story the user is least sure of.\
"""

# Epic's closing story; the Set check asserts it as rendered. Lands in references/slice.md, Epic into stories.
tickets_refactor_sweep = """
An epic of more than three entries closes with a story "Refactor sweep" that waits on every other entry except a closing end-to-end suite. Its scope, set at its start from plans and review findings the epic deferred, is cleanup only; scope pushed out of another story is a new story. Propose it by default; if declined, record a `Decision:` line in the epic's Notes.\
"""

# Closing a container. Lands in references/board.md, Close, drop, move.
board_closure = """
Closing every child does not close the parent. Run the closure check in `validate.md` against its requirements and Done when; the user confirms completion.\
"""

# Epic sizing norm. Lands in references/slice.md, Epic into stories.
tickets_story_count = """
Eight to twelve stories is typical, not a limit. Past that, offer a split first; the user decides.\
"""

# Board

Read a ticket, and its remote state if published, before changing it. Commit changed files on the current branch when `output_folder` is in a git repo; never push.

`tickets.py --help` on a command gives its arguments and output keys.

{{ workflow.board_before }}

## Reading the tree

- `find`'s `<ref>` is `<epic id>.<entry id>`, an entry id with its epic folder, a tracker id, a file name, or an unbroken title phrase; for looser words, pick from `status`, asking when several fit. Its `story_file` is null until pulled; `plan` is the plan's path, existing or not.
- `tickets.py next [<folder>]` groups open tickets by next step; `tickets.py status [<folder>]` lists every ticket.
- `<folder>` is an epic folder name, `backlog`, or the initiative, by default the active one, where `next` and `status` add a `backlog` key to show. On a parent initiative they list its child initiatives under `initiatives`.
- With a tracker, sync first (Tracker sync) and pass `--synced` to `next`.

Act with the user on any non-empty key:

- `unpinned_after`: add the epic's `after` prerequisite to the entry that needs it.
- `undeclared_after`: add the awaited epic, with its `needs`, to its epic's `after` in the breakdown, or drop the prerequisite.
- `order_conflict`: reorder or merge the epics, or record why in a `Decision:` line.
- `drift`: make the leaf file's and the entry's `after` or `hitl` equal.
- `problems`: an orphan plan gets its `ticket` set to the entry id or file stem; an unknown status is fixed with `mark`.
- `waiting_on` on a `blocked` row: say what the ticket waits on.
- `unknown` on a `blocked` row: settle it, remove it, and add a dated `Decision:` line to the file's Notes, or its epic's when it has no file.

## Publish

On the repo store, approval commits (Conventions in `workflow.md`) and `publication` does not apply. On a tracker, publication is `{{ workflow.publication }}`: `at_inception` offers to publish the agreed set after validation; `on_start` publishes each ticket when it starts. An explicit user request overrides it.

Publish to a tracker through `write`:

- Write the leaf's file first (`tickets.py pull` if none); send it minus the frontmatter lines `tracker_id`, `remote`, `tracker_status`, `status`, and `assignee`.
- Create parents and prerequisites first, adding local prerequisites to the scope. An `after` naming a sibling's id or `<epic id>.<entry id>` is that entry's remote item; `epic-<slug>` is the epic's.
- A type mapped to `""` is never sent; its children attach to the nearest published ancestor. Say so once.
- Offer `setup` at first publish, or when the tracker lacks something the maps or `fields` name.
- Then mirror, and edit a published container's `tracker_id` and `remote` into its file.

An unrefined ticket may be published with `refined: false`; refinement updates the same file and remote item.

## Start

Offer the first `ready_to_start` row of `next` in build order, or every unassigned one when work can run in parallel. `find` it with the row's `ref`, refine a `ready_to_refine` one first per `ticket.md`, and confirm it is unassigned; the builder reads the ticket. With the user's approval, start it: on a tracker, publish it if needed, `write` it to `[tickets.status].in-progress` and mirror; on the repo store, write nothing. Then hand it to `bmad-build` or `bmad-build-auto`.

Only an unattended run's orchestrator honors the row's `plan_checkpoint` and `done_checkpoint`.

## Status

The plan is `<type>-<slug>-plan.md` beside where the leaf's file is or would be, linked by its `ticket`; the leaf file's `status` counts only without a plan.

The builds write `draft` through `built`, and `bmad-build-auto` writes `blocked` when it halts; an orchestrator may set `done` through `mark`. Sofia sets `done`, `dropped`, or an assignee only when the user says so (a ticket done or dropped, or a person working it by hand).

- **Repo store.** `tickets.py mark [<folder>] <ref> <status> [--assignee <who>] [--blocked <reason>]`, then commit. `mark` writes only the plan; the checks here are yours.
- **Tracker.** `write` transitions the item to `[tickets.status].<state>`, with `<state>` as `tickets.py status` reports it, never a `status` word, and sets its assignee; then mirror. `mark` refuses a tracker store.
- **Waiting on a person or an answer** on the repo store: `mark` with the current status (`draft` when none) and `--blocked <reason>`; a `mark` without it clears the block.
- **A container's** `status` is Sofia's to edit in its file: absent until its first ticket starts, then `in-progress`, then `done` or `dropped` on the user's confirmation; a published one transitions through `write`.

## Close, drop, move

- {{ workflow.board_closure }}
- Drop only after a `Dropped:` line in the ticket's Notes, or its epic's when it has no file, says why; then set `dropped` per Status, or on a tracker an unpublished entry takes `tracker_status: dropped` through `mirror`. Never delete an entry from an approved `tickets.toml`. With the user, remove or repoint the dropped ticket in each dependent's `after` (`status` lists them under `blocks`). Cancelling a container cancels its descendants once the user confirms.
- Move an entry with `tickets.py move`; it renumbers the entry, moves its file and plan, rewrites every `after` naming it, and reports what it rewrote. It lands last in the destination's `tickets.toml`; put it in build order by hand. Move its `covers` ids between the two epics with the user if the requirement moves. An epic moves as its whole folder with its `[[epic]]` table. Then run `tickets.py status` on the initiative and clear what it reports.
- A done standalone ticket stays in `backlog/` beside its plan.

## Tracker sync

After every `write`, and after the `query` before a board view, run `tickets.py mirror [<folder>]` with a JSON array on stdin, one object per ticket: `ref` (or its `tracker_id`) plus any of `tracker_id`, `remote`, `tracker_status`, `assignee`, `after`.

- `tracker_status` is the state mapped back through `[tickets.status]`: after `write`, the state sent; when two states share a tracker status the query cannot tell apart, the current one if it matches, else the earlier.
- `after` holds tracker ids as strings; an empty string removes a key's line.
- A tracker leaf missing from the tree, even one a known ticket's `after` names, comes back under `unmatched`: add an `[[entry]]` to the parent's `tickets.toml` (the folder's `next_id`, the item's type and title, a one-sentence description), mirror again with that `ref`, and put the item's body in the pulled file.
- Query published containers with their leaves; `mirror` reports them under `containers`, never `unmatched`; tell the user when one's tracker status differs from its file's, and close one only through the closure check.
- A body that differs from the file: show both and ask.
- Move a ticket the tracker shows under another parent per Close, drop, move, then mirror again.
- A failed store call: tell the user what blocks it, confirm the setup, and ask them to authenticate if that is the cause. An empty verb field you cannot infer: ask, and offer to record it per `store-setup.md`.

{{ workflow.board_after }}

# Replanning

Leave the remaining tickets true and nothing already built rewritten. No proposal file.

{{ workflow.replan_before }}

1. **Name the change** briefly: what changed, its source and evidence. Read the ticket, its `.memlog.md`, breakdown, finished children, and any architecture or design it touches; ask only what the tree cannot tell you. Say whether what is required changed or only how it is delivered, and propose a way forward (adjust remaining entries, roll back built work, or cut scope) with the reason; the user decides. An in-progress ticket: ask whether its build stops; never edit its plan.
2. **What is required.** When the change alters what the ticket requires, run `spec.md` on it with the change as input: lines re-render with ids kept, naming entries whose `covers` cite a changed id. Otherwise say so and skip.
3. **Re-slice** each affected epic per Re-slicing in `slice.md`; an obsolete or new epic changes the initiative's breakdown per Initiative into epics there.
4. **Drop** what no longer belongs per Close, drop, move in `board.md`, and clear what waited on it.
5. **Record the reason** as a dated `Decision:` line in the Notes of each container touched, in step 1's words. If the architecture spine or design is invalidated, note it in the container's Notes and offer `bmad-architecture` or `bmad-ux`; never edit the companion.

Then report what changed, was dropped, and is ready.

{{ workflow.replan_after }}

# Slicing

An initiative is sliced into epics; an epic is refined, then planning it writes all its stories. When the user has a split in mind, use it and say where the rules below would cut differently.

{{ workflow.tickets_before }}

## Every container, in order

1. **Envelope.** A document in the folder is an input, not the container; a missing container is written from its input per `spec.md`.
2. **Requirements.** Unwritten spec sections are written per The Operation in `spec.md` from an input; an epic's are written when it is refined.
3. **Complete the container**: known unknowns in Notes, Outcome and Done when re-read against Requirements and Constraints. When the container was just written, confirm its scope with the user before slicing.

## Read the code

On an existing codebase, read enough code to see where deployment splits and which stories would touch the same code.

Where the code contradicts a Requirements line, add a `Source conflict:` line (the template's format) to the Notes of the container that owns the line, tell the user, and offer the correction through `spec.md`.

## One message, then revise

Before the draft is revised, one message holds what the spec and code show, each question that changes the split (first demo, least certain part, the team's epic rule) with your pick, and the draft in build order. Revise it in place until approved.

Offer to save a stated team rule as `slice_to_epics` (customize.toml).

Record each slicing choice, declined suggestions included, as a dated `Decision:` line in the container's Notes as it is made.

## Several initiatives

When the work is more than one initiative's worth, propose a parent initiative over several initiatives, not epics inside epics; every initiative keeps initiative, epic and story. The parent's file holds the Requirements and the decisions its children share. Draft its `tickets/tickets.toml` from `{{ workflow.tickets_template }}`: one `[[initiative]]` per child in build order, with an `id`, the child's folder name as `slug`, a `description` of what is true when it is done, the parent ids it `covers`, and an `after` naming what it needs from which child. Run the tree check in `validate.md`, present the draft, and revise it in place until approved.

On approval, create each child as its own `initiative-<slug>/` beside the parent, its file from `{{ workflow.ticket_template }}` with the parent's file in `sources:`; its Requirements, citing parent ids, are written per `spec.md` when it is taken up. A parent holds no epics or entries. Commit. To work a child, record it as `active_initiative` the way On activation in `workflow.md` does; `tickets.py status` on the parent shows every child.

## Initiative into epics

Revise or drop existing epics per `board.md`; add only when the user confirms the need.

{% if workflow.slice_to_epics %}
Draw boundaries by the team's own rule:

{{ workflow.slice_to_epics }}
{% else %}
- An epic is one source capability, or a tightly coupled pair, delivered and deployed together; two epics need at most a contract between them.
- A module, service, or bounded context is an epic boundary only when it is also a deployment boundary.
{% endif %}

Under either rule:

- One epic unless a distinct outcome or an early-usable part splits it. Offer to split an epic whose stories would run past the count in Epic into stories.
- A unit the work only consumes or configures is a touch point: list it in the initiative's Boundaries with its owning epic.
- The platform baseline (scaffold, CI, deployment) is the opening epic or the first stories of the first epic.
- Every epic delivers to production; its Done when includes integrated verification.

Draft the set as the initiative's `tickets/tickets.toml` from `{{ workflow.tickets_template }}`: one `[[epic]]` per epic in build order, each with an `id`, a `description` of what is true when it is done, the parent ids it `covers`, and an `after` naming what it needs from which epic; `after` on an epic file only for a whole-epic gate. Run the tree check in `validate.md`, then present the draft in build order, naming the rule that cut it, with each epic's boundary, the tracer path, touch points with owners, and decisions several epics share.

On approval, create each new epic's folder in the initiative's `tickets/` folder and its file from `{{ workflow.ticket_template }}`, written thin: Description, Outcome, Boundaries, Done when as far as known, References to the parent, `covers` and `after`, with `refined: false` and no Requirements. Commit (on a tracker, publish) per `board.md`.

Offer `bmad-architecture` to settle shared decisions in the architecture spine, cited in adopting epics' References. If declined, each becomes a `story` entry (`hitl = true`) in the opening epic's `tickets.toml`, written now, and each adopting epic gets `after = [{ epic = <opening>, needs = <the decision> }]`; planning keeps those entries first, ids kept.

Stop here unless the user wants to refine an epic now.

## Refining an epic

An epic with `refined: false` has its scope agreed, not its detail; refine it on request or before planning its stories. Read the epic file and the initiative Requirements lines it covers. In one message, scaled to the initiative (on a small one, a single question: does the epic need anything before stories?):

- What the epic, the initiative and their companions already settle for its stories.
- Each thing its stories would have to agree on that is still open, with your pick: settle it with the user now, plan a spike first when only code can answer it, or leave it to the stories.
- A reminder that the user can flesh the epic out further: more intent, as their own notes or a `bmad-prd` run on the epic; more design, as `bmad-ux` or `bmad-architecture` on the epic; or research with `bmad-deep-recon`.

A UX, architecture or PRD run the user chooses lands in the epic's folder, and SPEC adopts what it produces as a companion. Fill the epic in from the initiative's lines alone only when the user says to, or when it needs nothing more. Then write its Requirements per `spec.md`, re-read Outcome and Done when, set `refined: true`, and commit (on a tracker, publish) per `board.md`.

## Epic into stories

After the epic file, read its breakdown, child tickets, and results. Each blocking unknown is settled now per `board.md`, recorded as the `unknown` of the entries it blocks, or spiked if the user asks.

What makes a good story:

- One step toward the epic's Done when, sized for one agent session; it need not be user-visible. Too small when setup outweighs the work: merge. A one-session epic gets one story.
- {{ workflow.tickets_tracer_bullet }}
- Stories touching the same code form one ordered lane; an interface two lanes share gets a stub as its own early story.
- Done is one runnable check.
- Split, never shrink: "for now", "placeholder", "simplified", "wired later" is a second story.
- Setup and each hitl step go on the first story that needs them, initiative-wide setup in the opening epic. Tests go in every story; the only test story is the closing end-to-end suite.
- {{ workflow.tickets_story_count }}

What opens and closes the breakdown:

- Offer an opening refactor when code quality calls for one.
- {{ workflow.tickets_refactor_sweep }}
- Offer a closing end-to-end suite after the sweep if the source has a test plan or the user asks.

Draft the breakdown as the epic's `tickets/tickets.toml` from `{{ workflow.tickets_template }}`, run the set check in `validate.md`, and revise it in place with the user until approved. Beyond the template:

- `after`: if this epic has an `after` in the initiative's breakdown, add the provider to each entry that needs it: `<epic id>.<entry id>` once that entry exists, else `epic-<slug>`.
- `refine = true` only on request.
- When the user says the epic will build unattended, ask for `plan_checkpoint` and `done_checkpoint` per entry, proposing `plan_checkpoint` on risky ones.
- Each touch point this epic owns is an entry or part of one.

With the draft, name the tracer bullet, parallel work, and deferred scope; record the tracer bullet as a dated `Decision:` line in the epic's Notes.

A split while planning adds an epic (folder, envelope, `[[epic]]` with its `after`) and moves its covers ids and entries; no file is renamed.

Write no leaf file until its entry is pulled.

Run `tickets.py status <initiative folder>`. Repoint each ticket in this epic's `blocks` (under `epics`) to the entry it waits for. Clear any `unpinned_after`, `undeclared_after`, or `order_conflict`, then commit (on a tracker, publish), both per `board.md`.

Tell the user the entries are ready to build; for `bmad-build-auto`, a loop, or a factory, recommend reviewing each entry now (`ticket.md`).

## Re-slicing

When completed work changes the picture, revise unstarted entries and tickets in place with the user, keeping each `id`; run the set check before approval, commit (on a tracker, publish) per `board.md`, and record the reason in the epic's Notes.

{{ workflow.tickets_after }}

# Requirements

Turns any intent input into a container's **derived lines**: the Requirements and Constraints sections and the `Assumption:` and `Open question:` lines, plus its companions.

{{ workflow.spec_before }}

## Workspace

The workspace is the ticket the user or the calling capability names, else the active initiative; a ticket named by id or title resolves as Dispatch in `workflow.md` says. A ticket with no file yet is created first from `{{ workflow.ticket_template }}`; at create SPEC also fills Description, Outcome, Done when (three to six checks at the altitude of the source's ids), Boundaries (the input's non-goals) and References from the input, open questions to Notes; a later slice revises them. A new initiative, named from the input's title, is created and recorded as On activation in `workflow.md` says.

**No input.** Ask for a file path, pasted content, an explanation, or a source.

## Memory and derivation

`.memlog.md` is append-only, one line per entry; requirements carry their stable `R<n>`. Each run derives the derived lines and every spec-authored companion from the memlog plus the sources it cites.

On a container, SPEC is the only writer of the derived lines; a change goes to the memlog, never into them by hand. A derive replaces exactly those lines, sets `sources:` and `companions:` in the frontmatter, adds a References line per companion, and touches nothing else. A source id on a line must exist in a `sources:` file; neither list names the memlog, README files or process records.

Writes go through `{project-root}/_bmad/scripts/memlog.py`:

- `uv run {project-root}/_bmad/scripts/memlog.py init --workspace <ticket folder> --field topic="<the ticket's title>"` — once, at create.
- `uv run {project-root}/_bmad/scripts/memlog.py append --workspace <ticket folder> --type <decision|constraint|requirement|assumption|question|direction|note|event> --text "<one-line gist, reason included>"` — one per line; batch them in one shell call per section.
- Terminal moments (a validation verdict, "spec finalized") are `--type event` entries; the memlog has no status field.

## The Operation

Read the input and what it links. With a `.memlog.md`, this is an update: read the memlog, the authority on decisions and requirement ids, and re-read every file in `sources:`; preserve the ids and give new requirements the next unused `R<n>`. A ticket with Requirements lines and no memlog is also an update: `memlog.py init`, then append each existing derived line as its own entry, ids kept. Otherwise this is a create. An epic whose refinement settled that its parent's lines are its only input (the ids its `covers` names) gets one line per parent id, `R<n> (parent R<m>)`, in the parent's words, with no elicitation, self-validate or memlog; when the parent changes, its lines are re-rendered from the parent, ids kept.

Pre-sorted input (a PRD, a brief from an upstream BMad skill): keep its separation; lift each requirement and constraint into its line with the source's id in parentheses (`R1 (FR-2, FR-5)`), overflow into companions. Mixed input: sort each claim by the test in Load-bearing into a Requirements or Constraints line or a companion.

The template's comments say what Requirements and Constraints hold at this altitude. Rich input: extract directly. Sparse input: name the gaps and walk them with the user one section at a time, Requirements then Constraints; if the user would rather move on, each gap is an `Open question:` line.

An unaddressed domain implication (healthcare silent on PHI) becomes an `Open question:` line; never invent the answer.

A later memlog entry supersedes an earlier one. A new input, or a source changed since the memlog last cited it, is read against the memlog and the other sources; each disagreement, and any either/or never resolved, is put to the user as fix (a new memlog entry, the lines re-derived) or ignore (a `Decision:` line in Notes), never settled silently.

When the user would rather work the idea itself out first, offer `bmad-product-brief` or `bmad-prd`.

## Load-bearing

A claim is **load-bearing** if any consumer (downstream skill, implementing agent, verification pass) would change a decision without it.

## Companions

Load-bearing content that needs more than one line (tables, diagrams, long reference material) goes in a companion; a line growing sub-bullets wants one.

- **Spec-authored** companions sit beside the ticket, named for their content; SPEC may edit them on update.
- **Adopted** companions (upstream artifacts such as `DESIGN.md`, pre-existing project docs) keep their names; SPEC lists them in `companions:` and never edits or duplicates them.

## Spec Law

1. **Each Requirements line says what is possible and what shows it was met.**
2. **Requirements say WHAT, not HOW.** Implementation goes in a companion.
3. **Constraints bend design decisions.**
4. **Boundaries name at least one thing that is out.**
5. **The Outcome's signal is testable or demonstrable.**
6. **Requirement ids are never reused or renumbered.**
7. **Every load-bearing source claim lands in the ticket or a companion.**
8. **Every sentence in the ticket, companions, and `.memlog.md` carries load-bearing content.**

## Self-validate

{{ workflow.spec_self_validate }}

## When lines change

Offer to slice at most once per run, when the input reads as several independently shippable slices. Slicing runs per `slice.md`; load-bearing detail surfaced while slicing comes back here as an update, never into a child alone.

When an update changes or retires an id, name the stale entries and the tickets whose `covers` cite it, and offer to re-slice them per `slice.md`, Re-slicing; never edit a child.

## Output

In one or two sentences, share the ticket's path, requirement count, the files written beside it, and whether the checks passed. List any `Assumption:` or `Open question:` lines, one each, and invite the user to walk through them; answers go to the ticket, and to the source too when the user wants it. When the ticket is a container, offer to slice per `slice.md`, or to re-slice what went stale. On the repo store, save it through `write` per `board.md`; on a tracker it stays in the tree until published.

{{ workflow.spec_after }}

# Setting up the store

`{skill-root}/scripts/read_store.py` layers the project file `{project-root}/_bmad/custom/ticketing-store-config.toml` over the starter `{skill-root}/config/<store>-ticketing.toml`; the project file's keys win. No project file means the repo store.

Run this to connect, reconfigure, or switch a tracker.

{{ workflow.store_before }}

1. **Choose.** `uv run {skill-root}/scripts/read_store.py --starters` lists the shipped stores. The user may bring their own config file or url.
2. **Write the project file.** `[tickets]` with `store` and the starter's empty address fields, asked of the user. Copy nothing else from the starter; cut a whole-starter copy down to `store`, the address fields, and the team's changes. The user's own config goes in whole; a store with no starter is modelled on the nearest one. For the repo store, delete the project file.
3. **Connect.** Offer to run the store's `setup` verb. Tracker types or statuses that differ from the maps go under `[tickets.types]` or `[tickets.status]`. A state with no tracker status takes the previous state's status.
4. **Prove it.** Offer a test. With items, `query` a few back. Empty: create a ticket in `backlog/` titled "BMad setup test — safe to delete", query it back, move it to `in-progress`, assign it, then drop it. Give the user the `remote` url.

A house rule on a verb goes in `persistent_facts` in this skill's customization; set a whole verb under `[verbs]` in the project file only when the store's commands differ.

Then return to the user's task; when done, offer once a `bmad-customize` pass on this skill: its `customize.toml` holds the epic boundary rule, publication timing, own checks, templates, and before and after instructions.

{{ workflow.store_after }}

# Writing a ticket

Ask only what is unsettled, with a default for each. Take a new ticket's `id` from `next_id` in `tickets.py status <folder>`. Open only the template for its type: `{{ workflow.ticket_template }}` (initiative, epic, story), `{{ workflow.spike_template }}` or `{{ workflow.bug_template }}`. Match its example's detail, not its content.

{{ workflow.refine_before }}

## Each fact lives in one place

Reference the Requirements of the ticket above; never copy them. Full criteria ("Reviewing or refining an existing ticket" says when) cover what changes, failure paths and binding decisions; point at the epic for the rest.

## Rules the template cannot carry

- Criteria use the template's format, each provable without pointing at code, together failing any wrong build. Behavior that must not change goes under Boundaries.
- Short plain sentences in the project's vocabulary, no invented terms. Offer the user a rewrite where their text breaks this.
- A reference is a References line, `type — location, section`; location is a path from `{project-root}` or a url, opened before citing; cite the nearest document. Before drafting, read `tickets.reference` from the store config (`read_store.py -k tickets`; the repo store has none).
- No source-code paths or snippets unless the snippet is the decision; a path the user wants kept goes in Notes.
- A UI ticket links its design in References; criteria stay functional. User-facing without a design: offer `bmad-ux`; if declined, warn that the builder will guess the layout unless Notes gives details.
- `hitl: true` only when a person does part of the work; name that step in the Description and list known steps in criteria or Notes.
- Risk on every ticket: `low`, `medium` or `high` business impact if wrong, proposed with a one-line reason; the user decides. A high-risk ticket names in Notes one check beyond its criteria.
- Record an `estimate` only when the user gives one, on the entry or frontmatter; never propose one.
- A bug carries reproduction steps and a cause hypothesis, never a fix; ask for missing steps, tighten unclear ones. Run them when cheap; if the behavior already holds, show the evidence and create nothing. Criteria name the other valid outcome: proof no change is needed.
- A spike names the question, who waits on it, and where the answer goes. A question several epics wait on is a shared decision per `slice.md`.
- Notes holds marked lines local to this ticket; what spans tickets goes in the parent's Notes, referenced. Confirm each `Assumption:` with the user; the answer is a dated decision. An `Open question:` is answered during the work. An `Unknown:` on a leaf file or an entry's `unknown` gates the start: settle it per `board.md`, or as a spike in its dependents' `after` when the user asks. Never guess. With a `.memlog.md`, `Assumption:` and `Open question:` lines go to the memlog first (`spec.md`, Memory and derivation).

## Reviewing or refining an existing ticket

A story with no file (`find` gives `story_file` null) is pulled first: `tickets.py pull <epic folder> <id>` writes a fixed layout, not the template. Leave done or dropped tickets and plans as they are.

A refine sets `refine = true` on the entry and `refined: false` in the file. `tickets.py next` lists under `ready_to_refine` every bug, entry with `refine = true`, and leaf with no entry.

`title`, `type`, `after` and `hitl` change in the file and the entry (the file's `after` and `hitl` win on `drift`); order lives in `tickets.toml`. Once published, changes also go through `write`.

**Review** — the next story, or more on request. Read the file, its epic file, and finished siblings and their plans. With the user, improve the ticket, order and prerequisites; run the Set check in `validate.md` when order or prerequisites change. Set neither `status` nor `refined`.

**Refine** — before the ticket starts. Read what a review reads, plus the code it touches. Where the code contradicts a Requirements line, add a `Source conflict:` line (the template's format) to the Notes of the container that owns the line, tell the user, and offer the correction through `spec.md`; tickets covering that id stay `refined: false` until a `Decision:` line settles it. Confirm prerequisites are done or in review and expand per the type's template and "Rules the template cannot carry".

**A ticket that already carries full criteria, or a container** — with the user, confirm they still agree, check it against its references, tracker state and the ticket above, and settle questions blocking implementation. A container may end in a re-slice per `slice.md`.

Preserve identity, `status`, `tracker_status` and assignee.

## Self-review before the user sees it

Check its level (an entry: description and verification approach; refined: runnable criteria, settled prerequisites), size (one agent session), references and consistency with the ticket above. Fix, then mention fixes that change intent.

## After approval

On the repo store, save through `write`; on a tracker it stays in the tree until published per `board.md`. A ticket with full criteria gets `refined: true` on the user's approval, then the Ticket check in `validate.md` before a build starts it. Then say the file's path and the next step: build it, or start it per `board.md`.

{{ workflow.refine_after }}

# Validation checks

`validate.md` assigns sections to agents.

## Ticket

- References open; no unfilled placeholders; assumptions marked.
- Nothing contradicts its epic file, its companions, the architecture, or a recorded decision.
- Each entry has a description, `unknown` when one is known, and a `verify` a non-builder can run; only bugs, `refine = true` entries, and leaves with no entry need full criteria before execution.
- A ticket needing full criteria and covering an id in a `Source conflict:` line with no later `Decision:` has `refined: false`.

{% if workflow.checks.ticket %}
{{ workflow.checks.ticket }}
{% endif %}

## Set

Every entry in the epic's `tickets.toml`, draft or committed, pulled or not.

- Tickets cover every Requirements line not deferred with the user; each `covers` id exists there; tickets sharing an id say which part each delivers.
- UX, architecture, constraint and integration work is covered without ids; results satisfy Done when; citing an id alone is not coverage.
- Every `after`, cross-epic too, is a real prerequisite, not restated order; once `tickets.toml` exists, `tickets.py status` on the epic shows no cycle, `drift`, `unpinned_after`, `undeclared_after`, or `order_conflict`; the epic's Notes hold a dated `Decision:` line naming the tracer bullet, and the sequencing decisions.
- A builder could write each story's acceptance criteria from the epic, entry, and references; if not, fix the epic, not the one-sentence `description` and `verify`.
- Tracer bullet, as rendered in `slice.md`: {{ workflow.tickets_tracer_bullet }}
- Refactor sweep, as rendered in `slice.md`: {{ workflow.tickets_refactor_sweep }}

{% if workflow.checks.set %}
{{ workflow.checks.set }}
{% endif %}

## Tree

The initiative, its `tickets.toml`, draft or committed, and the epic files once written; for a parent initiative, the child initiatives in place of epics.

- Every initiative requirement has an accountable epic or agreed deferral; shared ones say which part each epic delivers; no unexplained overlap.
- Each epic's `covers` cites parent ids; each Requirements line names in parentheses its parent id, or its source section if the parent lacks it; a child's `covers` resolves through the Requirements line to a parent id.
- Every source id on a Requirements line exists in a `sources:` file, every parent id in the parent's Requirements; every adopted design or architecture document is in `companions:`.
- Every unit the source touches is an epic, out of scope, or a touch point (consumed or configured only, named in the initiative's Boundaries with its owner).
- Each decision several epics adopt has one home: an architecture spine section, or an opening-epic entry (first `[[epic]]`) the others wait on.
- Every written epic file has Outcome, Done when (including production delivery), Boundaries, its `covers`, and references; one with `refined: false` needs no Requirements or breakdown yet; the initiative's `tickets.toml` lists every epic in build order, each `after` naming what it needs, and `tickets.py status` on the initiative shows no cycle or `order_conflict`; it opens with a platform-baseline epic, or its first epic's first stories, or a `Decision:` line saying why not.
- A parent initiative's `tickets.toml` lists every child in build order, each `after` naming what it needs; every child's `sources:` lists the parent's file; `tickets.py status` on the parent shows no `order_conflict`.
- The initiative's `tickets.toml` lists child initiatives, epics, or entries, one kind only; with epics, no leaf sits in the initiative's `tickets/` folder.

{% if workflow.checks.tree %}
{{ workflow.checks.tree }}
{% endif %}

## Dependencies

An item is a set entry, a tree epic, or a parent's child initiative.

- Needs: what each item needs to start and to run its check is built by it or its `after`.
- Collisions: items with no dependency path share no code, config, schema, or setup; else add an `after` or merge them.
- Shared setup: the earliest item needing it owns it; others list it in `after`.
- Handoffs: when one item relies on another's output, both descriptions say so.

{% if workflow.checks.dependencies %}
{{ workflow.checks.dependencies }}
{% endif %}

## Closure

- Verify the container's Done when, requirements, constraints and companions against implementation evidence, not child status; explain dropped work and confirm remaining scope with the user.

{% if workflow.checks.closure %}
{{ workflow.checks.closure }}
{% endif %}

# Validating tickets

The checks are in `{{ rendered("references/validate-checks.md") }}`, one section per scope. Fresh-context subagents run them, given the file's path and their section names. They find missing prerequisites `tickets.py` cannot see.

{{ workflow.validate_before }}

- **Proposed child initiatives or epics, an epic breakdown, or a re-slice:** always, on the draft `tickets.toml` before approval. First run the Dependencies section yourself and fix the draft; give the agent its path. Show what the check changed and confirmed. Rerun when the user adds, removes, merges, or reorders items. At initiative slicing, check scope ownership; future epics need no stories or full detail.
- **A refined ticket:** before execution.
- **A container:** before closure, its implemented coverage and Done when.

How:

- Per epic: its container, its `tickets.toml`, every child ticket, done ones too, its Requirements, its `sources:` files, companions; sections Ticket, Set, Dependencies. The tree: the initiative, its `tickets.toml`, the epic files once written (on a parent, each child initiative's file), its `sources:` files; sections Tree, Dependencies. A single ticket: its epic file; section Ticket. Closure: section Closure.
- Name the `refined: true` tickets; point the agent at "Rules the template cannot carry" in `{{ rendered("references/ticket.md") }}`. Once `tickets.toml` exists, give the full `tickets.py status` command for the folder.
- Sort findings into fix (mechanical), suggest (with reason), or ask the user. Resolve gaps, missing prerequisites, and contradictions before proceeding; the user decides suggestions and scope changes.
- Do not re-raise suggestions declined in a `Decision:` line unless new evidence changes their basis.

{{ workflow.validate_after }}

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# ///
"""Read the ticket store's config without loading the rest into context.

The config is two layers. The base is the skill's starter for the store, `config/<store>-ticketing.toml`.
Over it goes the project's `_bmad/custom/ticketing-store-config.toml`: its `[tickets] store` names the
store, and any other key it sets replaces the starter's, tables merging key by key. With no project file
the store is `repo`. A store with no starter is read from the project file alone.

  read_store.py --project-root <root> [-k <dotted key> ...]
  read_store.py --starters

One key prints its value, a string bare and anything else as JSON; several print one JSON object; none
prints the whole merged config. `--starters` prints each shipped store with its description.
Exit 1 when a file cannot be read, 2 when a key is missing.
"""

import argparse
import json
import re
import sys
import tomllib
from pathlib import Path

sys.dont_write_bytecode = True

_MISSING = object()
STARTERS = Path(__file__).resolve().parents[1] / "config"
PROJECT_FILE = Path("_bmad") / "custom" / "ticketing-store-config.toml"
STORE_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")


def load(path: Path) -> dict:
    # utf-8-sig: Windows editors can save a byte-order mark.
    return tomllib.loads(path.read_text(encoding="utf-8-sig"))


def merge(base: dict, over: dict) -> dict:
    out = dict(base)
    for key, value in over.items():
        out[key] = merge(out[key], value) if isinstance(value, dict) and isinstance(out.get(key), dict) else value
    return out


def store_config(project_root: Path, starters: Path) -> dict:
    path = project_root / PROJECT_FILE
    project = load(path) if path.is_file() else {}
    tickets = project.get("tickets")
    store = tickets.get("store") if isinstance(tickets, dict) else None
    if not isinstance(store, str) or not store:
        store = "repo"  # as tickets.py reads it
    starter = starters / f"{store}-ticketing.toml"
    config = merge(load(starter), project) if STORE_RE.match(store) and starter.is_file() else project
    if isinstance(config.get("tickets"), dict):
        config["tickets"]["store"] = store
    return config


def extract(data, dotted: str):
    current = data
    for part in dotted.split("."):
        if isinstance(current, dict) and part in current:
            current = current[part]
        else:
            return _MISSING
    return current


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--project-root", help="project holding _bmad/")
    parser.add_argument("--key", "-k", action="append", default=[], help="dotted key (repeatable)")
    parser.add_argument("--starters", action="store_true", help="list the shipped stores and their descriptions")
    parser.add_argument(
        "--starters-dir", default=str(STARTERS), help="folder of starters; default: the skill's config/"
    )
    args = parser.parse_args()
    starters = Path(args.starters_dir).expanduser()

    try:
        if args.starters:
            found = {}
            for path in sorted(starters.glob("*-ticketing.toml")):
                data = load(path)
                found[data.get("tickets", {}).get("store", path.name)] = data.get("description", "")
            print(json.dumps(found, indent=2, ensure_ascii=False))
            return 0
        if not args.project_root:
            parser.error("--project-root is required")
        data = store_config(Path(args.project_root).expanduser(), starters)
    except (OSError, ValueError) as error:  # ValueError covers a TOML or encoding failure
        sys.stderr.write(f"error: cannot read the store config: {error}\n")
        return 1

    if not args.key:
        print(json.dumps(data, indent=2, ensure_ascii=False))
        return 0

    found = {}
    missing = []
    for key in args.key:
        value = extract(data, key)
        (missing.append(key) if value is _MISSING else found.__setitem__(key, value))
    for key in missing:
        sys.stderr.write(f"missing: {key}\n")

    if len(args.key) == 1:
        if missing:
            return 2
        value = found[args.key[0]]
        print(value.rstrip("\n") if isinstance(value, str) else json.dumps(value, indent=2, ensure_ascii=False))
    else:
        print(json.dumps(found, indent=2, ensure_ascii=False))
    return 2 if missing else 0


if __name__ == "__main__":
    if sys.platform == "win32":
        # Piped output on Windows defaults to a legacy code page, not UTF-8.
        sys.stdout.reconfigure(encoding="utf-8")
        sys.stderr.reconfigure(encoding="utf-8")
    raise SystemExit(main())

import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path

SCRIPT = Path(__file__).resolve().parents[1] / "read_store.py"

REPO = '''
# comment that must never be printed
description = "Files in the repo."

[tickets]
store = "repo"
access = "Read and write the files."

[verbs]
write = """
Edit the file.
Commit it.
"""

[tickets.status]
backlog = "backlog"
done = "done"
'''

JIRA = """
description = "Jira through acli."

[tickets]
store = "jira"
key = ""
site = ""

[verbs]
write = "acli create"
query = "acli view"

[tickets.status]
backlog = "To Do"
review = "In Review"
done = "Done"
"""


class ReadStoreTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name) / "project"
        self.starters = Path(self.tmp.name) / "config"
        (self.root / "_bmad" / "custom").mkdir(parents=True)
        self.starters.mkdir()
        (self.starters / "repo-ticketing.toml").write_text(REPO, encoding="utf-8")
        (self.starters / "jira-ticketing.toml").write_text(JIRA, encoding="utf-8")

    def tearDown(self):
        self.tmp.cleanup()

    def project(self, text):
        (self.root / "_bmad" / "custom" / "ticketing-store-config.toml").write_text(text, encoding="utf-8")

    def run_script(self, *args):
        return subprocess.run(
            [sys.executable, str(SCRIPT), "--starters-dir", str(self.starters), *args],
            encoding="utf-8",
            capture_output=True,
            check=False,
        )

    def read(self, *keys):
        args = [a for k in keys for a in ("-k", k)]
        return self.run_script("--project-root", str(self.root), *args)

    def test_no_project_file_reads_the_repo_starter(self):
        result = self.read("tickets.store")
        self.assertEqual((result.returncode, result.stdout), (0, "repo\n"))
        result = self.read("verbs.write")
        self.assertEqual(result.stdout, "Edit the file.\nCommit it.\n")
        self.assertNotIn("comment", self.read().stdout)

    def test_the_project_file_names_the_store_and_its_keys_win_one_by_one(self):
        self.project('[tickets]\nstore = "jira"\nkey = "SHOP"\n\n[tickets.status]\nreview = ""\n')
        result = self.read("tickets", "verbs.query")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(
            json.loads(result.stdout),
            {
                "tickets": {
                    "store": "jira",
                    "key": "SHOP",
                    "site": "",
                    "status": {"backlog": "To Do", "review": "", "done": "Done"},
                },
                "verbs.query": "acli view",
            },
        )

    def test_a_verb_the_project_file_sets_replaces_the_starters(self):
        self.project('[tickets]\nstore = "jira"\n\n[verbs]\nwrite = "our own create"\n')
        self.assertEqual(self.read("verbs.write").stdout, "our own create\n")
        self.assertEqual(self.read("verbs.query").stdout, "acli view\n")

    def test_a_store_with_no_starter_is_read_from_the_project_file_alone(self):
        self.project('[tickets]\nstore = "azure"\n\n[verbs]\nwrite = "az boards"\n')
        self.assertEqual(self.read("verbs.write").stdout, "az boards\n")
        result = self.read("verbs.query")
        self.assertEqual(result.returncode, 2)
        self.assertIn("missing: verbs.query", result.stderr)

    def test_a_store_name_that_is_no_file_name_reads_no_starter(self):
        self.project('[tickets]\nstore = "../repo"\n')
        self.assertEqual(self.read("verbs.write").returncode, 2)

    def test_a_project_file_that_names_no_store_reads_the_repo_starter(self):
        for text in ("[workflow]\nx = 1\n", "[tickets]\nstore = 5\n", '[tickets]\nstore = ""\nkey = "K"\n'):
            self.project(text)
            self.assertEqual(self.read("verbs.write").stdout, "Edit the file.\nCommit it.\n", text)
            self.assertEqual(self.read("tickets.store").stdout, "repo\n", text)
        self.assertEqual(self.read("tickets.key").stdout, "K\n")

    def test_several_keys_with_one_missing_print_the_rest_and_exit_2(self):
        result = self.read("tickets.store", "tickets.nope")
        self.assertEqual(result.returncode, 2)
        self.assertEqual(json.loads(result.stdout), {"tickets.store": "repo"})
        self.assertIn("missing: tickets.nope", result.stderr)

    def test_an_unreadable_project_file_exits_1(self):
        self.project("[tickets\nstore = ")
        result = self.read("tickets.store")
        self.assertEqual(result.returncode, 1)
        self.assertIn("cannot read the store config", result.stderr)

    def test_starters_lists_each_store_with_its_description(self):
        result = self.run_script("--starters")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(json.loads(result.stdout), {"jira": "Jira through acli.", "repo": "Files in the repo."})

    def test_every_shipped_starter_parses_and_is_named_for_its_store(self):
        shipped = SCRIPT.parents[1] / "config"
        result = subprocess.run(
            [sys.executable, str(SCRIPT), "--starters"], encoding="utf-8", capture_output=True, check=False
        )
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(
            sorted(f"{store}-ticketing.toml" for store in json.loads(result.stdout)),
            sorted(p.name for p in shipped.glob("*-ticketing.toml")),
        )


if __name__ == "__main__":
    unittest.main()

import json
import shutil
import subprocess
import sys
import tempfile
import tomllib
import unittest
from datetime import date
from pathlib import Path

SCRIPT = Path(__file__).resolve().parents[1] / "tickets.py"
CONFIG_UTILS = SCRIPT.parents[2] / "bmad" / "scripts" / "config_utils.py"


def ticket(
    status,
    ticket_id=None,
    after="[]",
    hitl="false",
    tracker_id='""',
    assignee='""',
    blocked_at=None,
    blocked_reason=None,
    kind="story",
    refined="false",
    tracker_status=None,
):
    lines = ["---"]
    if ticket_id is not None:
        lines.append(f"id: {ticket_id}")
    lines += [
        f"tracker_id: {tracker_id}",
        'remote: ""',
        f"type: {kind}",
        'title: "x"',
        "parent: epic-cart",
        "covers: [R1]",
        f"after: {after}",
        f"assignee: {assignee}",
    ]
    if status:
        lines.append(f"status: {status}")
    if tracker_status:
        lines.append(f"tracker_status: {tracker_status}")
    lines += [
        f"refined: {refined}",
        f"hitl: {hitl}",
        "risk: low",
    ]
    if blocked_at:
        lines.append(f'blocked_at: "{blocked_at}"')
    if blocked_reason:
        lines.append(f'blocked_reason: "{blocked_reason}"')
    lines += ["---", "", "# x", ""]
    return "\n".join(lines)


def plan(ticket_id, status=None, assignee=None, blocked_at=None, blocked_reason=None):
    """A plan as the build's template writes it: quoted values, trailing comments, block lists."""
    lines = [
        "---",
        "title: 'x'",
        "type: 'feature' # feature | bugfix | refactor | chore",
        f"ticket: '{ticket_id}' # the entry id from the ticket tree, or the story file's stem when the entry has no id; empty outside it",
    ]
    if status:
        lines.append(f"status: '{status}' # draft | ready-for-dev | in-progress | in-review | built | done")
    for key, value in (("assignee", assignee), ("blocked_at", blocked_at), ("blocked_reason", blocked_reason)):
        if value:
            lines.append(f"{key}: '{value}'")
    lines += ["lenses_ran: []", "deferred:", "  - summary: a", "    evidence: b", "---", "", "# x", ""]
    return "\n".join(lines)


def run(*args, cwd=None, stdin=None):
    # tickets.py writes UTF-8; text=True alone would decode with the locale's code page on Windows.
    return subprocess.run(
        [sys.executable, str(SCRIPT), *args], encoding="utf-8", capture_output=True, check=False, cwd=cwd, input=stdin
    )


def tk(folder):
    """A folder's tickets folder, made when missing; a backlog, with no file of its own, is its own."""
    if not (folder / f"{folder.name}.md").is_file():
        return folder
    (folder / "tickets").mkdir(exist_ok=True)
    return folder / "tickets"


class TreeCase(unittest.TestCase):
    """A temp project with `_bmad/custom/` and the epic `out/initiative-checkout/tickets/epic-cart`."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)
        (self.root / "_bmad" / "custom").mkdir(parents=True)
        self.initiative = self.root / "out" / "initiative-checkout"
        self.initiative.mkdir(parents=True)
        (self.initiative / "initiative-checkout.md").write_text("---\ntype: initiative\n---\n# Checkout\n")
        self.epic = self.add_epic("epic-cart")
        self.write_store("repo")

    def tearDown(self):
        self.tmp.cleanup()

    def write_store(self, name="repo", root=None):
        line = f'root = "{root}"\n' if root is not None else ""
        (self.root / "_bmad" / "custom" / "ticketing-store-config.toml").write_text(
            f'[tickets]\nstore = "{name}"\n{line}'
        )

    def add_epic(self, slug, status="", after="[]"):
        folder = self.initiative / "tickets" / slug
        folder.mkdir(parents=True)
        line = f"status: {status}\n" if status else ""
        (folder / f"{slug}.md").write_text(f"---\ntype: epic\n{line}after: {after}\n---\n# {slug}\n")
        return folder

    def add_backlog(self):
        backlog = self.root / "out" / "backlog"
        backlog.mkdir()
        return backlog

    def status_rows(self, folder):
        return json.loads(run("status", str(folder)).stdout)["tickets"]

    def add(self, name, text, folder=None):
        (tk(folder or self.epic) / name).write_text(text, encoding="utf-8")

    def seed(self, s1="", s2="", s3=""):
        self.add("story-scaffold.md", ticket(s1, 1, hitl="true"))
        self.add("story-ui-shell.md", ticket(s2, 2))
        self.add("story-tracer.md", ticket(s3, 3, after="[1, 2]"))
        self.add("story-codes.md", ticket("", 4, after="[story-tracer]"))
        self.add("spike-tax.md", ticket("", 5, after="[3]", kind="spike"))

    def pricing(self):
        pricing = self.add_epic("epic-pricing")
        (tk(pricing) / "tickets.toml").write_text(
            '[[entry]]\nid = 1\ntype = "story"\ntitle = "Pricing contract"\n\n'
            '[[entry]]\nid = 2\ntype = "story"\ntitle = "Pricing rules"\nafter = [1]\n'
        )
        (tk(self.initiative) / "tickets.toml").write_text(
            '[[epic]]\nid = 1\nslug = "epic-pricing"\n\n[[epic]]\nid = 2\nslug = "epic-cart"\n'
        )
        return pricing

    SHOP = (
        '[[initiative]]\nid = 1\nslug = "initiative-checkout"\ntitle = "Checkout"\n\n'
        '[[initiative]]\nid = 2\nslug = "initiative-payments"\ntitle = "Payments"\n'
        'after = [{ initiative = 1, needs = "the cart total" }]\n\n'
        '[[initiative]]\nid = 3\nslug = "initiative-refunds"\ntitle = "Refunds"\n'
        'after = [{ initiative = "initiative-payments" }, { initiative = 1 }]\n'
    )

    def shop(self, text=None):
        """The parent initiative `out/initiative-shop`; of its children only initiative-checkout has a folder."""
        parent = self.root / "out" / "initiative-shop"
        parent.mkdir(exist_ok=True)
        (parent / "initiative-shop.md").write_text("---\ntype: initiative\n---\n# Shop\n")
        (tk(parent) / "tickets.toml").write_text(text or self.SHOP)
        return parent


class TicketsTests(TreeCase):
    def next(self, *extra):
        r = run("next", str(self.epic), *extra)
        self.assertEqual(r.returncode, 0, r.stderr)
        return json.loads(r.stdout)

    def files(self, rows):
        return [r["file"] for r in rows]

    def test_unstarted_tickets_with_no_prerequisites_are_ready_to_refine(self):
        self.seed()
        out = self.next()
        self.assertEqual(self.files(out["ready_to_refine"]), ["story-scaffold.md", "story-ui-shell.md"])
        self.assertEqual(out["ready_to_start"], [])
        self.assertEqual(self.files(out["blocked"]), ["story-tracer.md", "story-codes.md", "spike-tax.md"])
        self.assertTrue(out["ready_to_refine"][0]["hitl"])

    def test_unrefined_ticket_is_never_ready_to_start(self):
        self.seed(s1="done", s2="draft")
        out = self.next()
        self.assertEqual(out["ready_to_start"], [])
        self.assertIn("story-ui-shell.md", self.files(out["ready_to_refine"]))

    def test_refined_is_ready_to_start_and_ready_set_moves_when_prerequisites_done(self):
        self.seed(s1="done")
        self.add("story-ui-shell.md", ticket("ready-for-dev", 2, refined="true"))
        out = self.next()
        self.assertEqual(self.files(out["ready_to_start"]), ["story-ui-shell.md"])
        self.assertIn("story-tracer.md", self.files(out["blocked"]))
        self.seed(s1="done", s2="done")
        out = self.next()
        self.assertEqual(self.files(out["ready_to_refine"]), ["story-tracer.md"])

    def test_prerequisites_resolve_by_id_stem_and_tracker_id(self):
        self.seed(s1="done", s2="done", s3="done")
        self.add("story-by-id.md", ticket("", 6, after="[CART-4]"))
        self.add("story-codes.md", ticket("done", 4, after="[story-tracer]", tracker_id='"CART-4"'))
        out = self.next()
        self.assertIn("story-by-id.md", self.files(out["ready_to_refine"]))
        self.assertIn("spike-tax.md", self.files(out["ready_to_refine"]))

    def test_in_progress_review_blocked_build_and_blocked_at(self):
        self.seed(s1="in-progress", s2="in-review")
        self.add("story-ui-shell.md", ticket("", 2, blocked_at="2026-09-05"))
        self.add("story-codes.md", ticket("blocked", 4))
        out = self.next()
        self.assertEqual(self.files(out["in_progress"]), ["story-scaffold.md"])
        self.assertEqual(
            self.files(out["blocked"]), ["story-ui-shell.md", "story-tracer.md", "story-codes.md", "spike-tax.md"]
        )
        self.assertEqual([r["state"] for r in out["blocked"]], ["backlog", "backlog", "in-progress", "backlog"])

    def test_tracker_status_wins_over_the_build_status_for_state(self):
        self.seed(s1="in-progress", s2="ready-for-dev")
        self.add("story-scaffold.md", ticket("in-progress", 1, tracker_status="done"))
        self.add("story-ui-shell.md", ticket("", 2, tracker_status="in-progress"))
        out = self.next()
        self.assertEqual(self.files(out["in_progress"]), ["story-ui-shell.md"])
        self.assertEqual(out["in_progress"][0]["status"], "")
        self.assertEqual(self.files(out["blocked"]), ["story-tracer.md", "story-codes.md", "spike-tax.md"])
        self.add("story-ui-shell.md", ticket("", 2, tracker_status="done"))
        self.assertEqual(self.files(self.next()["ready_to_refine"]), ["story-tracer.md"])
        self.add("story-ui-shell.md", ticket("", 2, tracker_status="todo"))
        self.assertIn("tracker_status", run("next", str(self.epic)).stderr)

    def test_status_counts_order_and_chain(self):
        self.seed(s1="done")
        r = run("status", str(self.epic))
        self.assertEqual(r.returncode, 0, r.stderr)
        out = json.loads(r.stdout)
        self.assertEqual([t["id"] for t in out["tickets"]], [1, 2, 3, 4, 5])
        self.assertEqual(out["counts"], {"total": 5, "done": 1, "backlog": 4})
        self.assertEqual(out["longest_remaining_chain"], ["epic-cart/2", "epic-cart/3", "epic-cart/4"])
        self.assertEqual(out["tickets"][2]["blocks"], [4, 5])

    BREAKDOWN = """
[[entry]]
id = 1
type = "story"
title = "Scaffold"
covers = ["R1"]

[[entry]]
id = 2
type = "story"
title = "UI shell"
after = [1]
covers = ["R1"]

[[entry]]
id = 3
type = "spike"
title = "Tax engine?"
after = [1]
covers = ["R4"]
hitl = true

[[entry]]
id = 4
type = "story"
title = "Codes"
after = [2, 3]
covers = ["R2", "R3"]
"""

    def breakdown_epic(self, text=None):
        (tk(self.epic) / "tickets.toml").write_text(text or self.BREAKDOWN, encoding="utf-8")

    def test_text_outside_ascii_is_read_and_written_as_utf8(self):
        self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Café ✓ menu"'))
        r = subprocess.run([sys.executable, str(SCRIPT), "next", str(self.epic)], capture_output=True, check=False)
        self.assertEqual(r.returncode, 0, r.stderr)
        out = json.loads(r.stdout.decode("utf-8"))
        self.assertEqual(out["ready_to_start"][0]["title"], "Café ✓ menu")
        r = subprocess.run([sys.executable, str(SCRIPT), "pull", str(self.epic), "1"], capture_output=True, check=False)
        self.assertEqual(r.returncode, 0, r.stderr)
        self.assertIn("# Café ✓ menu", (tk(self.epic) / "story-cafe-menu.md").read_text(encoding="utf-8"))

    def test_titles_that_slug_alike_get_their_own_file_and_plan(self):
        self.breakdown_epic(
            '[[entry]]\nid = 1\ntype = "story"\ntitle = "Café menu"\n\n'
            '[[entry]]\nid = 2\ntype = "story"\ntitle = "Cafe menu"\n'
        )
        plans = [Path(json.loads(run("find", str(self.epic), n).stdout)["plan"]).name for n in ("1", "2")]
        self.assertEqual(plans, ["story-cafe-menu-plan.md", "story-cafe-menu-2-plan.md"])
        self.assertEqual(json.loads(run("pull", str(self.epic), "2").stdout)["file"], "story-cafe-menu-2.md")
        self.assertEqual(json.loads(run("pull", str(self.epic), "1").stdout)["file"], "story-cafe-menu.md")
        plans = [Path(json.loads(run("find", str(self.epic), n).stdout)["plan"]).name for n in ("1", "2")]
        self.assertEqual(plans, ["story-cafe-menu-plan.md", "story-cafe-menu-2-plan.md"])

    def test_a_leaf_file_with_a_byte_order_mark_is_read(self):
        (tk(self.epic) / "story-scaffold.md").write_bytes(b"\xef\xbb\xbf" + ticket("done", 1).encode("utf-8"))
        rows = self.status_rows(self.epic)
        self.assertEqual([(r["file"], r["state"]) for r in rows], [("story-scaffold.md", "done")])

    def test_mark_keeps_a_plans_crlf_line_endings_and_byte_order_mark(self):
        self.breakdown_epic()
        path = tk(self.epic) / "story-scaffold-plan.md"
        text = plan(1, "in-progress").replace("\n", "\r\n")
        path.write_bytes(b"\xef\xbb\xbf" + text.encode("utf-8"))
        self.mark(str(self.epic), "1", "done")
        data = path.read_bytes()
        self.assertTrue(data.startswith(b"\xef\xbb\xbf"))
        self.assertNotIn(b"\n", data.replace(b"\r\n", b""))
        self.assertIn(b"\r\nstatus: done\r\n", data)
        self.assertEqual(self.status_rows(self.epic)[0]["status"], "done")

    def test_planned_entries_surface_when_unblocked(self):
        self.breakdown_epic()
        self.add("story-scaffold.md", ticket("done", 1))
        out = self.next()
        self.assertEqual(
            [(e["id"], e["type"], e["title"]) for e in out["ready_to_start"]],
            [(2, "story", "UI shell"), (3, "spike", "Tax engine?")],
        )
        self.assertEqual(out["ready_to_start"][1]["covers"], ["R4"])
        self.assertEqual(out["ready_to_start"][1]["after"], [1])
        self.assertTrue(out["ready_to_start"][1]["hitl"])
        self.assertEqual([e["id"] for e in out["blocked"]], [4])
        self.add("story-ui-shell.md", ticket("", 2, after="[1]"))
        out = self.next()
        self.assertEqual(self.files(out["ready_to_start"]), ["story-ui-shell.md", None])
        status = json.loads(run("status", str(self.epic)).stdout)
        self.assertEqual(status["counts"], {"total": 4, "done": 1, "backlog": 1, "planned": 2})
        self.assertEqual(status["tickets"][0]["blocks"], [2, 3])

    def test_table_order_is_build_order_not_id(self):
        self.breakdown_epic(
            '[[entry]]\nid = 2\ntype = "story"\ntitle = "Second first"\n\n'
            '[[entry]]\nid = 1\ntype = "story"\ntitle = "First second"\n\n'
            '[[entry]]\nid = 3\ntype = "story"\ntitle = "Third"\nafter = [1, 2]\n'
        )
        self.assertEqual([e["id"] for e in self.next()["ready_to_start"]], [2, 1])
        self.add("story-first-second.md", ticket("draft", 1))
        rows = self.status_rows(self.epic)
        self.assertEqual([r["id"] for r in rows], [2, 1, 3])

    def test_entry_waiting_on_unwritten_entry_stays_blocked(self):
        self.breakdown_epic()
        out = self.next()
        self.assertEqual([e["id"] for e in out["ready_to_start"]], [1])
        self.assertEqual([e["id"] for e in out["blocked"]], [2, 3, 4])

    def test_a_prerequisite_in_review_is_met_and_one_in_progress_is_not(self):
        self.breakdown_epic()
        self.add("story-scaffold-plan.md", plan(1, "in-review"))
        out = self.next()
        self.assertEqual([e["id"] for e in out["ready_to_start"]], [2, 3])
        self.assertEqual([e["id"] for e in out["in_progress"]], [1])
        self.add("story-scaffold-plan.md", plan(1, "in-progress"))
        out = self.next()
        self.assertEqual(out["ready_to_start"], [])
        self.assertEqual([e["id"] for e in out["blocked"]], [2, 3, 4])

    def test_an_epic_gate_waits_for_the_epic_to_be_done_even_with_its_tickets_in_review(self):
        pricing = self.pricing()
        self.breakdown_epic('[[entry]]\nid = 1\ntype = "story"\ntitle = "Cart"\nafter = ["1.2"]\n')
        self.add("story-pricing-contract-plan.md", plan(1, "in-review"), pricing)
        self.add("story-pricing-rules-plan.md", plan(2, "in-review"), pricing)
        self.assertEqual([e["id"] for e in self.next()["ready_to_start"]], [1])
        (self.epic / "epic-cart.md").write_text("---\ntype: epic\nafter: [epic-pricing]\n---\n# epic-cart\n")
        out = self.next()
        self.assertEqual(out["ready_to_start"], [])
        self.assertEqual(out["blocked"][0]["gated_by"], ["epic-pricing"])
        (pricing / "epic-pricing.md").write_text("---\ntype: epic\nstatus: done\n---\n# epic-pricing\n")
        self.assertEqual([e["id"] for e in self.next()["ready_to_start"]], [1])

    def test_every_row_carries_a_ref_that_find_resolves_to_the_same_row(self):
        self.pricing()
        self.breakdown_epic()
        backlog = self.add_backlog()
        self.add("bug-x.md", ticket("draft", kind="bug"), backlog)
        cases = (
            (self.initiative, ["1.1", "1.2", "2.1", "2.2", "2.3", "2.4"]),
            (self.epic, ["2.1", "2.2", "2.3", "2.4"]),
            (backlog, ["bug-x.md"]),
        )
        for folder, refs in cases:
            rows = self.status_rows(folder)
            self.assertEqual([r["ref"] for r in rows], refs, folder.name)
            for row in rows:
                r = run("find", str(folder), row["ref"])
                self.assertEqual(r.returncode, 0, r.stderr)
                found = json.loads(r.stdout)
                self.assertEqual(
                    [found[k] for k in ("epic", "id", "file", "ref")], [row[k] for k in ("epic", "id", "file", "ref")]
                )
            for group in ("ready_to_start", "blocked"):
                for row in json.loads(run("next", str(folder)).stdout)[group]:
                    self.assertIn(row["ref"], refs)

    def test_a_ref_is_the_bare_id_in_an_epic_the_initiative_does_not_number(self):
        self.breakdown_epic()
        self.assertEqual([r["ref"] for r in self.next()["ready_to_start"]], ["1"])
        self.assertEqual(json.loads(run("find", str(self.epic), "1").stdout)["title"], "Scaffold")

    def test_file_prerequisites_win_over_the_entry_and_status_flags_the_drift(self):
        self.breakdown_epic()
        self.add("story-scaffold.md", ticket("draft", 1))
        self.add("story-ui-shell.md", ticket("draft", 2, after="[]"))
        self.assertIn("story-ui-shell.md", self.files(self.next()["ready_to_start"]))
        rows = self.status_rows(self.epic)
        self.assertTrue(rows[1]["drift"])
        self.assertNotIn("drift", rows[0])

    def test_a_leaf_file_without_after_or_hitl_reads_them_as_empty_and_flags_the_drift(self):
        self.breakdown_epic()
        self.add("story-ui-shell.md", '---\nid: 2\ntype: story\ntitle: "UI shell"\n---\n# UI shell\n')
        self.add("spike-tax-engine.md", '---\nid: 3\ntype: spike\ntitle: "Tax engine?"\nafter: [1]\n---\n# Tax\n')
        rows = {r["id"]: r for r in self.status_rows(self.epic)}
        self.assertEqual((rows[2]["after"], rows[2]["drift"]), ([], {"after": {"file": [], "entry": [1]}}))
        self.assertEqual((rows[3]["hitl"], rows[3]["drift"]), (False, {"hitl": {"file": False, "entry": True}}))
        self.assertNotIn("drift", rows[1])

    def test_a_blocked_row_names_the_prerequisites_it_waits_on(self):
        self.pricing()
        self.breakdown_epic(self.BREAKDOWN.replace("after = [2, 3]", 'after = [2, 3, "1.2"]'))
        self.add("story-scaffold-plan.md", plan(1, "in-progress"))
        self.add("spike-tax-engine-plan.md", plan(3, "in-review"))
        self.add("story-ui-shell-plan.md", plan(2, "blocked", blocked_reason="legal"))
        out = self.next()
        blocked = {e["id"]: e for e in out["blocked"]}
        self.assertEqual(blocked[4]["waiting_on"], [2, "1.2"])
        self.assertEqual(blocked[2]["waiting_on"], [1])
        # 3 is in review with 1 still open: a ticket already under way is not waiting.
        self.assertEqual([e["id"] for e in out["in_progress"]], [1, 3])
        self.assertFalse(any("waiting_on" in e for e in out["in_progress"]))
        self.add("story-scaffold-plan.md", plan(1, "done"))
        self.assertNotIn("waiting_on", {e["id"]: e for e in self.next()["blocked"]}[2])

    def test_status_gives_the_id_a_new_ticket_takes(self):
        self.pricing()
        self.breakdown_epic()
        self.assertEqual(json.loads(run("status", str(self.epic)).stdout)["next_id"], 5)
        self.breakdown_epic(self.BREAKDOWN + '\n[[entry]]\nid = "6a"\ntype = "story"\ntitle = "Lettered"\n')
        self.assertEqual(json.loads(run("status", str(self.epic)).stdout)["next_id"], 7)
        self.add("story-unlisted.md", ticket("", 9))
        out = json.loads(run("status", str(self.epic)).stdout)
        self.assertEqual(out["next_id"], 10)
        out = json.loads(run("status", str(self.initiative)).stdout)
        self.assertNotIn("next_id", out)
        self.assertEqual({e["slug"]: e["next_id"] for e in out["epics"]}, {"epic-pricing": 3, "epic-cart": 10})
        backlog = self.add_backlog()
        self.assertEqual(json.loads(run("status", str(backlog)).stdout)["next_id"], 1)
        self.add("bug-x.md", ticket("draft", 7, kind="bug"), backlog)
        self.assertEqual(json.loads(run("status", str(backlog)).stdout)["next_id"], 8)

    def test_a_folders_own_next_id_keeps_a_moved_entrys_id_from_coming_back(self):
        self.breakdown_epic("next_id = 9\n" + self.BREAKDOWN)
        self.assertEqual(json.loads(run("status", str(self.epic)).stdout)["next_id"], 9)
        self.breakdown_epic("next_id = 3\n" + self.BREAKDOWN)
        self.assertEqual(json.loads(run("status", str(self.epic)).stdout)["next_id"], 5)
        self.breakdown_epic('next_id = "9"\n' + self.BREAKDOWN)
        r = run("status", str(self.epic))
        self.assertEqual(r.returncode, 1)
        self.assertIn("`next_id` must be a whole number", r.stderr)

    def test_next_and_find_carry_the_drift_too(self):
        self.breakdown_epic()
        self.add("story-scaffold.md", ticket("draft", 1))
        self.add("story-ui-shell.md", ticket("draft", 2, after="[]"))
        drift = {"after": {"file": [], "entry": [1]}}
        ready = {r["id"]: r for r in self.next()["ready_to_start"]}
        self.assertEqual(ready[2]["drift"], drift)
        self.assertNotIn("drift", ready[1])
        self.assertEqual(json.loads(run("find", str(self.epic), "2").stdout)["drift"], drift)

    def test_an_unknown_holds_a_ticket_until_it_is_settled(self):
        self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Scaffold"\nunknown = "Which host?"'))
        out = self.next()
        self.assertEqual(out["ready_to_start"], [])
        row = {e["id"]: e for e in out["blocked"]}[1]
        self.assertEqual(row["unknown"], "Which host?")
        self.assertNotIn("waiting_on", row)
        # Pulled, the file's Notes hold it.
        self.assertEqual(run("pull", str(self.epic), "1").returncode, 0)
        self.assertEqual({e["id"]: e for e in self.next()["blocked"]}[1]["unknown"], "Which host?")
        leaf = tk(self.epic) / "story-scaffold.md"
        text = leaf.read_text(encoding="utf-8")
        leaf.write_text(text.replace("- Unknown: Which host?", "- Decision: the EU host."), encoding="utf-8")
        out = self.next()
        self.assertEqual([e["id"] for e in out["ready_to_start"]], [1])
        self.assertNotIn("unknown", out["ready_to_start"][0])

    def test_an_unknown_does_not_keep_a_ticket_from_refining(self):
        self.breakdown_epic(
            self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Scaffold"\nrefine = true\nunknown = "Which host?"')
        )
        self.assertEqual([e["id"] for e in self.next()["ready_to_refine"]], [1])
        # Refined, and the only `Unknown:` left is a template's example in a comment.
        self.add("story-scaffold.md", ticket("", 1, refined="true") + "<!-- Example:\n- Unknown: not this one\n-->\n")
        self.assertEqual([e["id"] for e in self.next()["ready_to_start"]], [1])

    def test_rows_carry_the_checkpoints_an_entry_sets(self):
        self.breakdown_epic(
            self.BREAKDOWN.replace(
                'title = "Scaffold"', 'title = "Scaffold"\nplan_checkpoint = true\ndone_checkpoint = true'
            )
        )
        first, second = self.next()["ready_to_start"][0], self.status_rows(self.epic)[1]
        self.assertEqual((first["plan_checkpoint"], first["done_checkpoint"]), (True, True))
        self.assertNotIn("plan_checkpoint", second)
        self.assertTrue(json.loads(run("find", str(self.epic), "1").stdout)["done_checkpoint"])

    def test_every_row_carries_the_tickets_risk_and_the_leaf_files_wins(self):
        self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Scaffold"\nrisk = "medium"'))
        rows = {r["id"]: r for r in self.status_rows(self.epic)}
        self.assertEqual((rows[1]["risk"], rows[2]["risk"]), ("medium", ""))
        self.assertEqual(self.next()["ready_to_start"][0]["risk"], "medium")
        self.add("story-scaffold.md", ticket("", 1).replace("risk: low", "risk: high"))
        self.assertEqual(json.loads(run("find", str(self.epic), "1").stdout)["risk"], "high")

    def test_a_plan_without_an_assignee_keeps_the_leafs_and_one_with_it_wins(self):
        self.breakdown_epic()
        self.add("story-scaffold.md", ticket("", 1, assignee='"ann"'))
        self.add("story-scaffold-plan.md", plan(1, "in-progress"))
        self.assertEqual(self.status_rows(self.epic)[0]["assignee"], "ann")
        self.add("story-scaffold-plan.md", plan(1, "in-progress", assignee="bob"))
        self.assertEqual(self.status_rows(self.epic)[0]["assignee"], "bob")

    def mirror(self, items, folder=None):
        return run("mirror", str(folder or self.epic), stdin=json.dumps(items))

    def test_mirror_refuses_the_repo_store(self):
        self.breakdown_epic()
        r = self.mirror([{"ref": "1", "tracker_status": "done"}])
        self.assertEqual(r.returncode, 2)
        self.assertIn("no tracker", r.stderr)
        self.assertEqual(list(tk(self.epic).glob("story-*.md")), [])

    def test_mirror_pulls_a_leaf_and_writes_only_the_keys_given(self):
        self.write_store("jira")
        self.breakdown_epic()
        r = self.mirror(
            [
                {
                    "ref": 2,
                    "tracker_id": "42",
                    "remote": "https://x.test/42",
                    "tracker_status": "in-progress",
                    "assignee": "ann",
                    "after": [1, "41"],
                },
                {"ref": "scaffold", "tracker_id": "41", "tracker_status": "done"},
            ]
        )
        self.assertEqual(r.returncode, 0, r.stderr)
        out = json.loads(r.stdout)
        self.assertEqual(
            [(m["ref"], m["file"], m["pulled"]) for m in out["mirrored"]],
            [("2", "story-ui-shell.md", True), ("1", "story-scaffold.md", True)],
        )
        self.assertEqual(out["unmatched"], [])
        text = (tk(self.epic) / "story-ui-shell.md").read_text(encoding="utf-8")
        for line in (
            'tracker_id: "42"',
            'remote: "https://x.test/42"',
            "tracker_status: in-progress",
            'assignee: "ann"',
            'after: [1, "41"]',
        ):
            self.assertIn(f"\n{line}\n", text)
        self.assertNotIn("\nstatus:", text)
        rows = {t["id"]: t for t in self.status_rows(self.epic)}
        self.assertEqual((rows[2]["state"], rows[2]["assignee"], rows[2]["after"]), ("in-progress", "ann", [1]))
        self.assertEqual(rows[1]["state"], "done")
        self.assertNotIn("drift", rows[2])
        self.assertEqual([e["id"] for e in self.next("--synced")["ready_to_start"]], [3])

    def test_mirror_finds_a_ticket_by_its_tracker_id_and_an_empty_value_removes_the_line(self):
        self.write_store("github")
        self.breakdown_epic()
        self.add("story-scaffold.md", ticket("", 1, tracker_id='"41"', assignee='"ann"', tracker_status="backlog"))
        r = self.mirror([{"tracker_id": "41", "tracker_status": "review", "assignee": ""}])
        self.assertEqual(r.returncode, 0, r.stderr)
        self.assertFalse(json.loads(r.stdout)["mirrored"][0]["pulled"])
        text = (tk(self.epic) / "story-scaffold.md").read_text(encoding="utf-8")
        self.assertIn("\ntracker_status: review\n", text)
        self.assertNotIn("assignee", text)
        self.assertIn('\ntracker_id: "41"\n', text)

    def test_a_tracker_id_alone_matches_only_the_ticket_that_carries_it(self):
        self.write_store("github")
        self.breakdown_epic(self.BREAKDOWN.replace('title = "Codes"', 'title = "Port SHOP-12 codes"'))
        self.add("story-scaffold.md", ticket("", 1, tracker_id='"3"'))
        r = self.mirror([{"tracker_id": 3, "tracker_status": "done"}, {"tracker_id": "SHOP-12", "assignee": "ann"}])
        self.assertEqual(r.returncode, 0, r.stderr)
        out = json.loads(r.stdout)
        self.assertEqual([m["file"] for m in out["mirrored"]], ["story-scaffold.md"])
        self.assertEqual([u["ref"] for u in out["unmatched"]], ["SHOP-12"])
        self.assertEqual([p.name for p in tk(self.epic).glob("s*.md")], ["story-scaffold.md"])
        self.assertEqual(self.status_rows(self.epic)[0]["state"], "done")
        self.add("story-ui-shell.md", ticket("", 2, tracker_id='"3"'))
        r = self.mirror([{"tracker_id": "3", "tracker_status": "backlog"}])
        self.assertEqual(r.returncode, 1)
        self.assertIn("more than one", json.loads(r.stderr)["error"])

    def test_mirror_lists_a_ticket_the_tree_does_not_hold_and_writes_the_rest(self):
        self.write_store("linear")
        self.breakdown_epic()
        r = self.mirror([{"ref": "SHOP-99", "tracker_status": "done"}, {"ref": 1, "tracker_status": "done"}])
        self.assertEqual(r.returncode, 0, r.stderr)
        out = json.loads(r.stdout)
        self.assertEqual([u["ref"] for u in out["unmatched"]], ["SHOP-99"])
        self.assertEqual(self.status_rows(self.epic)[0]["state"], "done")

    def test_a_failed_mirror_still_lists_the_unmatched_tickets(self):
        self.write_store("linear")
        self.breakdown_epic()
        self.add("story-scaffold.md", ticket("", 1, tracker_id='"SHOP-1"'))
        before = (tk(self.epic) / "story-scaffold.md").read_bytes()
        # A known ticket now waits on a tracker item the tree does not hold yet.
        r = self.mirror(
            [{"tracker_id": "SHOP-1", "after": ["SHOP-2"]}, {"tracker_id": "SHOP-2", "tracker_status": "backlog"}]
        )
        self.assertEqual(r.returncode, 1)
        err = json.loads(r.stderr)
        self.assertIn("nothing was mirrored", err["error"])
        self.assertEqual([u["ref"] for u in err["unmatched"]], ["SHOP-2"])
        self.assertEqual((tk(self.epic) / "story-scaffold.md").read_bytes(), before)

    def test_a_tracker_id_on_two_tickets_is_refused_and_a_mirror_that_makes_one_rolls_back(self):
        self.write_store("linear")
        self.breakdown_epic()
        self.add("story-scaffold.md", ticket("", 1, tracker_id='"SHOP-1"'))
        before = (tk(self.epic) / "story-scaffold.md").read_bytes()
        # The same id again, in another case, would make `after = ["SHOP-1"]` point at whichever was read last.
        r = self.mirror([{"ref": 2, "tracker_id": "shop-1"}])
        self.assertEqual(r.returncode, 1)
        self.assertIn("more than one ticket", json.loads(r.stderr)["error"])
        self.assertEqual((tk(self.epic) / "story-scaffold.md").read_bytes(), before)
        self.assertFalse((tk(self.epic) / "story-ui-shell.md").exists())
        self.add("story-ui-shell.md", ticket("", 2, after="[1]", tracker_id='"SHOP-1"'))
        r = run("status", str(self.epic))
        self.assertEqual(r.returncode, 1)
        self.assertIn("tracker_id 'shop-1' is on more than one ticket", json.loads(r.stderr)["error"])

    def test_mirror_writes_nothing_when_a_value_is_refused_or_breaks_the_tree(self):
        self.write_store("jira")
        self.breakdown_epic()
        self.add("story-scaffold.md", ticket("", 1))
        before = (tk(self.epic) / "story-scaffold.md").read_text(encoding="utf-8")
        for items, message in (
            ([{"ref": 1, "tracker_status": "done"}, {"ref": 2, "status": "done"}], "never mirrored"),
            ([{"ref": 1, "tracker_status": "closed"}], "tracker_status"),
            ([{"ref": 1, "tracker_status": "done"}, {"ref": 2, "after": ["NOPE-7"]}], "nothing was mirrored"),
            ([{"ref": 1, "tracker_status": "done"}, {"ref": "1", "assignee": "ann"}], "named twice"),
            ({"ref": 1}, "JSON array"),
        ):
            r = self.mirror(items)
            self.assertEqual(r.returncode, 1, items)
            self.assertIn(message, json.loads(r.stderr)["error"])
            self.assertEqual((tk(self.epic) / "story-scaffold.md").read_text(encoding="utf-8"), before)
            self.assertFalse((tk(self.epic) / "story-ui-shell.md").exists())
        r = run("mirror", str(self.epic), stdin="not json")
        self.assertEqual(r.returncode, 1)

    def test_mirror_keeps_a_leafs_crlf_line_endings(self):
        self.write_store("jira")
        self.breakdown_epic()
        (tk(self.epic) / "story-scaffold.md").write_bytes(ticket("", 1).replace("\n", "\r\n").encode("utf-8"))
        self.assertEqual(self.mirror([{"ref": 1, "tracker_status": "done"}]).returncode, 0)
        raw = (tk(self.epic) / "story-scaffold.md").read_bytes()
        self.assertIn(b"\r\ntracker_status: done\r\n", raw)
        self.assertNotIn(b"\n", raw.replace(b"\r\n", b""))

    def test_a_pulled_leaf_matches_its_entry_until_the_entry_changes(self):
        self.breakdown_epic()
        for n in ("2", "3"):
            self.assertEqual(run("pull", str(self.epic), n).returncode, 0)
        self.assertFalse(any("drift" in r for r in self.status_rows(self.epic)))
        self.breakdown_epic(self.BREAKDOWN.replace("hitl = true", "hitl = false"))
        rows = {r["id"]: r for r in self.status_rows(self.epic)}
        self.assertTrue(rows[3]["drift"])
        self.assertTrue(rows[3]["hitl"])

    def test_pull_writes_the_leaf_and_only_a_refine_entry_waits_for_refinement(self):
        self.breakdown_epic(
            self.BREAKDOWN.replace(
                'title = "Scaffold"', 'title = "Scaffold: the cart!"\nverify = "It runs."\nrefine = true'
            )
        )
        r = run("pull", str(self.epic), "1")
        self.assertEqual(r.returncode, 0, r.stderr)
        self.assertEqual(json.loads(r.stdout), {"file": "story-scaffold-the-cart.md", "refine": True})
        text = (tk(self.epic) / "story-scaffold-the-cart.md").read_text(encoding="utf-8")
        self.assertIn("## Requirements\n\nVerify: It runs.", text)
        self.assertIn("covers: [R1]", text)
        self.assertEqual(self.files(self.next()["ready_to_refine"]), ["story-scaffold-the-cart.md"])
        self.assertEqual(run("pull", str(self.epic), "1").returncode, 1)
        run("mark", str(self.epic), "1", "done")
        run("pull", str(self.epic), "2")
        out = self.next()
        self.assertEqual(self.files(out["ready_to_start"]), ["story-ui-shell.md", None])
        self.assertEqual(out["ready_to_start"][0]["after"], [1])

    def test_a_bug_waits_for_refinement_without_refine_set(self):
        self.breakdown_epic('[[entry]]\nid = 1\ntype = "bug"\ntitle = "Missing criteria"\n')
        out = self.next()
        self.assertEqual([t["id"] for t in out["ready_to_refine"]], [1])
        self.assertEqual(out["ready_to_start"], [])
        r = run("pull", str(self.epic), "1")
        self.assertEqual(r.returncode, 0, r.stderr)
        self.assertEqual(json.loads(r.stdout), {"file": "bug-missing-criteria.md", "refine": True})
        self.assertEqual(self.files(self.next()["ready_to_refine"]), ["bug-missing-criteria.md"])

    def test_a_leaf_file_that_says_bug_waits_for_refinement(self):
        self.breakdown_epic('[[entry]]\nid = 1\ntype = "story"\ntitle = "Was a story"\n')
        self.add("bug-was-a-story.md", ticket("ready-for-dev", 1, kind="bug"))
        self.assertEqual(self.files(self.next()["ready_to_refine"]), ["bug-was-a-story.md"])

    def test_a_refined_bug_is_ready_to_start(self):
        self.breakdown_epic('[[entry]]\nid = 1\ntype = "bug"\ntitle = "Refined bug"\n')
        self.add("bug-refined-bug.md", ticket("ready-for-dev", 1, kind="bug", refined="true"))
        out = self.next()
        self.assertEqual(self.files(out["ready_to_start"]), ["bug-refined-bug.md"])
        self.assertEqual(out["ready_to_refine"], [])

    def test_pull_leaves_out_empty_fields_but_always_writes_after_and_hitl(self):
        self.breakdown_epic()
        run("pull", str(self.epic), "1")
        head = (tk(self.epic) / "story-scaffold.md").read_text(encoding="utf-8").split("---")[1]
        self.assertEqual(
            head, '\nid: 1\ntype: story\ntitle: "Scaffold"\nparent: epic-cart\ncovers: [R1]\nafter: []\nhitl: false\n'
        )
        self.assertEqual(self.next()["ready_to_start"][0]["state"], "backlog")
        run("pull", str(self.epic), "3")
        head = (tk(self.epic) / "spike-tax-engine.md").read_text(encoding="utf-8").split("---")[1]
        self.assertIn("\nafter: [1]\n", head)
        self.assertIn("\nhitl: true\n", head)
        self.assertNotIn("refined", head)

    def test_pull_ends_after_the_references(self):
        self.breakdown_epic()
        run("pull", str(self.epic), "1")
        text = (tk(self.epic) / "story-scaffold.md").read_text(encoding="utf-8")
        self.assertTrue(
            text.endswith("## References\n\n- parent — out/initiative-checkout/tickets/epic-cart/epic-cart.md\n"), text
        )

    def test_find_by_cross_ref_id_file_tracker_id_and_title(self):
        pricing = self.pricing()
        self.breakdown_epic()
        self.add("story-contract.md", ticket("done", 1, tracker_id='"PRICE-1"'), pricing)
        run("pull", str(self.epic), "1")

        def find(folder, ref):
            r = run("find", str(folder), ref)
            self.assertEqual(r.returncode, 0, r.stderr)
            return json.loads(r.stdout)

        hit = find(self.initiative, "2.1")
        self.assertEqual((hit["folder"], hit["id"], hit["file"]), ("epic-cart", 1, "story-scaffold.md"))
        self.assertEqual(hit["story_file"], str((tk(self.epic) / "story-scaffold.md").resolve()))
        self.assertNotIn("path", hit)
        self.assertEqual(find(self.epic, "3")["title"], "Tax engine?")
        self.assertIsNone(find(self.epic, "3")["story_file"])
        self.assertEqual(find(self.epic, "1.1")["file"], "story-contract.md")
        self.assertEqual(find(self.epic, "price-1")["file"], "story-contract.md")
        self.assertEqual(find(self.initiative, "story-scaffold")["folder"], "epic-cart")
        self.assertEqual(find(self.initiative, "ui shell")["id"], 2)
        self.assertEqual(find(self.epic, "codes")["id"], 4)
        r = run("find", str(self.initiative), "in")
        self.assertEqual(r.returncode, 1)
        self.assertIn("more than one", r.stderr)
        r = run("find", str(self.initiative), "9.1")
        self.assertEqual(r.returncode, 1)
        self.assertIn("no ticket matches", r.stderr)

    def test_pull_refuses_a_file_name_already_taken(self):
        self.breakdown_epic()
        self.add("story-ui-shell.md", "notes, not a ticket\n")
        r = run("pull", str(self.epic), "2")
        self.assertEqual(r.returncode, 1)
        self.assertIn("exists already", r.stderr)

    def test_pull_writes_references_and_notes(self):
        self.breakdown_epic(
            self.BREAKDOWN.replace(
                'title = "Scaffold"',
                'title = "Scaffold"\nunknown = "Which host?"\nreferences = ["SPINE.md#ad-8"]\nnotes = ["Reuse the mailer."]',
            )
        )
        self.assertEqual(run("pull", str(self.epic), "1").returncode, 0)
        text = (tk(self.epic) / "story-scaffold.md").read_text(encoding="utf-8")
        parent = (self.epic / "epic-cart.md").resolve().relative_to(self.root.resolve()).as_posix()
        self.assertIn(f"## References\n\n- parent — {parent}\n- SPINE.md#ad-8\n", text)
        self.assertIn("## Notes\n\n- Unknown: Which host?\n- Reuse the mailer.\n", text)

    def test_a_quoted_title_survives_the_pull(self):
        self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', "title = 'Say \"hi\"'"))
        self.assertEqual(run("pull", str(self.epic), "1").returncode, 0)
        self.assertEqual(self.next()["ready_to_start"][0]["title"], 'Say "hi"')

    def test_references_must_be_a_list(self):
        self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Scaffold"\nreferences = "SPINE.md"'))
        r = run("status", str(self.epic))
        self.assertEqual(r.returncode, 1)
        self.assertIn("`references` must be a list", r.stderr)

    def test_dropped_prerequisite_still_blocks(self):
        self.breakdown_epic()
        self.add("story-scaffold.md", ticket("dropped", 1))
        self.add("story-ui-shell.md", ticket("", 2, after="[1]"))
        out = self.next()
        self.assertEqual(self.files(out["blocked"])[0], "story-ui-shell.md")
        self.assertEqual(out["ready_to_refine"], [])
        self.assertEqual(out["ready_to_start"], [])

    def test_malformed_breakdown_errors(self):
        for text, message in (
            (self.BREAKDOWN.replace("id = 3", "id = 2"), "two entries with id 2"),
            (self.BREAKDOWN.replace("after = [2, 3]", "after = [2, 9]"), "names no entry in epic-cart"),
            (self.BREAKDOWN.replace('type = "spike"', 'type = "task"'), "is not one of"),
            (self.BREAKDOWN.replace("id = 4", 'id = "3"'), "two entries with id 3"),
            (self.BREAKDOWN.replace("id = 4", 'id = "4-b"'), "needs an `id`"),
            (self.BREAKDOWN.replace("id = 4", 'id = "true"'), "needs an `id`"),
            (self.BREAKDOWN.replace("id = 4\n", ""), "needs an `id`"),
            ("[[entry]\nid = ", "tickets.toml"),
            ('[entry]\nid = 1\ntype = "story"\n', "[[entry]]"),
            (self.BREAKDOWN.replace('covers = ["R4"]', "covers = 1"), "must be a list"),
        ):
            self.breakdown_epic(text)
            r = run("next", str(self.epic))
            self.assertEqual(r.returncode, 1, text)
            self.assertIn(message, json.loads(r.stderr)["error"])

    def test_ticket_waits_on_an_entry_in_another_epic(self):
        pricing = self.pricing()
        self.breakdown_epic(
            self.BREAKDOWN.replace('title = "UI shell"\nafter = [1]', 'title = "UI shell"\nafter = [1, "1.1"]')
        )
        self.add("story-scaffold.md", ticket("done", 1))
        out = self.next()
        self.assertEqual([e["id"] for e in out["ready_to_start"]], [3])
        self.assertEqual(out["blocked"][0]["after"], [1, "1.1"])
        self.add("story-contract.md", ticket("done", 1), pricing)
        self.assertEqual([e["id"] for e in self.next()["ready_to_start"]], [2, 3])

    def test_whole_epic_prerequisite_and_epic_gate(self):
        pricing = self.pricing()
        self.add("story-scaffold.md", ticket("draft", 1, after="[epic-pricing]"))
        self.assertEqual(self.files(self.next()["blocked"]), ["story-scaffold.md"])
        (pricing / "epic-pricing.md").write_text("---\ntype: epic\nstatus: done\n---\n")
        self.assertEqual(self.files(self.next()["ready_to_refine"]), ["story-scaffold.md"])
        gated = self.add_epic("epic-tax", after="[epic-cart]")
        self.add("story-rates.md", ticket("draft", 1), gated)
        out = json.loads(run("next", str(gated)).stdout)
        self.assertEqual(out["blocked"][0]["after"], [])
        self.assertEqual(out["blocked"][0]["gated_by"], ["epic-cart"])

    def test_gate_deadlock_with_a_ticketless_epic_is_a_cycle(self):
        self.add_epic("epic-tax", after="[epic-cart]")
        self.add("story-scaffold.md", ticket("draft", 1, after="[epic-tax]"))
        r = run("next", str(self.epic))
        self.assertEqual(r.returncode, 1)
        self.assertIn("cycle", r.stderr)

    def test_tracker_ids_resolve_across_epics_and_duplicates_collapse(self):
        pricing = self.pricing()
        self.add("story-contract.md", ticket("done", 1, tracker_id='"PRICE-1"'), pricing)
        self.add("story-scaffold.md", ticket("draft", 1, after="[PRICE-1, 1.1, 1.01]"))
        out = self.next()
        self.assertEqual(out["ready_to_refine"][0]["after"], ["1.1"])

    def test_malformed_files_name_their_folder(self):
        pricing = self.pricing()
        self.seed()
        cases = (
            (ticket("todo", 1), "epic-pricing/tickets/story-contract.md"),
            (ticket("draft", 1).replace("after: []", "after:\n  - 1"), "must be inline"),
            (ticket("draft", 1).replace("---\n\n# x", "\n# x"), "does not close"),
        )
        for text, message in cases:
            self.add("story-contract.md", text, pricing)
            self.assertIn(message, json.loads(run("next", str(self.epic)).stderr)["error"])
        self.add("story-contract.md", ticket("draft", 1), pricing)
        (pricing / "epic-pricing.md").write_text("---\ntype: epic\nstatus: Finished\n---\n")
        self.assertIn("epic-pricing/epic-pricing.md", json.loads(run("next", str(self.epic)).stderr)["error"])

    def test_cross_epic_unknown_target_and_cycle_error(self):
        pricing = self.pricing()
        for ref, message in (("9.1", "names no epic id"), ("1.9", "names no entry"), ("epic-missing", "names no epic")):
            self.add("story-scaffold.md", ticket("draft", 1, after=f"[{ref}]"))
            r = run("next", str(self.epic))
            self.assertEqual(r.returncode, 1)
            self.assertIn(message, json.loads(r.stderr)["error"])
        self.add("story-scaffold.md", ticket("draft", 1, after="[1.2]"))
        self.add("story-contract.md", ticket("draft", 1, after="[2.1]"), pricing)
        r = run("next", str(self.epic))
        self.assertEqual(r.returncode, 1)
        self.assertIn("cycle", r.stderr)

    def test_initiative_view_spans_epics_in_file_order_and_reports_unpinned_after(self):
        self.pricing()
        self.breakdown_epic()
        (tk(self.initiative) / "tickets.toml").write_text(
            '[[epic]]\nid = 1\nslug = "epic-pricing"\n\n'
            '[[epic]]\nid = 2\nslug = "epic-cart"\nafter = [{ epic = 1, needs = "the pricing contract" }]\n'
        )
        out = json.loads(run("next", str(self.initiative)).stdout)
        self.assertEqual([(e["epic"], e["id"]) for e in out["ready_to_start"]], [("epic-pricing", 1), ("epic-cart", 1)])
        self.assertEqual(
            out["unpinned_after"], [{"epic": "epic-cart", "after": "epic-pricing", "needs": "the pricing contract"}]
        )
        self.breakdown_epic(
            self.BREAKDOWN.replace('title = "UI shell"\nafter = [1]', 'title = "UI shell"\nafter = [1, "1.1"]')
        )
        status = json.loads(run("status", str(self.initiative)).stdout)
        self.assertEqual(status["unpinned_after"], [])
        self.assertEqual(status["counts"], {"total": 6, "planned": 6})
        self.assertEqual([(e["slug"], e["id"]) for e in status["epics"]], [("epic-pricing", 1), ("epic-cart", 2)])
        self.assertEqual(status["tickets"][0]["blocks"], [2, "2.2"])
        self.assertEqual(status["longest_remaining_chain"][1:], ["2.2", "2.4"])
        (tk(self.initiative) / "tickets.toml").write_text(
            '[[epic]]\nid = 1\nslug = "epic-pricing"\n\n[[epic]]\nid = 2\nslug = "epic-cart"\nafter = [{ epic = "epic-x" }]\n'
        )
        r = run("status", str(self.initiative))
        self.assertEqual(r.returncode, 1)
        self.assertIn("no epic listed", r.stderr)
        (tk(self.initiative) / "tickets.toml").write_text(
            '[[epic]]\nid = 2\nslug = "epic-cart"\n\n[[epic]]\nid = 2\nslug = "epic-pricing"\n'
        )
        self.assertIn("two epics with id 2", run("status", str(self.initiative)).stderr)

    def test_status_shows_the_declared_epic_graph_before_and_after_inception(self):
        self.pricing()
        (tk(self.initiative) / "tickets.toml").write_text(
            '[[epic]]\nid = 1\nslug = "epic-pricing"\n\n'
            '[[epic]]\nid = 2\nslug = "epic-cart"\nafter = [{ epic = 1, needs = "the pricing contract" }]\n'
        )
        for _ in range(2):
            status = json.loads(run("status", str(self.initiative)).stdout)
            self.assertEqual(
                [(e["slug"], e["after"], e["gated_by"]) for e in status["epics"]],
                [
                    ("epic-pricing", [], []),
                    ("epic-cart", [{"epic": "epic-pricing", "needs": "the pricing contract"}], []),
                ],
            )
            self.breakdown_epic()

    def test_an_undeclared_or_backward_cross_epic_after_is_reported(self):
        self.pricing()
        self.breakdown_epic()
        (tk(self.initiative) / "tickets.toml").write_text(
            '[[epic]]\nid = 1\nslug = "epic-pricing"\n\n[[epic]]\nid = 2\nslug = "epic-cart"\n'
        )
        (tk(tk(self.initiative) / "epic-pricing") / "tickets.toml").write_text(
            '[[entry]]\nid = 1\ntype = "story"\ntitle = "Pricing contract"\nafter = ["2.1"]\n'
        )
        for command, folder in (("status", self.initiative), ("next", self.initiative), ("status", self.epic)):
            out = json.loads(run(command, str(folder)).stdout)
            if folder == self.epic:
                self.assertEqual((out["undeclared_after"], out["order_conflict"]), ([], []))
                continue
            self.assertEqual(
                out["undeclared_after"], [{"epic": "epic-pricing", "after": "epic-cart", "ref": "1.1", "names": "2.1"}]
            )
            self.assertEqual(out["order_conflict"], [{"epic": "epic-pricing", "after": "epic-cart"}])
        (tk(self.initiative) / "tickets.toml").write_text(
            '[[epic]]\nid = 1\nslug = "epic-pricing"\nafter = [{ epic = 2, needs = "the cart" }]\n\n'
            '[[epic]]\nid = 2\nslug = "epic-cart"\n'
        )
        out = json.loads(run("status", str(self.initiative)).stdout)
        self.assertEqual(out["undeclared_after"], [])
        self.assertEqual(out["order_conflict"], [{"epic": "epic-pricing", "after": "epic-cart"}])

    def test_after_naming_a_backlog_file_says_to_move_it_into_the_epic(self):
        self.add("story-thing.md", ticket("", kind="story"), self.add_backlog())
        self.add("story-scaffold.md", ticket("draft", 1, after="[story-thing]"))
        r = run("next", str(self.epic))
        self.assertEqual(r.returncode, 1)
        self.assertIn("move a backlog ticket into the epic as an entry", json.loads(r.stderr)["error"])
        self.breakdown_epic()
        self.add("story-scaffold.md", ticket("draft", 1, after="[story-ui-shell]"))
        r = run("next", str(self.epic))
        self.assertEqual(r.returncode, 1)
        self.assertIn("use the entry's id", json.loads(r.stderr)["error"])

    def test_an_epic_gate_counts_once_and_its_order_before_inception(self):
        self.pricing()
        (tk(self.initiative) / "tickets.toml").write_text(
            '[[epic]]\nid = 1\nslug = "epic-cart"\n\n[[epic]]\nid = 2\nslug = "epic-pricing"\n'
        )
        (self.epic / "epic-cart.md").write_text("---\ntype: epic\nafter: [epic-pricing]\n---\n# epic-cart\n")
        out = json.loads(run("status", str(self.initiative)).stdout)
        self.assertEqual(
            (out["undeclared_after"], out["order_conflict"]), ([], [{"epic": "epic-cart", "after": "epic-pricing"}])
        )
        self.breakdown_epic('[[entry]]\nid = 1\ntype = "story"\ntitle = "Scaffold"\nafter = ["epic-pricing"]\n')
        out = json.loads(run("status", str(self.initiative)).stdout)
        self.assertEqual(
            out["undeclared_after"],
            [{"epic": "epic-cart", "after": "epic-pricing", "ref": "1.1", "names": "epic-pricing"}],
        )
        self.assertEqual(out["order_conflict"], [{"epic": "epic-cart", "after": "epic-pricing"}])

    def test_epics_need_an_id_a_slug_and_a_valid_after(self):
        self.pricing()
        toml = tk(self.initiative) / "tickets.toml"
        good = toml.read_text(encoding="utf-8")
        for bad, message in (
            (good.replace("id = 2\n", ""), "needs an `id`"),
            (good.replace('slug = "epic-cart"', 'slug = "epic-pricing"'), "two epics with slug"),
            (good.replace('slug = "epic-cart"\n', ""), "needs a `slug`"),
            (good + "after = [{ epic = true }]\n", "epic = <id or slug>"),
        ):
            toml.write_text(bad, encoding="utf-8")
            r = run("status", str(self.initiative))
            self.assertEqual(r.returncode, 1, r.stdout)
            self.assertIn(message, r.stderr)

    def test_find_prefers_the_folder_asked_about_for_a_file_name(self):
        pricing = self.pricing()
        self.add("story-scaffold.md", ticket("", 1), pricing)
        self.breakdown_epic()
        run("pull", str(self.epic), "1")
        r = run("find", str(pricing), "story-scaffold")
        self.assertEqual(r.returncode, 0, r.stderr)
        self.assertEqual(json.loads(r.stdout)["folder"], "epic-pricing")
        r = run("find", str(self.initiative), "story-scaffold")
        self.assertEqual(r.returncode, 1)
        self.assertIn("more than one", r.stderr)

    def test_a_numeric_tracker_id_is_never_a_sibling_id(self):
        self.seed(s1="done")
        self.add("story-ui-shell.md", ticket("done", 2, tracker_id='"47"'))
        self.add("story-tracer.md", ticket("", 3, after='["47"]'))
        self.assertIn("story-tracer.md", self.files(self.next()["ready_to_refine"]))

    def test_a_quoted_number_is_a_tracker_id_and_a_bare_one_a_sibling(self):
        self.add("story-scaffold.md", ticket("done", 1, tracker_id="101"))
        self.add("story-ui-shell.md", ticket("draft", 2, after='["101"]'))
        self.assertEqual(self.files(self.next()["ready_to_refine"]), ["story-ui-shell.md"])
        self.add("story-ui-shell.md", ticket("draft", 2, after="[101]"))
        self.assertIn("names no entry", run("next", str(self.epic)).stderr)

    def test_an_id_takes_letters_and_digits_wherever_a_number_goes(self):
        pricing = self.add_epic("epic-pricing")
        (tk(pricing) / "tickets.toml").write_text(
            '[[entry]]\nid = "6a"\ntype = "story"\ntitle = "Pricing contract"\n\n'
            '[[entry]]\nid = 7\ntype = "story"\ntitle = "Pricing rules"\nafter = ["6a"]\n'
        )
        (tk(self.initiative) / "tickets.toml").write_text(
            '[[epic]]\nid = "1b"\nslug = "epic-pricing"\n\n'
            '[[epic]]\nid = 2\nslug = "epic-cart"\nafter = [{ epic = "1b", needs = "prices" }]\n'
        )
        self.breakdown_epic('[[entry]]\nid = "2a"\ntype = "story"\ntitle = "Totals for 6a"\nafter = ["1b.6a"]\n')
        status = json.loads(run("status", str(self.initiative)).stdout)
        rows = {r["ref"]: r for r in status["tickets"]}
        self.assertEqual(list(rows), ["1b.6a", "1b.7", "2.2a"])
        self.assertEqual((rows["1b.7"]["after"], rows["2.2a"]["after"]), (["6a"], ["1b.6a"]))
        self.assertEqual(status["unpinned_after"] + status["undeclared_after"], [])

        self.assertEqual(run("pull", str(pricing), "7").returncode, 0)
        self.assertIn("id: 7\n", (tk(pricing) / "story-pricing-rules.md").read_text(encoding="utf-8"))
        self.assertIn("after: [6a]\n", (tk(pricing) / "story-pricing-rules.md").read_text(encoding="utf-8"))
        self.assertEqual(run("pull", str(pricing), "6a").returncode, 0)
        self.assertIn("id: 6a\n", (tk(pricing) / "story-pricing-contract.md").read_text(encoding="utf-8"))
        self.assertNotIn("drift", json.dumps(self.status_rows(pricing)))

        self.assertEqual([e["id"] for e in self.next()["blocked"]], ["2a"])
        self.add("story-pricing-contract-plan.md", plan("6a", "done"), pricing)
        self.assertEqual([e["id"] for e in self.next()["ready_to_start"]], ["2a"])
        # A bare lettered id names its ticket from any folder, before the title that contains it.
        for folder, ref in ((self.initiative, "1b.6a"), (pricing, "6a"), (self.initiative, "6a"), (self.epic, "6a")):
            found = json.loads(run("find", str(folder), ref).stdout)
            self.assertEqual((found["id"], found["state"]), ("6a", "done"))
        self.assertEqual(run("mark", str(self.epic), "2a", "in-review").returncode, 0)
        self.assertIn("ticket: 2a\n", (tk(self.epic) / "story-totals-for-6a-plan.md").read_text(encoding="utf-8"))
        self.assertEqual(self.status_rows(self.epic)[0]["state"], "review")

    def test_two_files_sharing_an_id_error(self):
        self.add("story-a.md", ticket("done", 1))
        self.add("story-b.md", ticket("draft", 1))
        r = run("next", str(self.epic))
        self.assertEqual(r.returncode, 1)
        self.assertIn("share the id 1", json.loads(r.stderr)["error"])

    def test_malformed_input_returns_json_error(self):
        (tk(self.epic) / "story-bad.md").write_bytes(ticket("draft", 1).encode().replace(b"# x", b"\xff"))
        r = run("next", str(self.epic))
        self.assertEqual(r.returncode, 1, r.stderr)
        self.assertIn("error", json.loads(r.stderr))
        (tk(self.epic) / "story-bad.md").unlink()
        self.seed()
        (self.root / "_bmad" / "custom" / "ticketing-store-config.toml").write_text("[tickets\nstore = ")
        r = run("next", str(self.epic))
        self.assertEqual(r.returncode, 1, r.stderr)
        self.assertIn("error", json.loads(r.stderr))

    def test_leaf_without_id_sorts_last_and_container_file_is_ignored(self):
        self.seed()
        self.add("bug-stray.md", ticket("", kind="bug"))
        out = json.loads(run("status", str(self.epic)).stdout)
        self.assertEqual(out["tickets"][-1]["file"], "bug-stray.md")
        self.assertIsNone(out["tickets"][-1]["id"])
        self.assertEqual(out["counts"]["total"], 6)

    def mark(self, *args):
        r = run("mark", *args)
        self.assertEqual(r.returncode, 0, r.stderr)
        return json.loads(r.stdout)

    def test_mark_writes_a_plan_beside_the_leaf_and_never_the_leaf(self):
        self.seed()
        leaf = (tk(self.epic) / "story-scaffold.md").read_text(encoding="utf-8")
        out = self.mark(str(self.epic), "1", "in-progress", "--assignee", "ann")
        plan_file = tk(self.epic) / "story-scaffold-plan.md"
        self.assertEqual(
            out,
            {
                "plan": str(plan_file.resolve()),
                "created": True,
                "status": "in-progress",
                "assignee": "ann",
                "blocked_at": "",
                "blocked_reason": "",
            },
        )
        self.assertEqual((tk(self.epic) / "story-scaffold.md").read_text(encoding="utf-8"), leaf)
        self.assertEqual(
            plan_file.read_text(encoding="utf-8"),
            '---\ntitle: "x"\nticket: 1\nstatus: in-progress\nassignee: "ann"\n---\n',
        )
        self.assertEqual(self.files(self.next()["in_progress"]), ["story-scaffold.md"])

    def test_mark_clears_the_leafs_blocking_fields_and_takes_a_literal_assignee(self):
        path = tk(self.epic) / "story-scaffold.md"
        path.write_text(ticket("", 1, blocked_at="2026-09-05", blocked_reason="legal"))
        out = self.mark(str(self.epic), "1", "draft", "--assignee", "\\1")
        self.assertEqual(out["assignee"], "\\1")
        text = (tk(self.epic) / "story-scaffold-plan.md").read_text(encoding="utf-8")
        self.assertNotIn("blocked_at", text)
        self.assertIn('assignee: "\\\\1"\n', text)
        self.assertIn("blocked_reason", path.read_text(encoding="utf-8"))
        self.assertEqual(self.files(self.next()["ready_to_refine"]), ["story-scaffold.md"])

    def test_mark_on_a_pulled_leaf_keeps_the_leafs_assignee_in_the_new_plan(self):
        self.breakdown_epic()
        run("pull", str(self.epic), "1")
        self.add(
            "story-scaffold.md",
            (tk(self.epic) / "story-scaffold.md").read_text().replace("---\n\n", 'assignee: "bob"\n---\n\n', 1),
        )
        self.mark(str(self.epic), "1", "ready-for-dev")
        text = (tk(self.epic) / "story-scaffold-plan.md").read_text(encoding="utf-8")
        self.assertEqual(text, '---\ntitle: "Scaffold"\nticket: 1\nstatus: ready-for-dev\nassignee: "bob"\n---\n')
        self.assertEqual(
            [(e["file"], e["assignee"]) for e in self.next()["ready_to_start"]], [("story-scaffold.md", "bob")]
        )

    def test_find_on_an_entry_with_nothing_yet(self):
        self.breakdown_epic(
            self.BREAKDOWN.replace(
                'title = "UI shell"',
                'title = "UI shell"\ndescription = "The page frame."\nverify = "It renders."\n'
                'unknown = "Which grid?"\nreferences = ["SPINE.md#ad-8"]\nnotes = ["Reuse the header."]',
            )
        )
        r = run("find", str(self.epic), "2")
        self.assertEqual(r.returncode, 0, r.stderr)
        hit = json.loads(r.stdout)
        epic = self.epic.resolve()
        self.assertEqual((hit["id"], hit["title"], hit["state"], hit["after"]), (2, "UI shell", "planned", [1]))
        self.assertEqual(
            {k: hit[k] for k in ("folder", "description", "verify", "references", "notes", "unknown")},
            {
                "folder": "epic-cart",
                "description": "The page frame.",
                "verify": "It renders.",
                "references": ["SPINE.md#ad-8"],
                "notes": ["Reuse the header."],
                "unknown": "Which grid?",
            },
        )
        self.assertIsNone(hit["story_file"])
        self.assertEqual(hit["plan"], str(tk(epic) / "story-ui-shell-plan.md"))
        self.assertFalse(Path(hit["plan"]).exists())
        self.assertEqual(hit["epic_file"], str(epic / "epic-cart.md"))

    def test_find_names_a_joined_plan_whatever_its_name(self):
        self.breakdown_epic()
        self.add("custom-name.md", plan(2, "in-progress"))
        hit = json.loads(run("find", str(self.epic), "2").stdout)
        self.assertEqual(hit["plan"], str(tk(self.epic).resolve() / "custom-name.md"))
        self.assertEqual(hit["status"], "in-progress")

    def test_find_on_a_refined_entry_names_its_story_file_and_the_plan_after_it(self):
        self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Scaffold, renamed"'))
        self.add("story-scaffold.md", ticket("", 1))
        hit = json.loads(run("find", str(self.epic), "1").stdout)
        self.assertEqual(hit["story_file"], str(tk(self.epic).resolve() / "story-scaffold.md"))
        self.assertEqual(hit["plan"], str(tk(self.epic).resolve() / "story-scaffold-plan.md"))

    def test_find_on_a_pulled_entry_leaves_the_text_to_the_file(self):
        self.breakdown_epic(
            self.BREAKDOWN.replace(
                'title = "Scaffold"',
                'title = "Scaffold"\ndescription = "The service."\nverify = "It runs."\n'
                'unknown = "Which host?"\nreferences = ["SPINE.md#ad-8"]\nnotes = ["Reuse the mailer."]',
            )
        )
        self.assertEqual(run("pull", str(self.epic), "1").returncode, 0)
        leaf = tk(self.epic) / "story-scaffold.md"
        leaf.write_text(leaf.read_text(encoding="utf-8").replace("Which host?", "Which region?"), encoding="utf-8")
        hit = json.loads(run("find", str(self.epic), "1").stdout)
        self.assertEqual([hit[k] for k in ("description", "verify", "references", "notes")], ["", "", [], []])
        self.assertEqual(hit["unknown"], "Which region?")
        self.assertEqual(hit["story_file"], str(tk(self.epic).resolve() / "story-scaffold.md"))

    def test_find_names_the_initiative_file_for_a_leaf_directly_under_it(self):
        solo = self.root / "out" / "initiative-solo"
        solo.mkdir()
        (solo / "initiative-solo.md").write_text("---\ntype: initiative\n---\n# Solo\n")
        (tk(solo) / "tickets.toml").write_text('[[entry]]\nid = 1\ntype = "story"\ntitle = "Only"\n')
        hit = json.loads(run("find", str(solo), "1").stdout)
        self.assertEqual(hit["epic_file"], str(solo.resolve() / "initiative-solo.md"))
        self.assertIsNone(hit["story_file"])

    def test_find_and_mark_in_a_backlog(self):
        backlog = self.add_backlog()
        self.add("bug-x.md", ticket("draft", kind="bug"), backlog)
        r = run("find", str(backlog), "bug-x")
        self.assertEqual(r.returncode, 0, r.stderr)
        hit = json.loads(r.stdout)
        self.assertIsNone(hit["epic_file"])
        self.assertEqual(
            [hit[k] for k in ("description", "verify", "references", "notes", "unknown")], ["", "", [], [], ""]
        )
        self.assertEqual(hit["plan"], str(backlog.resolve() / "bug-x-plan.md"))
        self.mark(str(backlog), "bug-x", "in-review")
        self.assertIn('\nticket: "bug-x"\n', (backlog / "bug-x-plan.md").read_text(encoding="utf-8"))
        rows = self.status_rows(backlog)
        self.assertEqual([(t["file"], t["state"]) for t in rows], [("bug-x.md", "review")])

    def test_mark_done_on_a_plan_only_entry_clears_blocking_and_keeps_the_plan(self):
        self.breakdown_epic()
        path = tk(self.epic) / "story-ui-shell-plan.md"
        self.add(path.name, plan(2, "blocked", assignee="ann", blocked_at="2026-09-05", blocked_reason="legal"))
        before = path.read_text(encoding="utf-8")
        out = self.mark(str(self.epic), "2", "done")
        self.assertEqual((out["created"], out["status"], out["assignee"]), (False, "done", "ann"))
        text = path.read_text(encoding="utf-8")
        self.assertIn("\nstatus: done\n", text)
        self.assertNotIn("blocked_", text)
        expected = [
            line for line in before.splitlines() if not line.startswith(("status:", "blocked_at:", "blocked_reason:"))
        ]
        self.assertEqual([line for line in text.splitlines() if not line.startswith("status:")], expected)
        rows = self.status_rows(self.epic)
        self.assertEqual(rows[1]["state"], "done")
        out = self.next()
        self.assertNotIn(2, [e["id"] for group in ("ready_to_start", "in_progress", "blocked") for e in out[group]])

    def test_mark_on_a_build_plan_keeps_its_body_and_other_frontmatter(self):
        self.breakdown_epic()
        path = tk(self.epic) / "story-scaffold-plan.md"
        text = plan(1, "in-progress").replace("# x\n", "# x\n\nbody: stays\n") + "## Code Review\n"
        self.add(path.name, text)
        self.mark(str(self.epic), "1", "in-review", "--assignee", "ann")
        after = path.read_text(encoding="utf-8")
        self.assertEqual(
            after,
            text.replace(
                "status: 'in-progress' # draft | ready-for-dev | in-progress | in-review | built | done",
                "status: in-review",
            ).replace("\n---\n\n# x", '\nassignee: "ann"\n---\n\n# x'),
        )

    def test_a_blocked_reason_with_colons_and_quotes_reads_back_exactly(self):
        self.breakdown_epic()
        for reason in ('waiting on legal: "terms"', "it's 'odd' \\ and: # not a comment   # still not", "Café ✓
next"):
            out = self.mark(str(self.epic), "2", "blocked", "--blocked", reason)
            self.assertEqual(out["blocked_reason"], reason)
            row = self.next()["blocked"][0]
            self.assertEqual((row["id"], row["status"], row["blocked_reason"]), (2, "blocked", reason))

    def test_mark_with_no_plan_creates_one_that_next_reads(self):
        self.breakdown_epic()
        self.mark(str(self.epic), "2", "ready-for-dev", "--assignee", "ann")
        self.assertEqual(
            (tk(self.epic) / "story-ui-shell-plan.md").read_text(encoding="utf-8"),
            '---\ntitle: "UI shell"\nticket: 2\nstatus: ready-for-dev\nassignee: "ann"\n---\n',
        )
        self.assertEqual(
            [(e["id"], e["state"], e["status"], e["assignee"]) for e in self.next()["blocked"]][0],
            (2, "backlog", "ready-for-dev", "ann"),
        )

    def test_mark_blocked_writes_the_design_notes_plan(self):
        self.breakdown_epic()
        self.mark(str(self.epic), "UI shell", "blocked", "--blocked", 'waiting on legal: "terms"')
        today = date.today().isoformat()
        self.assertEqual(
            (tk(self.epic) / "story-ui-shell-plan.md").read_text(encoding="utf-8"),
            f'---\ntitle: "UI shell"\nticket: 2\nstatus: blocked\nblocked_at: "{today}"\n'
            'blocked_reason: "waiting on legal: \\"terms\\""\n---\n',
        )

    def test_mark_refuses_an_empty_ref_and_never_matches_digits_against_titles(self):
        self.pricing()
        self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Phase 2"'))
        r = run("mark", str(self.initiative), "2", "done")
        self.assertEqual(r.returncode, 1, r.stdout)
        self.assertIn("no ticket matches", r.stderr)
        for ref in ("", " "):
            r = run("mark", str(self.epic), ref, "done")
            self.assertEqual(r.returncode, 1, r.stdout)
            self.assertIn("empty", r.stderr)
        self.assertEqual(sorted(p.name for p in self.initiative.rglob("*-plan.md")), [])

    @unittest.skipIf(sys.platform == "win32", "Windows passes arguments as Unicode, so none is undecodable")
    def test_mark_with_an_undecodable_value_leaves_the_plan_unchanged(self):
        self.breakdown_epic()
        path = tk(self.epic) / "story-scaffold-plan.md"
        self.add(path.name, plan(1, "in-progress"))
        before = path.read_bytes()
        r = subprocess.run(
            [sys.executable, str(SCRIPT), "mark", str(self.epic), "1", "blocked", "--blocked", b"\xff"],
            capture_output=True,
            check=False,
        )
        self.assertNotEqual(r.returncode, 0, r.stdout)
        self.assertEqual(path.read_bytes(), before)

    def test_mark_refuses_a_plan_name_taken_by_another_file(self):
        self.breakdown_epic()
        for text in (plan(3), "# notes, not a plan\n"):
            self.add("story-ui-shell-plan.md", text)
            r = run("mark", str(self.epic), "2", "done")
            self.assertEqual(r.returncode, 1, r.stdout)
            self.assertIn("story-ui-shell-plan.md", json.loads(r.stderr)["error"])
            self.assertEqual((tk(self.epic) / "story-ui-shell-plan.md").read_text(encoding="utf-8"), text)

    def test_project_root_flag_finds_the_store_for_tickets_outside_the_project(self):
        self.write_store("jira")
        outside = tempfile.TemporaryDirectory()
        self.addCleanup(outside.cleanup)
        folder = Path(outside.name) / "epic-cart"
        folder.mkdir()
        (folder / "story-scaffold.md").write_text(ticket("", 1))
        self.assertEqual(json.loads(run("next", str(folder)).stdout)["store"], "repo")
        r = run("--project-root", str(self.root), "next", str(folder))
        self.assertEqual(r.returncode, 2)
        r = run("--project-root", str(self.root), "mark", str(folder), "1", "done")
        self.assertEqual(r.returncode, 2)

    def test_mark_refuses_on_tracker_store(self):
        self.write_store("jira")
        self.seed()
        r = run("mark", str(self.epic), "1", "done")
        self.assertEqual(r.returncode, 2)
        self.assertIn("write verb", r.stderr)

    def test_next_on_tracker_store_needs_synced_flag(self):
        self.write_store("linear")
        self.seed()
        r = run("next", str(self.epic))
        self.assertEqual(r.returncode, 2)
        self.assertIn("sync", r.stderr)
        self.assertEqual(self.next("--synced")["store"], "linear")

    def test_cycle_unknown_prerequisite_and_bad_status_are_errors(self):
        self.seed()
        self.add("story-scaffold.md", ticket("draft", 1, after="[3]"))
        r = run("next", str(self.epic))
        self.assertEqual(r.returncode, 1)
        self.assertIn("cycle", r.stderr)
        self.add("story-scaffold.md", ticket("draft", 1, after="[9]"))
        r = run("next", str(self.epic))
        self.assertEqual(r.returncode, 1)
        self.assertIn("names no entry in epic-cart", r.stderr)
        self.add("story-scaffold.md", ticket("draft", 1, after='["9"]'))
        r = run("next", str(self.epic))
        self.assertEqual(r.returncode, 1)
        self.assertIn("matches no ticket", r.stderr)
        self.add("story-scaffold.md", ticket("backlog", 1))
        r = run("next", str(self.epic))
        self.assertEqual(r.returncode, 1)
        self.assertIn("status 'backlog'", r.stderr)
        self.add("story-scaffold.md", ticket("", 1))
        (self.epic / "epic-cart.md").write_text("---\ntype: epic\nstatus: backlog\n---\n")
        r = run("next", str(self.epic))
        self.assertEqual(r.returncode, 1)
        self.assertIn("in-progress, done, dropped", r.stderr)

    def test_entries_with_no_files_start_without_a_pull(self):
        self.breakdown_epic()
        self.add("old-spec.md", "---\ntype: 'feature'\nstatus: 'done'\nsteps:\n  - a\n---\n")
        out = self.next()
        self.assertNotIn("to_pull", out)
        self.assertEqual([(e["id"], e["state"], e["file"]) for e in out["ready_to_start"]], [(1, "planned", None)])
        self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Scaffold"\nrefine = true'))
        out = self.next()
        self.assertEqual([e["id"] for e in out["ready_to_refine"]], [1])
        self.assertEqual(out["ready_to_start"], [])

    def test_a_plan_sets_the_state_with_or_without_a_story_file(self):
        self.breakdown_epic()
        self.add("story-scaffold-plan.md", plan(1, "done"))
        self.add("story-ui-shell-plan.md", plan(2, "in-review", assignee="ann"))
        for _ in range(2):
            out = self.next()
            self.assertEqual([(e["id"], e["state"], e["assignee"]) for e in out["in_progress"]], [(2, "review", "ann")])
            self.assertEqual([e["id"] for e in out["ready_to_start"]], [3])
            rows = self.status_rows(self.epic)
            self.assertEqual([r["state"] for r in rows], ["done", "review", "planned", "planned"])
            self.add("story-ui-shell.md", ticket("draft", 2, after="[1]", refined="true", assignee='"bob"'))
        self.assertEqual(out["in_progress"][0]["file"], "story-ui-shell.md")
        self.assertEqual(out["in_progress"][0]["status"], "in-review")
        self.add("story-ui-shell.md", ticket("draft", 2, after="[1]", tracker_status="done"))
        self.assertEqual(json.loads(run("status", str(self.epic)).stdout)["tickets"][1]["state"], "done")

    def test_a_built_plan_reads_as_review_and_mark_accepts_built(self):
        self.breakdown_epic()
        self.add("story-scaffold-plan.md", plan(1, "built"))
        out = self.next()
        self.assertEqual([(e["id"], e["status"], e["state"]) for e in out["in_progress"]], [(1, "built", "review")])
        self.assertEqual([e["id"] for e in out["ready_to_start"]], [2, 3])
        self.assertEqual(self.mark(str(self.epic), "2", "built")["status"], "built")
        self.assertEqual(self.status_rows(self.epic)[1]["state"], "review")

    def test_a_plan_with_blocked_at_blocks_its_entry(self):
        self.breakdown_epic()
        self.add("story-scaffold-plan.md", plan(1, "in-progress", blocked_at="2026-09-05", blocked_reason="legal"))
        out = self.next()
        self.assertEqual([(e["id"], e["blocked_reason"]) for e in out["blocked"]][0], (1, "legal"))
        self.assertEqual(out["in_progress"], [])

    def test_a_leaf_with_blocked_at_and_no_plan_is_blocked_with_its_reason(self):
        self.add(
            "story-scaffold.md",
            ticket("in-progress", 1, blocked_at="2026-09-05", blocked_reason="legal"),
        )
        out = self.next()
        self.assertEqual([(e["file"], e["blocked_reason"]) for e in out["blocked"]], [("story-scaffold.md", "legal")])
        self.assertEqual(out["in_progress"], [])

    def test_a_plan_leaving_out_a_field_blanks_the_leafs_value(self):
        self.breakdown_epic()
        self.add("story-scaffold-plan.md", plan(1, "done"))
        self.add(
            "story-ui-shell.md",
            ticket("in-review", 2, after="[1]", blocked_at="2026-09-05", blocked_reason="legal"),
        )
        self.add("story-ui-shell-plan.md", plan(2))
        self.assertNotIn(2, [e["id"] for e in self.next()["blocked"]])
        row = self.status_rows(self.epic)[1]
        self.assertEqual((row["status"], row["state"], row["blocked_reason"]), ("", "backlog", ""))

    def test_a_quoted_numeric_ticket_joins_its_entry_and_a_doubled_quote_reads_as_one(self):
        self.breakdown_epic()
        self.add("story-ui-shell-plan.md", plan(2, "blocked", blocked_reason="can''t"))
        blocked = self.next()["blocked"]
        self.assertEqual([(e["id"], e["status"], e["blocked_reason"]) for e in blocked][0], (2, "blocked", "can't"))

    def test_a_backlog_plan_joins_its_leaf_by_stem_and_a_leaf_without_a_plan_keeps_its_status(self):
        backlog = self.add_backlog()
        self.add("bug-x.md", ticket("draft", kind="bug"), backlog)
        self.add("bug-x-plan.md", plan("bug-x", "in-progress"), backlog)
        self.add("bug-y.md", ticket("in-review", kind="bug"), backlog)
        rows = self.status_rows(backlog)
        self.assertEqual([(r["file"], r["state"]) for r in rows], [("bug-x.md", "in-progress"), ("bug-y.md", "review")])

    def test_two_plans_for_one_ticket_error(self):
        self.breakdown_epic()
        self.add("story-a-plan.md", plan(1))
        self.add("story-b-plan.md", plan(1))
        r = run("next", str(self.epic))
        self.assertEqual(r.returncode, 1, r.stdout)
        for name in ("story-a-plan.md", "story-b-plan.md"):
            self.assertIn(name, json.loads(r.stderr)["error"])

    def test_an_orphan_plan_is_a_problem_and_the_rest_of_the_tree_still_reads(self):
        self.breakdown_epic()
        for ref in (9, "story-nothing"):
            self.add("story-a-plan.md", plan(ref))
            for command in ("next", "status"):
                r = run(command, str(self.epic))
                self.assertEqual(r.returncode, 0, r.stderr)
                out = json.loads(r.stdout)
                self.assertEqual(len(out["problems"]), 1)
                self.assertIn("story-a-plan.md", out["problems"][0])
            self.assertEqual(self.files(self.next()["ready_to_start"]), [None])

    def test_a_plan_with_an_unknown_status_blocks_its_ticket_until_mark_fixes_it(self):
        self.breakdown_epic()
        self.add("story-scaffold-plan.md", plan(1, "already-satisfied"))
        out = self.next()
        self.assertEqual(out["ready_to_start"], [])
        blocked = {t["id"]: t for t in out["blocked"]}
        self.assertIn("already-satisfied", blocked[1]["blocked_reason"])
        self.assertIn("status 'already-satisfied'", out["problems"][0])
        self.assertEqual(run("mark", str(self.epic), "1", "done").returncode, 0)
        out = self.next()
        self.assertNotIn("problems", out)
        self.assertEqual([t["id"] for t in out["ready_to_start"]], [2, 3])

    def test_status_lists_a_declared_epic_before_its_folder_exists(self):
        (tk(self.initiative) / "tickets.toml").write_text(
            '[[epic]]\nid = 1\nslug = "epic-base"\n\n'
            '[[epic]]\nid = 2\nslug = "epic-cart"\nafter = [{ epic = 1, needs = "the scaffold" }]\n'
        )
        r = run("status", str(self.initiative))
        self.assertEqual(r.returncode, 0, r.stderr)
        base, cart = json.loads(r.stdout)["epics"]
        self.assertEqual((base["slug"], base["id"], base["folder"], base["after"]), ("epic-base", 1, None, []))
        self.assertFalse({"status", "gated_by", "blocks", "next_id"} & base.keys())
        self.assertEqual((cart["folder"], cart["status"], cart["gated_by"], cart["next_id"]), ("epic-cart", "", [], 1))
        self.assertEqual(cart["after"], [{"epic": "epic-base", "needs": "the scaffold"}])
        self.add_epic("epic-extra")
        rows = json.loads(run("status", str(self.initiative)).stdout)["epics"]
        self.assertEqual(
            [(e["slug"], e["folder"]) for e in rows],
            [("epic-base", None), ("epic-cart", "epic-cart"), ("epic-extra", "epic-extra")],
        )

    def test_an_after_naming_a_listed_epic_with_no_folder_says_so(self):
        (tk(self.initiative) / "tickets.toml").write_text(
            '[[epic]]\nid = 1\nslug = "epic-base"\n\n[[epic]]\nid = 2\nslug = "epic-cart"\n'
        )
        self.breakdown_epic('[[entry]]\nid = 1\ntype = "story"\ntitle = "Scaffold"\nafter = ["epic-base"]\n')
        r = run("status", str(self.initiative))
        self.assertEqual(r.returncode, 1)
        self.assertIn("has no folder yet; create tickets/epic-base/epic-base.md", json.loads(r.stderr)["error"])
        self.add_epic("epic-base")
        self.assertEqual(run("status", str(self.initiative)).returncode, 0)

    def test_mirror_lists_a_published_container_apart_from_unmatched_and_writes_nothing_to_it(self):
        self.write_store("jira")
        self.breakdown_epic()
        epic_file = self.epic / "epic-cart.md"
        epic_file.write_text('---\ntype: epic\ntracker_id: "CART-E"\nstatus: in-progress\n---\n# epic-cart\n')
        (self.initiative / "initiative-checkout.md").write_text(
            '---\ntype: initiative\ntracker_id: "INIT-1"\n---\n# Checkout\n'
        )
        before = epic_file.read_text(encoding="utf-8")
        r = self.mirror(
            [
                {"ref": "epic-cart", "tracker_status": "done"},
                {"tracker_id": "INIT-1", "tracker_status": "in-progress"},
                {"ref": 1, "tracker_status": "done"},
                {"tracker_id": "NOPE-1", "tracker_status": "done"},
            ]
        )
        self.assertEqual(r.returncode, 0, r.stderr)
        out = json.loads(r.stdout)
        self.assertEqual(
            out["containers"],
            [
                {"ref": "epic-cart", "folder": "epic-cart", "status": "in-progress", "tracker_status": "done"},
                {"ref": "INIT-1", "folder": "initiative-checkout", "status": "", "tracker_status": "in-progress"},
            ],
        )
        self.assertEqual([m["ref"] for m in out["mirrored"]], ["1"])
        self.assertEqual([u["ref"] for u in out["unmatched"]], ["NOPE-1"])
        self.assertEqual(epic_file.read_text(encoding="utf-8"), before)
        r = self.mirror([{"tracker_id": "cart-e", "tracker_status": "done"}])
        self.assertEqual([c["folder"] for c in json.loads(r.stdout)["containers"]], ["epic-cart"])

    def test_move_takes_the_entry_its_file_and_plan_along_and_rewrites_every_after(self):
        pricing = self.pricing()
        self.breakdown_epic()
        (tk(pricing) / "tickets.toml").write_text(
            '[[entry]]\nid = 1\ntype = "story"\ntitle = "Pricing contract"\n\n'
            '[[entry]]\nid = 2\ntype = "story"\ntitle = "Pricing rules"\nafter = [1, "2.2"] # waits on the shell\n'
        )
        self.assertEqual(run("pull", str(self.epic), "2").returncode, 0)
        self.mark(str(self.epic), "2", "in-progress")
        r = run("move", str(self.epic), "2", str(pricing))
        self.assertEqual(r.returncode, 0, r.stderr)
        out = json.loads(r.stdout)
        self.assertEqual(
            [out[k] for k in ("from", "to", "old_id", "id", "ref", "file", "plan", "next_id")],
            ["epic-cart", "epic-pricing", 2, 3, "1.3", "story-ui-shell.md", "story-ui-shell-plan.md", {"epic-cart": 5}],
        )
        self.assertEqual(
            {r["where"]: r["after"] for r in out["rewritten"]},
            {
                "epic-cart/tickets/tickets.toml entry 4": ["1.3", 3],
                "epic-pricing/tickets/tickets.toml entry 2": [1, 3],
                "epic-pricing/tickets/tickets.toml entry 3": ["2.1"],
                "epic-pricing/tickets/story-ui-shell.md": ["2.1"],
            },
        )
        cart = (tk(self.epic) / "tickets.toml").read_text(encoding="utf-8")
        self.assertTrue(cart.startswith("next_id = 5\n\n[[entry]]\nid = 1\n"), cart)
        self.assertNotIn("UI shell", cart)
        self.assertIn('after = ["1.3", 3]\n', cart)
        text = (tk(pricing) / "tickets.toml").read_text(encoding="utf-8")
        self.assertIn("after = [1, 3] # waits on the shell\n", text)
        self.assertTrue(
            text.endswith('[[entry]]\nid = 3\ntype = "story"\ntitle = "UI shell"\nafter = ["2.1"]\ncovers = ["R1"]\n'),
            text,
        )
        self.assertEqual(sorted(p.name for p in tk(self.epic).glob("story-*")), [])
        moved = (tk(pricing) / "story-ui-shell.md").read_text(encoding="utf-8")
        head = moved.split("---")[1]
        for line in ("id: 3", "parent: epic-pricing", 'after: ["2.1"]'):
            self.assertIn(f"\n{line}\n", head)
        self.assertIn("\n- parent — out/initiative-checkout/tickets/epic-pricing/epic-pricing.md\n", moved)
        self.assertNotIn("epic-cart", moved)
        self.assertIn("\nticket: 3\n", (tk(pricing) / "story-ui-shell-plan.md").read_text(encoding="utf-8"))
        status = json.loads(run("status", str(self.initiative)).stdout)
        self.assertNotIn("problems", status)
        self.assertEqual({e["slug"]: e["next_id"] for e in status["epics"]}, {"epic-pricing": 4, "epic-cart": 5})
        hit = json.loads(run("find", str(self.initiative), "1.3").stdout)
        self.assertEqual((hit["title"], hit["state"], hit["after"]), ("UI shell", "in-progress", ["2.1"]))

    def test_move_to_another_initiative_leaves_its_own_cross_refs_alone(self):
        pricing = self.pricing()
        (tk(pricing) / "tickets.toml").write_text(
            '[[entry]]\nid = 1\ntype = "story"\ntitle = "Pricing contract"\n\n'
            '[[entry]]\nid = 2\ntype = "story"\ntitle = "Pricing rules"\n'
        )
        payments = self.root / "out" / "initiative-payments"
        payments.mkdir()
        (payments / "initiative-payments.md").write_text("---\ntype: initiative\n---\n# Payments\n")
        ledger = payments / "tickets" / "epic-ledger"
        settle = payments / "tickets" / "epic-settle"
        for folder in (ledger, settle):
            folder.mkdir(parents=True)
            (folder / f"{folder.name}.md").write_text(f"---\ntype: epic\nafter: []\n---\n# {folder.name}\n")
        (tk(payments) / "tickets.toml").write_text(
            '[[epic]]\nid = 1\nslug = "epic-ledger"\n\n[[epic]]\nid = 2\nslug = "epic-settle"\n'
        )
        (tk(ledger) / "tickets.toml").write_text(
            '[[entry]]\nid = 1\ntype = "story"\ntitle = "Accounts"\n\n[[entry]]\nid = 2\ntype = "story"\ntitle = "Postings"\n'
        )
        (tk(settle) / "tickets.toml").write_text(
            '[[entry]]\nid = 1\ntype = "story"\ntitle = "Payouts"\nafter = ["1.2"]\n'
        )
        r = run("move", str(pricing), "2", str(settle))
        self.assertEqual(r.returncode, 0, r.stderr)
        self.assertIn('title = "Payouts"\nafter = ["1.2"]\n', (tk(settle) / "tickets.toml").read_text(encoding="utf-8"))

    def test_move_creates_the_destinations_breakdown_and_leaves_what_status_then_reports(self):
        self.pricing()
        self.breakdown_epic("next_id = 6\n" + self.BREAKDOWN)
        tax = self.add_epic("epic-tax")
        (tk(self.initiative) / "tickets.toml").write_text(
            '[[epic]]\nid = 1\nslug = "epic-pricing"\n\n[[epic]]\nid = 2\nslug = "epic-cart"\n\n[[epic]]\nid = 3\nslug = "epic-tax"\n'
        )
        r = run("move", str(self.initiative), "2.3", str(tax))
        self.assertEqual(r.returncode, 0, r.stderr)
        out = json.loads(r.stdout)
        self.assertEqual(
            [out[k] for k in ("id", "ref", "file", "plan", "next_id")], [1, "3.1", None, None, {"epic-cart": 6}]
        )
        self.assertEqual(
            (tk(tax) / "tickets.toml").read_text(encoding="utf-8"),
            '[[entry]]\nid = 1\ntype = "spike"\ntitle = "Tax engine?"\nafter = ["2.1"]\ncovers = ["R4"]\nhitl = true\n',
        )
        cart = (tk(self.epic) / "tickets.toml").read_text(encoding="utf-8")
        self.assertTrue(cart.startswith("next_id = 6\n\n[[entry]]"), cart)
        self.assertNotIn("Tax engine", cart)
        self.assertIn('after = [2, "3.1"]\n', cart)
        status = json.loads(run("status", str(self.initiative)).stdout)
        self.assertEqual(
            {e["slug"]: e["next_id"] for e in status["epics"]}, {"epic-pricing": 3, "epic-cart": 6, "epic-tax": 2}
        )
        self.assertEqual(
            status["undeclared_after"],
            [
                {"epic": "epic-cart", "after": "epic-tax", "ref": "2.4", "names": "3.1"},
                {"epic": "epic-tax", "after": "epic-cart", "ref": "3.1", "names": "2.1"},
            ],
        )

    def test_a_tickets_folder_given_as_the_dir_is_read_as_its_container(self):
        self.pricing()
        self.breakdown_epic()
        for folder in (self.epic, self.initiative):
            expected = json.loads(run("status", str(folder)).stdout)
            self.assertEqual(json.loads(run("status", str(folder / "tickets")).stdout), expected)
        self.assertEqual(json.loads(run("next", str(self.epic / "tickets")).stdout)["folder"], "epic-cart")
        self.assertEqual(json.loads(run("find", str(self.epic / "tickets"), "2").stdout)["title"], "UI shell")

    def test_the_earlier_layout_is_refused_and_bmad_migrate_named(self):
        old = self.root / "out" / "initiative-old"
        (old / "epic-a").mkdir(parents=True)
        (old / "initiative-old.md").write_text("---\ntype: initiative\n---\n# Old\n")
        (old / "epic-a" / "epic-a.md").write_text("---\ntype: epic\n---\n# A\n")
        (old / "epic-a" / "story-a.md").write_text(ticket("", 1))
        self.assertEqual(json.loads(run("status", str(old)).stderr)["error"].split(" uses")[0], "initiative-old")
        (old / "tickets").mkdir()
        (old / "epic-a").rename(old / "tickets" / "epic-a")
        (self.initiative / "tickets.toml").write_text("")
        for folder, name in ((old, "epic-a"), (self.epic, "initiative-checkout")):
            r = run("status", str(folder))
            self.assertEqual(r.returncode, 1, r.stdout)
            error = json.loads(r.stderr)["error"]
            self.assertTrue(error.startswith(f"{name} uses the earlier layout"), error)
            self.assertIn("`bmad migrate` moves it", error)

    def test_epic_rows_carry_refined_false_only_when_the_epic_file_says_so(self):
        self.pricing()
        (self.epic / "epic-cart.md").write_text("---\ntype: epic\nrefined: false\n---\n# epic-cart\n")
        (self.add_epic("epic-tax") / "epic-tax.md").write_text("---\ntype: epic\nrefined: true\n---\n# epic-tax\n")
        epics = json.loads(run("status", str(self.initiative)).stdout)["epics"]
        self.assertEqual(
            [(e["slug"], e["refined"]) for e in epics],
            [("epic-pricing", True), ("epic-cart", False), ("epic-tax", True)],
        )

    def test_a_breakdown_holds_one_kind_of_table(self):
        (tk(self.initiative) / "tickets.toml").write_text(
            '[[epic]]\nid = 1\nslug = "epic-cart"\n\n[[entry]]\nid = 1\ntype = "story"\ntitle = "Stray"\n'
        )
        r = run("status", str(self.initiative))
        self.assertEqual(r.returncode, 1, r.stdout)
        self.assertEqual(
            json.loads(r.stderr)["error"],
            "initiative-checkout/tickets/tickets.toml: a breakdown holds one kind of table, not [[epic]] and [[entry]]",
        )

    def test_a_parent_initiative_lists_its_initiatives_with_what_each_waits_on(self):
        self.seed(s1="done")
        parent = self.shop()
        out = json.loads(run("status", str(parent)).stdout)
        self.assertEqual((out["folder"], out["store"], out["order_conflict"]), ("initiative-shop", "repo", []))
        checkout, payments, refunds = out["initiatives"]
        self.assertEqual(
            checkout,
            {
                "id": 1,
                "slug": "initiative-checkout",
                "title": "Checkout",
                "after": [],
                "folder": "initiative-checkout",
                "status": "",
                "counts": {"total": 5, "done": 1, "backlog": 4},
                "waiting_on": [],
            },
        )
        self.assertEqual(
            payments,
            {
                "id": 2,
                "slug": "initiative-payments",
                "title": "Payments",
                "after": [{"initiative": "initiative-checkout", "needs": "the cart total"}],
                "folder": None,
                "waiting_on": ["initiative-checkout"],
            },
        )
        self.assertEqual(refunds["waiting_on"], ["initiative-payments", "initiative-checkout"])
        nxt = json.loads(run("next", str(parent)).stdout)
        self.assertEqual((nxt["initiatives"], nxt["ready"]), (out["initiatives"], ["initiative-checkout"]))
        (self.initiative / "initiative-checkout.md").write_text(
            "---\ntype: initiative\nstatus: done\n---\n# Checkout\n"
        )
        nxt = json.loads(run("next", str(parent)).stdout)
        self.assertEqual(nxt["ready"], ["initiative-payments"])
        self.assertEqual([r["waiting_on"] for r in nxt["initiatives"]], [[], [], ["initiative-payments"]])

    def test_a_parent_reports_an_order_conflict_and_refuses_a_cycle(self):
        text = (
            '[[initiative]]\nid = 1\nslug = "initiative-checkout"\nafter = [{ initiative = 2 }]\n\n'
            '[[initiative]]\nid = 2\nslug = "initiative-payments"\n'
        )
        parent = self.shop(text)
        out = json.loads(run("next", str(parent)).stdout)
        self.assertEqual(out["order_conflict"], [{"initiative": "initiative-checkout", "after": "initiative-payments"}])
        self.assertEqual(out["ready"], ["initiative-payments"])
        self.shop(text + "after = [{ initiative = 1 }]\n")
        for command in ("status", "next"):
            r = run(command, str(parent))
            self.assertEqual(r.returncode, 1, r.stdout)
            self.assertEqual(
                json.loads(r.stderr)["error"],
                "cycle through initiative-checkout -> initiative-payments -> initiative-checkout",
            )

    def test_find_pull_mark_and_move_on_a_parent_ask_for_one_of_its_initiatives(self):
        self.breakdown_epic()
        parent = self.shop()
        for args in (
            ("find", str(parent), "1"),
            ("pull", str(parent), "1"),
            ("mark", str(parent), "1", "done"),
            ("move", str(parent), "1", str(self.epic)),
            ("move", str(self.epic), "1", str(parent)),
        ):
            r = run(*args)
            self.assertEqual(r.returncode, 1, args)
            self.assertEqual(
                json.loads(r.stderr)["error"],
                "initiative-shop: a parent initiative holds initiatives; name one of them: "
                "initiative-checkout, initiative-payments, initiative-refunds",
            )

    def test_move_refuses_and_writes_nothing_when_it_cannot_finish(self):
        pricing = self.pricing()
        self.breakdown_epic()
        tax = self.add_epic("epic-tax")
        self.add("story-stray.md", ticket("", 9))
        self.assertEqual(run("pull", str(self.epic), "3").returncode, 0)
        self.add("spike-tax-engine.md", "# taken\n", pricing)
        before = {p: p.read_bytes() for p in self.initiative.rglob("*") if p.is_file()}
        for args, message in (
            ((str(self.epic), "2", str(self.epic)), "already in epic-cart"),
            ((str(self.epic), "2", str(self.initiative)), "holds epics"),
            ((str(self.epic), "9", str(pricing)), "has no entry"),
            ((str(self.epic), "2", str(tax)), "has no id in the initiative's tickets.toml"),
            ((str(self.epic), "3", str(pricing)), "spike-tax-engine.md exists already"),
            ((str(self.epic), "2", "epic-nowhere"), "not a folder"),
        ):
            r = run("move", *args)
            self.assertEqual(r.returncode, 1, r.stdout)
            self.assertIn(message, json.loads(r.stderr)["error"])
        self.assertEqual({p: p.read_bytes() for p in self.initiative.rglob("*") if p.is_file()}, before)


class ActiveInitiativeTests(TreeCase):
    """With no folder, next, status, and find run on `{tickets.root}/{active_initiative}`."""

    def setUp(self):
        super().setUp()
        (self.root / "_bmad" / "scripts").mkdir()
        shutil.copy(CONFIG_UTILS, self.root / "_bmad" / "scripts" / "config_utils.py")
        self.configure("initiative-checkout")
        self.seed()

    def configure(self, initiative, layer="config.toml"):
        line = f'active_initiative = "{initiative}"\n' if initiative is not None else ""
        path = self.root / "_bmad" / ("custom" if layer != "config.toml" else "") / layer
        path.write_text(f'[core]\noutput_folder = "{{project-root}}/out"\n{line}')

    def elsewhere(self):
        other = tempfile.TemporaryDirectory()
        self.addCleanup(other.cleanup)
        return other.name

    def ok(self, r):
        self.assertEqual(r.returncode, 0, r.stderr)
        return json.loads(r.stdout)

    def fails(self, r, message):
        self.assertEqual(r.returncode, 1, r.stdout)
        self.assertIn(message, json.loads(r.stderr)["error"])

    def test_next_with_no_folder_runs_on_the_active_initiative(self):
        expected = self.ok(run("next", str(self.initiative)))
        self.assertEqual(self.ok(run("next", cwd=self.epic)), expected)
        self.assertEqual(expected["folder"], "initiative-checkout")

    def test_with_no_folder_next_and_status_also_report_the_backlog(self):
        self.assertNotIn("backlog", self.ok(run("next", cwd=self.root)))
        backlog = self.add_backlog()
        self.add("bug-x.md", ticket("ready-for-dev", kind="bug", refined="true"), backlog)
        out = self.ok(run("next", cwd=self.root))
        self.assertEqual(out["folder"], "initiative-checkout")
        self.assertEqual(
            (out["backlog"]["folder"], out["backlog"]["ready_to_start"][0]["ref"]), ("backlog", "bug-x.md")
        )
        self.assertEqual(self.ok(run("find", "backlog", "bug-x.md", cwd=self.root))["file"], "bug-x.md")
        out = self.ok(run("status", cwd=self.root))
        self.assertEqual((out["counts"]["total"], out["backlog"]["counts"]), (5, {"total": 1, "backlog": 1}))
        self.assertNotIn("backlog", self.ok(run("next", str(self.initiative))))
        self.add("bug-y.md", ticket("nonsense", kind="bug"), backlog)
        out = self.ok(run("next", cwd=self.root))
        self.assertIn("status", out["backlog"]["error"])
        self.assertEqual(len(out["ready_to_refine"]), 2)

    def test_with_no_folder_a_ticket_the_initiative_lacks_is_found_in_the_backlog(self):
        backlog = self.add_backlog()
        self.add("bug-x.md", ticket("draft", 7, kind="bug", tracker_id='"SHOP-7"'), backlog)
        hit = self.ok(run("find", "bug-x.md", cwd=self.root))
        self.assertEqual((hit["folder"], hit["plan"]), ("backlog", str(backlog.resolve() / "bug-x-plan.md")))
        self.assertEqual(self.ok(run("find", "SHOP-7", cwd=self.root))["ref"], "bug-x.md")
        self.fails(run("find", "bug-nope", cwd=self.root), "no ticket matches")
        self.fails(run("find", str(self.initiative), "bug-x.md"), "no ticket matches")
        self.ok(run("mark", "bug-x.md", "done", cwd=self.root))
        self.assertTrue((backlog / "bug-x-plan.md").is_file())
        self.write_store("jira")
        r = run("mirror", cwd=self.root, stdin=json.dumps([{"tracker_id": "SHOP-7", "tracker_status": "done"}]))
        self.assertEqual([m["ref"] for m in self.ok(r)["mirrored"]], ["bug-x.md"])
        self.assertIn("\ntracker_status: done\n", (backlog / "bug-x.md").read_text(encoding="utf-8"))

    def test_a_backlog_that_cannot_be_read_is_reported_and_the_initiative_still_answers(self):
        backlog = self.add_backlog()
        (backlog / "bug-x.md").write_bytes(b"---\ntype: bug\ntitle: \xff\n---\n")
        out = self.ok(run("status", cwd=self.root))
        self.assertEqual((out["counts"]["total"], out["backlog"]["folder"]), (5, "backlog"))
        self.assertIn("error", out["backlog"])
        self.fails(run("find", "bug-x.md", cwd=self.root), "no ticket matches")
        self.write_store("jira")
        r = run("mirror", cwd=self.root, stdin=json.dumps([{"ref": "story-scaffold", "tracker_status": "done"}]))
        self.assertEqual([m["file"] for m in self.ok(r)["mirrored"]], ["story-scaffold.md"])

    def test_an_epic_is_found_by_its_folder_name_under_the_active_initiative(self):
        out = self.ok(run("--project-root", str(self.root), "status", "epic-cart", cwd=self.elsewhere()))
        self.assertEqual((out["folder"], out["counts"]["total"]), ("epic-cart", 5))
        self.configure(None)
        self.fails(run("--project-root", str(self.root), "status", "epic-cart", cwd=self.elsewhere()), "not a folder")

    def test_a_folder_not_found_names_where_it_looked(self):
        r = run("--project-root", str(self.root), "status", "epic-nope", cwd=self.elsewhere())
        self.assertEqual(r.returncode, 1)
        error = json.loads(r.stderr)["error"]
        self.assertTrue(error.startswith("not a folder: epic-nope (looked under "), error)
        for base in (self.root / "out", self.initiative, self.root):
            self.assertIn(str(base.resolve()), error)

    def test_the_backlog_as_active_initiative_is_reported_once(self):
        backlog = self.add_backlog()
        self.add("bug-x.md", ticket("draft", kind="bug"), backlog)
        self.configure("backlog")
        out = self.ok(run("status", cwd=self.root))
        self.assertEqual(out["folder"], "backlog")
        self.assertNotIn("backlog", out)

    def test_project_root_flag_names_the_project_from_anywhere(self):
        self.write_store(root="out")
        out = self.ok(run("--project-root", str(self.root), "status", cwd=self.elsewhere()))
        self.assertEqual(out["folder"], "initiative-checkout")
        self.assertEqual(out["counts"]["total"], 5)

    def test_a_relative_folder_is_found_under_the_store_from_anywhere(self):
        out = self.ok(run("--project-root", str(self.root), "status", "initiative-checkout", cwd=self.elsewhere()))
        self.assertEqual(out["folder"], "initiative-checkout")
        out = self.ok(
            run(
                "--project-root",
                str(self.root),
                "next",
                "out/initiative-checkout/tickets/epic-cart",
                cwd=self.elsewhere(),
            )
        )
        self.assertEqual(out["folder"], "epic-cart")
        self.fails(
            run("--project-root", str(self.root), "status", "initiative-gone", cwd=self.elsewhere()), "not a folder"
        )

    def test_root_substitutes_project_root_and_output_folder(self):
        for root in ("{project-root}/out", "{output_folder}"):
            self.write_store(root=root)
            self.assertEqual(self.ok(run("status", cwd=self.root))["folder"], "initiative-checkout", root)

    def test_find_by_ref_only(self):
        self.pricing()
        expected = self.ok(run("find", str(self.initiative), "1.2"))
        self.assertEqual(self.ok(run("find", "1.2", cwd=self.root)), expected)
        self.assertEqual(expected["title"], "Pricing rules")

    def test_a_given_folder_never_reads_the_config(self):
        (self.root / "_bmad" / "config.toml").write_text("not toml [")
        self.assertEqual(self.ok(run("next", str(self.epic), cwd=self.root))["folder"], "epic-cart")
        self.fails(run("next", cwd=self.root), "config.toml")

    def test_unset_active_initiative_names_the_key(self):
        for value in (None, ""):
            self.configure(value)
            self.fails(run("next", cwd=self.root), "core.active_initiative")

    def test_no_project_root_found(self):
        self.fails(run("status", cwd=self.elsewhere()), "no project root")

    def test_missing_initiative_folder_names_the_resolved_path(self):
        self.configure("initiative-gone")
        self.fails(run("next", cwd=self.root), str((self.root / "out" / "initiative-gone").resolve()))

    def test_store_is_read_from_the_project_found_when_root_lies_outside_it(self):
        outside = Path(self.elsewhere())
        shutil.copytree(self.initiative, outside / "initiative-checkout")
        self.write_store("linear", outside.as_posix())
        r = run("next", cwd=self.root)
        self.assertEqual(r.returncode, 2, r.stderr)
        self.assertIn("sync ticket status", r.stderr)
        self.assertEqual(self.ok(run("status", cwd=self.root))["store"], "linear")

    def test_the_skill_installs_the_script_and_a_lone_copy_runs(self):
        bmod = tomllib.loads((SCRIPT.parents[1] / "bmod.toml").read_text(encoding="utf-8"))
        self.assertIn("scripts/tickets.py", bmod["skill"]["scripts"])
        installed = self.root / "_bmad" / "method" / "scripts"
        installed.mkdir(parents=True)
        shutil.copy(SCRIPT, installed)
        r = subprocess.run(
            [sys.executable, str(installed / "tickets.py"), "next"],
            text=True,
            capture_output=True,
            check=False,
            cwd=self.root,
        )
        self.assertEqual(self.ok(r)["folder"], "initiative-checkout")

    def test_mark_with_no_folder_blocks_a_ticket_in_the_active_initiative(self):
        pricing = self.pricing()
        r = run("mark", "1.2", "in-progress", "--blocked", "legal", cwd=self.root)
        self.assertEqual(self.ok(r)["plan"], str(tk(pricing).resolve() / "story-pricing-rules-plan.md"))
        text = (tk(pricing) / "story-pricing-rules-plan.md").read_text(encoding="utf-8")
        self.assertIn(f'\nblocked_at: "{date.today().isoformat()}"\nblocked_reason: "legal"\n', text)
        blocked = self.ok(run("next", cwd=self.root))["blocked"]
        self.assertIn(
            ("epic-pricing", 2, "in-progress", "legal"),
            [(e["epic"], e["id"], e["status"], e["blocked_reason"]) for e in blocked],
        )

    def test_a_parent_as_the_active_initiative_shows_its_initiatives_and_find_wants_one(self):
        self.shop()
        self.configure("initiative-shop")
        out = self.ok(run("status", cwd=self.root))
        self.assertEqual(
            [(r["slug"], r["folder"]) for r in out["initiatives"]][0], ("initiative-checkout", "initiative-checkout")
        )
        self.assertEqual(self.ok(run("next", cwd=self.root))["ready"], ["initiative-checkout"])
        self.fails(run("find", "1", cwd=self.root), "a parent initiative holds initiatives; name one of them")
        self.assertEqual(self.ok(run("find", "initiative-checkout", "story-scaffold", cwd=self.root))["id"], 1)

    def test_user_layer_overrides_the_base_config(self):
        self.configure("initiative-gone")
        self.configure("initiative-checkout", layer="config.user.toml")
        self.assertEqual(self.ok(run("next", cwd=self.root))["folder"], "initiative-checkout")


if __name__ == "__main__":
    unittest.main()

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# ///
"""tickets — read a ticket tree and answer what is next.

The tree's layout, the plan file, who writes each `status`, and how it becomes a board `state` are in
the repo's `tools/ticket-tree-rules.md`; the keys of an entry, an epic, and an initiative table, and the forms `after`
takes, are in the skill's `assets/tickets-template.toml`. Each command's `--help` says what it reads
and returns.

An initiative or epic folder keeps its own file at the top and its tickets in its `tickets/` folder:
the `tickets.toml`, an initiative's epic folders, and the leaf files and plans. A backlog folder has no
file of its own and holds its leaf files and plans directly. A tree with a `tickets.toml`, an epic
folder, or a leaf file beside the container's own file is in the earlier layout and is refused; `bmad
migrate` moves it. A parent initiative's `tickets.toml` lists `[[initiative]]` tables, its children,
which are initiative folders beside it; `status` and `next` report them, and the other commands want
one of the children.

What this script decides itself: a leaf file joins its entry through the `id` in its frontmatter (`6`
and `"6"` are one id); a plan joins through `ticket`, an entry id or a backlog file's stem, and is
never a row. A leaf file's `after` and `hitl` replace the entry's, absent reading as empty; a
difference is `drift` on the row. A prerequisite is met when it is done or in review, a dropped one
still blocks, and an epic file's own `after` waits for that epic to be done. A plan whose `ticket`
names nothing is skipped and one whose `status` is unknown blocks its ticket; both are listed under
`problems`. A moved entry takes the destination's next id and goes last there; the folder it left keeps
that id from coming back, and every `after` naming it is rewritten.

The project root is `--project-root`, else the first folder at or above the working directory that
holds `_bmad/`; `output_folder` and `active_initiative` come from the merged BMad config. A relative
`<dir>` that is not under the working directory is looked up under `{output_folder}`, then the
active initiative and its `tickets/` folder, then the project root. A `tickets/` folder given as
`<dir>` is read as the folder that holds it.

Output is one JSON object on stdout. Exit 0 on success, 1 on a malformed tree, 2 when the store
forbids the operation.
"""

import argparse
import codecs
import importlib.util
import json
import os
import re
import sys
import tomllib
import unicodedata
from datetime import date
from pathlib import Path

sys.dont_write_bytecode = True

STATUSES = ("draft", "ready-for-dev", "in-progress", "in-review", "built", "done", "blocked", "dropped")
STATES = ("backlog", "in-progress", "review", "done", "dropped")
CONTAINER_STATUSES = ("in-progress", "done", "dropped")
STATE_OF = {
    "": "backlog",
    "draft": "backlog",
    "ready-for-dev": "backlog",
    "in-progress": "in-progress",
    "blocked": "in-progress",
    "in-review": "review",
    "built": "review",
    "done": "done",
    "dropped": "dropped",
}
LEAF_TYPES = ("story", "spike", "bug")
CONTAINER_TYPES = ("initiative", "epic")
NAME_RE = re.compile(r"^(story|spike|bug)-(.+)\.md$")
ID_RE = re.compile(r"^[0-9A-Za-z]+$")
CROSS_RE = re.compile(r"^([0-9A-Za-z]+)\.([0-9A-Za-z]+)$")
EPIC_RE = re.compile(r"^epic-[^/]+$")
BREAKDOWN = "tickets.toml"
TICKETS = "tickets"
TABLES = ("initiative", "epic", "entry")
QUOTED_COMMENT_RE = re.compile(r"""^("(?:[^"\\]|\\.)*"|'(?:[^']|'')*')\s+#.*$""")
FRONTMATTER_RE = re.compile(r"\A---\n(.*?)\n---(?:\n|\Z)", re.S)
PLAN_FIELDS = ("status", "assignee", "blocked_at", "blocked_reason")
COMMENT_RE = re.compile(r"<!--.*?-->", re.S)  # a template's example sits in one
UNKNOWN_RE = re.compile(r"^[ \t]*(?:[-*][ \t]+)?Unknown:[ \t]*(\S.*)$", re.M)


class TicketError(Exception):
    """`data` is added to the error object the command prints."""

    def __init__(self, message: str, **data):
        super().__init__(message)
        self.data = data


class NoMatch(TicketError):
    pass


class StoreRefusal(Exception):
    pass


# ---------------------------------------------------------------- frontmatter


def parse_frontmatter(text: str, lenient: bool = False) -> dict:
    """Minimal YAML subset: `key: value`, lists as `[a, b]`, quoted or bare scalars. Lenient
    skips block lists instead of refusing them, for plans written from the build's template."""
    m = FRONTMATTER_RE.match(text)
    if not m:
        return {}
    data = {}
    for line in m.group(1).splitlines():
        if lenient and (line[:1].isspace() or line.startswith("- ")):
            continue
        if line.lstrip().startswith("- "):
            raise TicketError("frontmatter lists must be inline: `key: [a, b]`")
        if not line.strip() or line.lstrip().startswith("#") or ":" not in line:
            continue
        key, _, value = line.partition(":")
        value = value.split("   #")[0].strip()
        quoted = QUOTED_COMMENT_RE.match(value)
        data[key.strip()] = _scalar(quoted.group(1) if quoted else value)
    return data


def _scalar(value: str):
    if value.startswith("[") and value.endswith("]"):
        inner = value[1:-1].strip()
        return [] if not inner else [_scalar(v.strip()) for v in inner.split(",")]
    if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
        if value[0] == '"':
            try:
                return str(json.loads(value))
            except ValueError:
                pass
        return value[1:-1] if value[0] == '"' else value[1:-1].replace("''", "'")
    if value in ("true", "false"):
        return value == "true"
    if re.fullmatch(r"-?\d+", value):
        return int(value)
    return value


def set_frontmatter_value(text: str, key: str, value: str) -> str:
    m = FRONTMATTER_RE.match(text)
    if not m:
        raise TicketError("ticket has no frontmatter")
    block = m.group(1)
    pattern = re.compile(rf"^{re.escape(key)}:.*$\n?", re.M)
    if value == "":
        block = pattern.sub("", block).rstrip("\n")
    elif pattern.search(block):
        block = pattern.sub(lambda _: f"{key}: {value}\n", block, count=1).rstrip("\n")
    else:
        block = f"{block}\n{key}: {value}"
    return text[: m.start(1)] + block + text[m.end(1) :]


def _list(value, where: str) -> list:
    if value in (None, ""):
        return []
    if not isinstance(value, list):
        raise TicketError(f"{where}: after must be a list")
    return value


def _flag(value) -> bool:
    return str(value).lower() == "true"


def _one_of(value, allowed: tuple, where: str, field: str):
    """`value` when it is absent (`""`) or one of `allowed`; else the error naming them."""
    if value not in ("", *allowed):
        raise TicketError(f"{where}: {field} {value!r} is not one of {', '.join(allowed)}")
    return value


# ---------------------------------------------------------------- loading


def read_text(path: Path) -> str:
    # utf-8-sig: Windows editors can save a byte-order mark, which would hide the frontmatter.
    return path.read_text(encoding="utf-8-sig")


def tickets_dir(folder: Path) -> Path:
    """Where a folder's `tickets.toml`, epic folders, leaf files, and plans are: `<folder>/tickets/` for an
    initiative or an epic, the folder itself for a backlog, which has no file of its own."""
    if not (folder / f"{folder.name}.md").is_file() and not (folder / TICKETS).is_dir():
        return folder
    if (folder / BREAKDOWN).is_file() or any(
        NAME_RE.match(p.name) or (EPIC_RE.match(p.name) and p.is_dir()) for p in folder.iterdir()
    ):
        raise TicketError(
            f"{folder.name} uses the earlier layout, its tickets beside {folder.name}.md instead of in "
            f"{folder.name}/{TICKETS}/; `bmad migrate` moves it"
        )
    return folder / TICKETS


def shown(folder: Path) -> str:
    """A folder's tickets as messages name them: `epic-cart/tickets`, or `backlog`."""
    return f"{folder.name}/{TICKETS}" if tickets_dir(folder) != folder else folder.name


def container_of(folder: Path) -> Path:
    """A `tickets/` folder given as `<dir>` stands for the folder that holds it."""
    return folder.parent if folder.name == TICKETS else folder


def load_breakdown(folder: Path) -> dict:
    path = tickets_dir(folder) / BREAKDOWN
    if not path.is_file():
        return {}
    where = f"{shown(folder)}/{BREAKDOWN}"
    try:
        data = tomllib.loads(read_text(path))
    except tomllib.TOMLDecodeError as e:
        raise TicketError(f"{where}: {e}") from e
    floor = data.get("next_id", 0)
    if isinstance(floor, bool) or not isinstance(floor, int):
        raise TicketError(f"{where}: `next_id` must be a whole number")
    kinds = [f"[[{t}]]" for t in TABLES if t in data]
    if len(kinds) > 1:
        raise TicketError(f"{where}: a breakdown holds one kind of table, not {' and '.join(kinds)}")
    for table in TABLES:
        rows = data.get(table, [])
        if not isinstance(rows, list) or not all(isinstance(r, dict) for r in rows):
            raise TicketError(f"{where}: write `[[{table}]]` tables, one per {table}")
        for r in rows:
            for key in ("covers", "after", "references", "notes"):
                if not isinstance(r.get(key, []), list):
                    raise TicketError(f"{where}: `{key}` must be a list")
            r["id"] = _id(r.get("id"))
            if r["id"] is None:
                raise TicketError(f"{where}: every {table} needs an `id`: a number, or letters and digits")
            if table != "entry":
                if not isinstance(r.get("slug"), str) or not r["slug"]:
                    raise TicketError(f"{where}: {table} {r['id']} needs a `slug`")
                for a in r.get("after", []):
                    if not isinstance(a, dict) or (_id(a.get(table)) is None and not isinstance(a.get(table), str)):
                        raise TicketError(
                            f'{where}: an {table}\'s `after` takes tables: [{{ {table} = <id or slug>, needs = "..." }}]'
                        )
    return data


def _id(value):
    """An id as the tree compares it, else None: a number, or letters and digits; digits alone are the number."""
    if isinstance(value, bool):
        return None
    if isinstance(value, int):
        return value
    # A frontmatter value of true or false is read as a boolean, so neither can be an id.
    if isinstance(value, str) and ID_RE.match(value) and value not in ("true", "false"):
        return int(value) if value.isdigit() else value
    return None


def load_container(folder: Path) -> dict:
    path = folder / f"{folder.name}.md"
    if not path.is_file():
        raise TicketError(f"{folder.name}: no {path.name}")
    fm = parse_frontmatter(read_text(path))
    if fm.get("type") not in CONTAINER_TYPES:
        raise TicketError(
            f"{folder.name}/{path.name}: type {fm.get('type')!r} is not one of {', '.join(CONTAINER_TYPES)}"
        )
    status = _one_of(fm.get("status", ""), CONTAINER_STATUSES, f"{folder.name}/{path.name}", "status")
    return {
        "slug": folder.name,
        "tracker_id": str(fm.get("tracker_id", "") or ""),
        "status": status,
        "refined": str(fm.get("refined", "")).lower() != "false",
        "raw_after": _list(fm.get("after"), f"{folder.name}.md"),
    }


def file_unknown(text: str) -> str:
    """The `Unknown:` lines of a leaf file's body, as `pull` writes an entry's `unknown` into Notes."""
    body = COMMENT_RE.sub("", FRONTMATTER_RE.sub("", text, count=1))
    return "; ".join(m.group(1).strip() for m in UNKNOWN_RE.finditer(body))


def load_folder(folder: Path, problems: list[str]) -> list[dict]:
    """One row per ticket in a folder, in build order: every breakdown entry, joined to its
    leaf file when one exists, then leaf files the breakdown does not list. Plans then set
    the status fields of the rows they join."""
    where, at = folder.name, shown(folder)
    rows = {}
    for e in load_breakdown(folder).get("entry", []):
        n, kind = e["id"], e.get("type")
        if kind not in LEAF_TYPES:
            raise TicketError(f"{at}/{BREAKDOWN}: entry {n} type {kind!r} is not one of {', '.join(LEAF_TYPES)}")
        if n in rows:
            raise TicketError(f"{at}/{BREAKDOWN}: two entries with id {n}")
        rows[n] = {
            "epic": where,
            "id": n,
            "file": None,
            "type": kind,
            "tracker_id": "",
            "title": str(e.get("title", "")),
            "status": "",
            "tracker_status": "",
            "state": "planned",
            "assignee": "",
            "refined": False,
            "refine": kind == "bug" or _flag(e.get("refine", False)),
            "description": str(e.get("description", "")),
            "verify": str(e.get("verify", "")),
            "unknown": str(e.get("unknown", "")),
            "references": [str(v) for v in e.get("references", [])],
            "notes": [str(v) for v in e.get("notes", [])],
            "risk": str(e.get("risk", "")),
            "hitl": _flag(e.get("hitl", False)),
            "plan_checkpoint": _flag(e.get("plan_checkpoint", False)),
            "done_checkpoint": _flag(e.get("done_checkpoint", False)),
            "covers": [str(c) for c in e.get("covers", [])],
            "estimate": e.get("estimate", ""),
            "blocked_at": "",
            "blocked_reason": "",
            "raw_after": _list(e.get("after"), f"{at}/{BREAKDOWN} entry {n}"),
            "entry_after": None,
        }
    unlisted, stray, plans = {}, [], []
    seen = {}
    for path in sorted(tickets_dir(folder).glob("*.md")):
        text = read_text(path)
        fm = parse_frontmatter(text, lenient=True)
        if not fm and text.startswith("---") and NAME_RE.match(path.name):
            raise TicketError(f"{at}/{path.name}: frontmatter does not close")
        if fm.get("type") not in LEAF_TYPES:
            if "ticket" in fm:
                plans.append((path.name, fm))
            continue
        try:
            fm = parse_frontmatter(text)
        except TicketError as e:
            raise TicketError(f"{at}/{path.name}: {e}") from e
        status = _one_of(fm.get("status", ""), STATUSES, f"{at}/{path.name}", "status")
        tracker_status = _one_of(fm.get("tracker_status", ""), STATES, f"{at}/{path.name}", "tracker_status")
        n = _id(fm.get("id"))
        if n is not None:
            if n in seen:
                raise TicketError(f"{seen[n]} and {path.name} share the id {n}")
            seen[n] = path.name
        row = rows.get(n) if n is not None else None
        if row is None:
            row = {"epic": where, "id": n, "raw_after": [], "entry_after": None, "covers": [], "title": ""}
            row["refine"] = True
            if n is None:
                stray.append(row)
            else:
                unlisted[n] = row
        else:
            row["entry_after"] = row["raw_after"]
            row["entry_hitl"] = row["hitl"]
        row.update(
            {
                "file": path.name,
                "type": fm.get("type"),
                "tracker_id": str(fm.get("tracker_id", "") or ""),
                "title": str(fm.get("title", "") or row["title"]),
                "status": status,
                "tracker_status": tracker_status,
                "state": tracker_status or STATE_OF[status],
                "assignee": str(fm.get("assignee", "") or ""),
                "refined": _flag(fm.get("refined", False)),
                "refine": row["refine"] or fm.get("type") == "bug",
                "hitl": _flag(fm.get("hitl", False)),
                "covers": [str(c) for c in fm["covers"]] if isinstance(fm.get("covers"), list) else row["covers"],
                "estimate": fm.get("estimate", row.get("estimate", "")),
                "risk": fm["risk"] if isinstance(fm.get("risk"), str) else row.get("risk", ""),
                "blocked_at": fm.get("blocked_at", ""),
                "blocked_reason": str(fm.get("blocked_reason", "") or ""),
                "unknown": file_unknown(text),
            }
        )
        row["raw_after"] = _list(fm.get("after", []), f"{at}/{path.name}")
    out = list(rows.values()) + [unlisted[n] for n in sorted(unlisted, key=lambda n: (isinstance(n, str), n))] + stray
    join_plans(out, plans, at, problems)
    return out


def join_plans(rows: list[dict], plans: list[tuple[str, dict]], where: str, problems: list[str]) -> None:
    """Set each plan's status fields on the one row its `ticket` names; a bad plan is a problem, not an error."""
    for name, fm in plans:
        ticket = fm["ticket"]
        row = None
        if _id(ticket) is not None:
            ticket = _id(ticket)
            row = next((r for r in rows if r["id"] == ticket), None)
        if row is None and isinstance(ticket, str) and ticket:
            row = next((r for r in rows if r["file"] == f"{ticket}.md"), None)
        if row is None:
            problems.append(f"{where}/{name}: ticket {ticket!r} names no entry or leaf file in {where}; skipped")
            continue
        if "plan" in row:
            raise TicketError(f"{where}/{row['plan']} and {name} are both plans for ticket {ticket!r}")
        fields = {k: str(fm.get(k, "") or "") for k in PLAN_FIELDS}
        if "assignee" not in fm:
            # A tracker's assignee is mirrored into the leaf file; the builds' plans carry no assignee.
            fields["assignee"] = row.get("assignee", "")
        try:
            _one_of(fields["status"], STATUSES, f"{where}/{name}", "status")
        except TicketError as e:
            problems.append(f"{e}; the ticket reads as blocked until the plan is fixed")
            fields.update(status="blocked", blocked_reason=f"{name} has an unknown status {fields['status']!r}")
        row.update({"plan": name, "state": row["tracker_status"] or STATE_OF[fields["status"]], **fields})


def epic_folders(initiative: Path) -> list[Path]:
    return sorted(
        d
        for d in tickets_dir(initiative).glob("epic-*")
        if d.is_dir() and ((d / f"{d.name}.md").is_file() or (tickets_dir(d) / BREAKDOWN).is_file())
    )


def load_tree(folder: Path) -> dict:
    """The folder asked about plus every epic its tickets can name."""
    children = load_breakdown(folder).get("initiative")
    if children is not None:
        names = ", ".join(c["slug"] for c in children)
        raise TicketError(f"{folder.name}: a parent initiative holds initiatives; name one of them: {names}")
    epics = epic_folders(folder)
    holder = folder.parent.parent if folder.parent.name == TICKETS else None
    if epics or "epic" in load_breakdown(folder):
        scope, initiative, folders = None, folder, None
    elif holder is not None and folder in epic_folders(holder):
        scope, initiative, folders = folder.name, holder, epic_folders(holder)
        epics = folders
    else:
        scope, initiative, folders = folder.name, None, [folder]
    listed = load_breakdown(initiative).get("epic", []) if initiative else []
    order = [e.get("slug") for e in listed]
    epics.sort(key=lambda d: (order.index(d.name) if d.name in order else len(order), d.name))
    if folders is None:
        folders = [*epics, folder]
    epic_ids = {}
    for e in listed:
        if e["id"] in epic_ids.values():
            raise TicketError(f"{shown(initiative)}/{BREAKDOWN}: two epics with id {e['id']}")
        if e["slug"] in epic_ids:
            raise TicketError(f"{shown(initiative)}/{BREAKDOWN}: two epics with slug {e['slug']}")
        epic_ids[e["slug"]] = e["id"]
    problems = []
    tickets = [t for f in folders for t in load_folder(f, problems)]
    for t in tickets:
        t["key"] = f"{t['epic']}/{t['id']}" if t["id"] is not None else f"{t['epic']}/{t['file']}"
    tree = {
        "scope": scope,
        "initiative": initiative,
        "folders": {f.name: f for f in folders},
        "epic_ids": epic_ids,
        "containers": {f.name: load_container(f) for f in epics},
        "tickets": tickets,
        "problems": problems,
    }
    _resolve(tree)
    _check_cycles(tickets, tree["containers"])
    return tree


def _resolve(tree: dict) -> None:
    tickets, containers = tree["tickets"], tree["containers"]
    by_key = {t["key"]: t for t in tickets}
    slugs = {i: slug for slug, i in tree["epic_ids"].items()}
    ids = {c["tracker_id"]: slug for slug, c in containers.items() if c["tracker_id"]}
    ids.update({t["tracker_id"]: t["key"] for t in tickets if t["tracker_id"]})
    owners: dict[str, list[str]] = {}
    carriers = [(s, c["tracker_id"]) for s, c in containers.items()] + [(t["key"], t["tracker_id"]) for t in tickets]
    for key, tid in carriers:
        if tid:
            owners.setdefault(tid.lower(), []).append(key)
    for tid, keys in owners.items():
        if len(keys) > 1:
            raise TicketError(f"tracker_id {tid!r} is on more than one ticket: {', '.join(keys)}")

    def sibling(t, ref, where):
        """A bare number is always a sibling's id and a quoted one never is; an id with a letter is one when a
        sibling has it."""
        mates = [o for o in tickets if o["epic"] == t["epic"]]
        if isinstance(ref, int) and not isinstance(ref, bool):
            hit = next((o for o in mates if o["id"] == ref), None)
            if hit is None:
                raise TicketError(f"{where}: after {ref!r} names no entry in {t['epic']}")
            return hit["key"]
        for o in mates:
            if isinstance(o["id"], str) and o["id"] == ref:
                return o["key"]
        for o in mates:
            if o["file"] and ref in (o["file"], o["file"][:-3]):
                return o["key"]
        return None

    def resolve(t, refs, where):
        keys = []
        for ref in refs:
            text = str(ref)
            key = sibling(t, ref, where) if "id" in t else None
            m = CROSS_RE.match(text)
            slug = slugs.get(_id(m.group(1))) if m else None
            if key is None and slug:
                key = f"{slug}/{_id(m.group(2))}"
                if key not in by_key:
                    raise TicketError(f"{where}: after {ref!r} names no entry in {slug}")
            if key is None and EPIC_RE.match(text):
                if text in tree["epic_ids"] and text not in containers:
                    raise TicketError(
                        f"{where}: after {ref!r} names an epic {BREAKDOWN} lists that has no folder yet; "
                        f"create {TICKETS}/{text}/{text}.md first"
                    )
                if text not in containers:
                    raise TicketError(f"{where}: after {ref!r} names no epic in this initiative")
                key = text
            if key is None:
                key = ids.get(text)
            if key is None and m:
                raise TicketError(f"{where}: after {ref!r} names no epic id in this initiative's {BREAKDOWN}")
            if key is None and NAME_RE.match(text if text.endswith(".md") else f"{text}.md"):
                raise TicketError(
                    f"{where}: after {ref!r} matches no ticket in {t['epic']}; a file name names a pulled ticket in the "
                    "same folder only: use the entry's id, or move a backlog ticket into the epic as an entry"
                )
            if key is None:
                raise TicketError(f"{where}: after {ref!r} matches no ticket")
            if key not in keys:
                keys.append(key)
        return keys

    at = {name: shown(f) for name, f in tree["folders"].items()}
    for t in tickets:
        where = f"{at[t['epic']]}/{t['file']}" if t["file"] else f"{at[t['epic']]}/{BREAKDOWN} entry {t['id']}"
        t["after"] = resolve(t, t.pop("raw_after"), where)
        planned = t.pop("entry_after")
        t["gated_by"] = []
        entry_hitl = t.pop("entry_hitl", None)
        t["drift"] = {}
        if planned is not None:
            entry_after = resolve(t, planned, where)
            if sorted(entry_after) != sorted(t["after"]):
                t["drift"]["after"] = {"file": t["after"], "entry": entry_after}
            if entry_hitl != t["hitl"]:
                t["drift"]["hitl"] = {"file": t["hitl"], "entry": entry_hitl}
    for slug, c in containers.items():
        gates = resolve({"epic": slug}, c.pop("raw_after"), f"{slug}.md")
        c["after"] = gates
        for t in tickets:
            if t["epic"] == slug:
                t["gated_by"] = gates


def _check_cycles(tickets: list[dict], containers: dict) -> None:
    sys.setrecursionlimit(max(1000, 3 * len(tickets) + 100))
    graph = {t["key"]: t["after"] + t["gated_by"] for t in tickets}
    members = {}
    for t in tickets:
        members.setdefault(t["epic"], []).append(t["key"])
    for slug, c in containers.items():
        members.setdefault(slug, []).extend(c["after"])
    state = {}

    def visit(node, path):
        if state.get(node) == "done":
            return
        if state.get(node) == "active":
            raise TicketError("cycle through " + " -> ".join(path + [node]))
        state[node] = "active"
        for b in graph.get(node, members.get(node, [])):
            visit(b, path + [node])
        state[node] = "done"

    for node in graph:
        visit(node, [])


# ---------------------------------------------------------------- views


def done_keys(tree: dict) -> set:
    done = {t["key"] for t in tree["tickets"] if t["state"] == "done"}
    return done | {slug for slug, c in tree["containers"].items() if c["status"] == "done"}


def in_scope(tree: dict) -> list[dict]:
    return [t for t in tree["tickets"] if tree["scope"] in (None, t["epic"])]


def classify(tree: dict) -> dict:
    done = done_keys(tree)
    # A ticket in review meets an `after`; an epic gate still waits for the epic to be done.
    met = done | {t["key"] for t in tree["tickets"] if t["state"] == "review"}
    groups = {"ready_to_refine": [], "ready_to_start": [], "in_progress": [], "blocked": []}
    for t in in_scope(tree):
        s = t["state"]
        if s in ("done", "dropped"):
            continue
        unmet = [b for b in t["after"] if b not in met] + [b for b in t["gated_by"] if b not in done]
        if t["status"] == "blocked" or t["blocked_at"]:
            t["waiting_on"] = unmet
            groups["blocked"].append(t)
        elif s in ("in-progress", "review"):
            groups["in_progress"].append(t)
        elif unmet:
            t["waiting_on"] = unmet
            groups["blocked"].append(t)
        elif t["refine"] and not t["refined"]:
            groups["ready_to_refine"].append(t)  # refining is where its unknown gets settled
        elif t.get("unknown"):
            groups["blocked"].append(t)
        else:
            groups["ready_to_start"].append(t)
    return groups


def longest_remaining_chain(tree: dict) -> list[str]:
    remaining = {t["key"]: t for t in tree["tickets"] if t["state"] not in ("done", "dropped")}
    memo = {}

    def chain(k):
        if k in memo:
            return memo[k]
        best = []
        for b in remaining[k]["after"] + remaining[k]["gated_by"]:
            if b in remaining:
                c = chain(b)
                if len(c) > len(best):
                    best = c
        memo[k] = best + [k]
        return memo[k]

    longest = []
    for t in in_scope(tree):
        if t["key"] in remaining:
            c = chain(t["key"])
            if len(c) > len(longest):
                longest = c
    return [ref(k, None, tree) for k in longest]


def ref(key: str, epic: str | None, tree: dict) -> str | int:
    """A key as the plan writes it: a sibling's id, `<epic id>.<id>` elsewhere, an epic's slug."""
    slug, _, n = key.partition("/")
    if not n:
        return slug
    if _id(n) is None:
        return key
    if slug == epic:
        return _id(n)
    if slug in tree["epic_ids"]:
        return f"{tree['epic_ids'][slug]}.{n}"
    return key


def declared_after(tree: dict) -> dict:
    """Each epic the initiative's `tickets.toml` lists, in build order, with its `after` as `[{epic, needs}]`."""
    if not tree["initiative"]:
        return {}
    listed = load_breakdown(tree["initiative"]).get("epic", [])
    slugs = [e["slug"] for e in listed]
    by_id = {i: slug for slug, i in tree["epic_ids"].items()}
    out = {}
    for e in listed:
        out[e["slug"]] = []
        for a in e.get("after", []):
            needed = by_id.get(_id(a.get("epic")), a.get("epic"))
            if needed not in slugs:
                raise TicketError(
                    f"{shown(tree['initiative'])}/{BREAKDOWN}: {e['slug']} is after {a.get('epic')!r}, which is no epic listed"
                )
            out[e["slug"]].append({"epic": needed, "needs": a.get("needs", "")})
    return out


def unpinned_after(tree: dict, declared: dict) -> list[dict]:
    """Declared `after` lines whose waiting epic has tickets but none waiting on the named epic."""
    out = []
    for slug, edges in declared.items():
        mine = [t for t in tree["tickets"] if t["epic"] == slug]
        for a in edges:
            needed = a["epic"]
            pinned = any(b == needed or b.startswith(f"{needed}/") for t in mine for b in t["after"] + t["gated_by"])
            if mine and not pinned and tree["scope"] in (None, slug):
                out.append({"epic": slug, "after": needed, "needs": a["needs"]})
    return out


def cross_epic_after(tree: dict, declared: dict) -> dict:
    """`undeclared_after`: an entry's `after` into an epic its epic does not declare. `order_conflict`: an epic
    that waits, declared or through its tickets, on an epic later in the initiative's build order."""
    order = list(declared)
    undeclared, conflicts = [], []

    def conflict(epic, needed):
        pair = {"epic": epic, "after": needed}
        if order.index(needed) > order.index(epic) and pair not in conflicts and tree["scope"] in (None, epic):
            conflicts.append(pair)

    for slug, edges in declared.items():
        for a in edges:
            conflict(slug, a["epic"])
    for slug, c in tree["containers"].items():
        for b in c["after"]:
            if slug in declared and b.partition("/")[0] in declared:
                conflict(slug, b.partition("/")[0])
    for t in tree["tickets"]:
        if t["epic"] not in declared:
            continue
        allowed = {a["epic"] for a in declared[t["epic"]]}
        for b in t["after"]:
            needed = b.partition("/")[0]
            if needed == t["epic"] or needed not in declared:
                continue
            conflict(t["epic"], needed)
            if needed not in allowed and tree["scope"] in (None, t["epic"]):
                undeclared.append(
                    {"epic": t["epic"], "after": needed, "ref": row_ref(t, tree), "names": ref(b, t["epic"], tree)}
                )
    return {"undeclared_after": undeclared, "order_conflict": conflicts}


def row_ref(t: dict, tree: dict) -> str | None:
    """What `find` resolves to this ticket in the folder the command ran on; never the title,
    which can repeat across epics."""
    if t["id"] is not None and t["epic"] in tree["epic_ids"]:
        return f"{tree['epic_ids'][t['epic']]}.{t['id']}"
    if t["id"] is not None and t["epic"] == tree["scope"]:
        return str(t["id"])
    return t["file"]


def public(t: dict, tree: dict, blocks: dict | None = None) -> dict:
    row = {
        k: t[k]
        for k in (
            "epic",
            "id",
            "file",
            "type",
            "tracker_id",
            "title",
            "status",
            "tracker_status",
            "state",
            "assignee",
            "hitl",
            "risk",
            "covers",
            "estimate",
            "refine",
            "refined",
            "blocked_at",
            "blocked_reason",
        )
    }
    row["ref"] = row_ref(t, tree)
    row["after"] = [ref(b, t["epic"], tree) for b in t["after"]]
    if t["gated_by"]:
        row["gated_by"] = t["gated_by"]
    if blocks is not None:
        row["blocks"] = [ref(b, t["epic"], tree) for b in blocks.get(t["key"], [])]
    if t["drift"]:
        row["drift"] = dict(t["drift"])
        if "after" in row["drift"]:
            row["drift"]["after"] = {k: [ref(b, t["epic"], tree) for b in v] for k, v in t["drift"]["after"].items()}
    if t.get("waiting_on"):
        row["waiting_on"] = [ref(b, t["epic"], tree) for b in t["waiting_on"]]
    for key in ("unknown", "plan_checkpoint", "done_checkpoint"):
        if t.get(key):
            row[key] = t[key]
    return row


# ---------------------------------------------------------------- store


def find_project_root(start: Path) -> Path | None:
    for p in [start, *start.parents]:
        if (p / "_bmad").is_dir():
            return p
    return None


def project_root_for(args, start: Path) -> Path | None:
    return Path(args.project_root).resolve() if args.project_root else find_project_root(start)


def store_config(project_root: Path | None) -> dict:
    """The `[tickets]` table of the project's store config, empty when there is none."""
    if not project_root:
        return {}
    cfg = project_root / "_bmad" / "custom" / "ticketing-store-config.toml"
    if not cfg.is_file():
        return {}
    tickets = tomllib.loads(read_text(cfg)).get("tickets", {})
    return tickets if isinstance(tickets, dict) else {}


def store_name(project_root: Path | None) -> str:
    store = store_config(project_root).get("store", "repo")
    return store if isinstance(store, str) and store else "repo"


def central_config(project_root: Path) -> dict:
    """The BMad config with its layers merged by the project's own `config_utils.py`."""
    path = project_root / "_bmad" / "scripts" / "config_utils.py"
    if not path.is_file():
        raise TicketError(f"cannot read the BMad config: {path} is missing")
    spec = importlib.util.spec_from_file_location("bmad_config_utils", path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    try:
        return module.load_central_config(project_root)
    except module.ConfigError as e:
        raise TicketError(str(e)) from e


def tickets_root(project_root: Path, config: dict | None = None) -> Path:
    """`{output_folder}` for the project: the ticket tree lives beside the documents."""
    config = central_config(project_root) if config is None else config
    core = config.get("core", {})
    output = str(core.get("output_folder", "") if isinstance(core, dict) else "")
    output = output.replace("{project-root}", str(project_root))
    return project_root / output


def active_initiative(project_root: Path) -> Path:
    """`{output_folder}/{active_initiative}` for the project."""
    config = central_config(project_root)
    core = config.get("core", {})
    name = core.get("active_initiative") if isinstance(core, dict) else None
    if not isinstance(name, str) or not name.strip():
        raise TicketError(
            "no active initiative: set core.active_initiative in _bmad/custom/config.user.toml, or pass a folder"
        )
    folder = (tickets_root(project_root, config) / name.strip()).resolve()
    if not folder.is_dir():
        raise TicketError(f"active initiative folder not found: {folder}")
    return folder


# ---------------------------------------------------------------- commands


def _folder(args) -> Path:
    if args.dir is None:
        root = project_root_for(args, Path.cwd())
        if root is None:
            raise TicketError("no project root found: no _bmad/ at or above the working directory; pass --project-root")
        # The store is then read from this project even when output_folder lies outside it.
        args.project_root = str(root)
        folder = active_initiative(root)
        backlog = (tickets_root(root) / "backlog").resolve()
        args.backlog = backlog if backlog.is_dir() and backlog != folder else None
        return folder
    return locate_dir(args, args.dir)


def locate_dir(args, name: str) -> Path:
    """A `<dir>` argument as a folder: as given, else under the store, the active initiative and its `tickets/`
    folder, or the project root."""
    folder = Path(name).resolve()
    root = None if folder.is_dir() or Path(name).is_absolute() else project_root_for(args, Path.cwd())
    bases = []
    if root is not None:
        try:
            bases = [tickets_root(root), root]
        except TicketError:  # no BMad config to name the store: the project root alone
            bases = [root]
        else:
            try:
                active = active_initiative(root)
            except TicketError:  # none set: the store and the project root
                pass
            else:
                bases[1:1] = [active, *([active / TICKETS] if (active / TICKETS).is_dir() else [])]
        for base in bases:
            if (base / name).is_dir():
                return container_of((base / name).resolve())
    if not folder.is_dir():
        looked = [] if Path(name).is_absolute() else [Path.cwd(), *bases]
        where = f" (looked under {', '.join(str(p) for p in looked)})" if looked else ""
        raise TicketError(f"not a folder: {name}{where}")
    return container_of(folder)


def with_backlog(args, out: dict, view) -> dict:
    """Add the backlog's view to a command that ran on the active initiative."""
    backlog = getattr(args, "backlog", None)
    if backlog is not None:
        try:
            out["backlog"] = view(backlog)
        except (TicketError, OSError, ValueError) as e:
            out["backlog"] = {"folder": backlog.name, "error": str(e)}
        for group in ("ready_to_refine", "ready_to_start", "in_progress", "blocked", "tickets"):
            for row in out["backlog"].get(group, []):
                # An id here could also name a ticket in the initiative; the file name cannot.
                row["ref"] = row["file"] or row["ref"]
    return out


def locate(args, folder: Path, text: str) -> tuple[dict, dict]:
    """The ticket a reference names and the tree holding it: the folder's, else, with no `<dir>`, the backlog's."""
    tree = load_tree(folder)
    try:
        return resolve_ticket(tree, text), tree
    except NoMatch as miss:
        backlog = getattr(args, "backlog", None)
        if backlog is None:
            raise
        try:
            other = load_tree(backlog)
            other["fallback"] = True
            return resolve_ticket(other, text), other
        except (NoMatch, OSError, ValueError):  # not there, or a backlog that cannot be read
            raise miss from None


def next_view(folder: Path) -> dict:
    tree = load_tree(folder)
    declared = declared_after(tree)
    return {
        "folder": folder.name,
        **{k: [public(t, tree) for t in v] for k, v in classify(tree).items()},
        "unpinned_after": unpinned_after(tree, declared),
        **cross_epic_after(tree, declared),
        **({"problems": tree["problems"]} if tree["problems"] else {}),
    }


def is_parent(folder: Path) -> bool:
    return "initiative" in load_breakdown(folder)


def parent_view(folder: Path) -> dict:
    """A parent initiative's children in build order: each with its declared `after` and, once its folder exists,
    its file's status and its ticket counts; `waiting_on` names the prerequisites not yet done."""
    listed = load_breakdown(folder)["initiative"]
    where = f"{shown(folder)}/{BREAKDOWN}"
    ids = {}
    for c in listed:
        if c["id"] in ids.values():
            raise TicketError(f"{where}: two initiatives with id {c['id']}")
        if c["slug"] in ids:
            raise TicketError(f"{where}: two initiatives with slug {c['slug']}")
        ids[c["slug"]] = c["id"]
    by_id = {i: slug for slug, i in ids.items()}
    rows = []
    for c in listed:
        after = []
        for a in c.get("after", []):
            needed = by_id.get(_id(a.get("initiative")), a.get("initiative"))
            if needed not in ids:
                raise TicketError(
                    f"{where}: {c['slug']} is after {a.get('initiative')!r}, which is no initiative listed"
                )
            after.append({"initiative": needed, "needs": str(a.get("needs", ""))})
        row = {"id": c["id"], "slug": c["slug"], "title": str(c.get("title", "")), "after": after, "folder": None}
        child = folder.parent / c["slug"]
        if child.is_dir():
            row.update(
                folder=child.name, status=load_container(child)["status"], counts=counts(load_tree(child)["tickets"])
            )
        rows.append(row)
    graph = {r["slug"]: [a["initiative"] for a in r["after"]] for r in rows}
    state = {}

    def visit(node, path):
        if state.get(node) == "active":
            raise TicketError("cycle through " + " -> ".join(path + [node]))
        if state.get(node) is None:
            state[node] = "active"
            for b in graph[node]:
                visit(b, path + [node])
            state[node] = "done"

    for node in graph:
        visit(node, [])
    order = list(graph)
    done = {r["slug"] for r in rows if r.get("status") == "done"}
    conflicts = []
    for r in rows:
        r["waiting_on"] = [b for b in graph[r["slug"]] if b not in done]
        conflicts += [
            {"initiative": r["slug"], "after": b} for b in graph[r["slug"]] if order.index(b) > order.index(r["slug"])
        ]
    return {"folder": folder.name, "initiatives": rows, "order_conflict": conflicts}


def cmd_next(args) -> dict:
    folder = _folder(args)
    store = store_name(project_root_for(args, folder))
    if store != "repo" and not args.synced:
        raise StoreRefusal(f"store is {store}: sync ticket status from the tracker first, then rerun with --synced")
    if is_parent(folder):
        out = parent_view(folder)
        out["ready"] = [
            r["slug"] for r in out["initiatives"] if not r["waiting_on"] and r.get("status") not in ("done", "dropped")
        ]
        return with_backlog(args, {"store": store, **out}, next_view)
    return with_backlog(args, {"store": store, **next_view(folder)}, next_view)


def next_id(tree: dict, folder: str) -> int:
    """The id a new ticket in the folder takes: one past the highest number its entries and leaf files use, or
    `next_id` at the top of its `tickets.toml` when that is higher."""
    # `6a` uses up 6: a lettered id is a split of that number.
    used = [re.match(r"\d+", str(t["id"])) for t in tree["tickets"] if t["epic"] == folder]
    # An entry moved to another folder took its id along; the counter it left keeps that id from coming back.
    floor = load_breakdown(tree["folders"][folder]).get("next_id", 0)
    return max(max((int(m.group()) for m in used if m), default=0) + 1, floor)


def counts(tickets: list[dict]) -> dict:
    out = {"total": len(tickets)}
    for t in tickets:
        out[t["state"]] = out.get(t["state"], 0) + 1
    return out


def status_view(folder: Path) -> dict:
    tree = load_tree(folder)
    declared = declared_after(tree)
    tickets = in_scope(tree)
    blocks = {}
    for t in tree["tickets"]:
        for b in t["after"]:
            blocks.setdefault(b, []).append(t["key"])
    out = {
        "folder": folder.name,
        "tickets": [public(t, tree, blocks) for t in tickets],
        "counts": counts(tickets),
        "longest_remaining_chain": longest_remaining_chain(tree),
        "unpinned_after": unpinned_after(tree, declared),
        **cross_epic_after(tree, declared),
        **({"problems": tree["problems"]} if tree["problems"] else {}),
    }
    if tree["scope"] is None:
        # The declared epics in build order, folder or not, then any folder the breakdown does not list.
        out["epics"] = []
        for slug in [*declared, *sorted(s for s in tree["containers"] if s not in declared)]:
            row = {"slug": slug, "id": tree["epic_ids"].get(slug), "after": declared.get(slug, [])}
            c = tree["containers"].get(slug)
            if c is None:
                row["folder"] = None
            else:
                row.update(
                    folder=slug,
                    status=c["status"],
                    refined=c["refined"],
                    gated_by=c["after"],
                    blocks=[ref(b, None, tree) for b in blocks.get(slug, [])],
                    next_id=next_id(tree, slug),
                )
            out["epics"].append(row)
    else:
        out["next_id"] = next_id(tree, tree["scope"])
    return out


def cmd_status(args) -> dict:
    folder = _folder(args)
    store = store_name(project_root_for(args, folder))
    view = parent_view if is_parent(folder) else status_view
    return with_backlog(args, {"store": store, **view(folder)}, status_view)


PULLED = """---
{frontmatter}
---

# {heading}

## Description

{description}

## Requirements

Verify: {verify}

## References

- parent — {parent}
{references}{notes}"""


def title_slug(title: str) -> str:
    title = unicodedata.normalize("NFKD", title).encode("ascii", "ignore").decode("ascii")
    return re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")[:60].rstrip("-") or "untitled"


def resolve_ticket(tree: dict, text: str) -> dict:
    """The one row a reference names; see `<ref>` above."""
    tickets = tree["tickets"]
    ref = text.strip()
    if not ref:
        raise TicketError("the ticket reference is empty")
    low = ref.lower()
    hits = []
    m = CROSS_RE.match(ref)
    if m:
        slug = {i: s for s, i in tree["epic_ids"].items()}.get(_id(m.group(1)))
        hits = [t for t in tickets if slug and t["epic"] == slug and t["id"] == _id(m.group(2))]
    elif _id(ref) is not None and tree["scope"]:
        hits = [t for t in tickets if t["epic"] == tree["scope"] and t["id"] == _id(ref)]
    if not hits and isinstance(_id(ref), str):
        # An id with a letter names its ticket in any epic before the words of a title can match it.
        hits = [t for t in tickets if t["id"] == _id(ref)]
    for pool in (in_scope(tree), tickets):
        if not hits:
            hits = [t for t in pool if t["file"] and low in (t["file"].lower(), t["file"][:-3].lower())]
    if not hits:
        hits = [t for t in tickets if t["tracker_id"] and low == t["tracker_id"].lower()]
    if not hits and not ref.isdigit():
        hits = [t for t in in_scope(tree) if low in t["title"].lower()]
    if not hits:
        raise NoMatch(f"no ticket matches {ref!r}")
    if len(hits) > 1:
        names = ", ".join(ref_name(t, tree) for t in hits)
        raise TicketError(f"{ref!r} matches more than one ticket: {names}")
    return hits[0]


def leaf_stem(t: dict, tree: dict) -> str:
    """The leaf file's stem, or the one `pull` gives it: `<type>-<slug of the title>`, with `-<id>` added when a
    file or an earlier entry in the folder already has that name."""
    if t["file"]:
        return t["file"][:-3]
    stem = f"{t['type']}-{title_slug(t['title'])}"
    taken = set()
    for o in tree["tickets"]:
        if o is t:
            break
        if o["epic"] == t["epic"] and not o["file"]:
            taken.add(f"{o['type']}-{title_slug(o['title'])}")
    taken |= {o["file"][:-3] for o in tree["tickets"] if o["epic"] == t["epic"] and o["file"]}
    return f"{stem}-{t['id']}" if stem in taken else stem


def plan_path(t: dict, tree: dict) -> Path:
    """The joined plan, else `<leaf stem>-plan.md`."""
    folder = tickets_dir(tree["folders"][t["epic"]])
    if t.get("plan"):
        return folder / t["plan"]
    return folder / f"{leaf_stem(t, tree)}-plan.md"


def cmd_find(args) -> dict:
    folder = _folder(args)
    t, tree = locate(args, folder, args.ref)
    home = tree["folders"][t["epic"]]
    row = public(t, tree)
    if tree.get("fallback") and t["file"]:
        row["ref"] = t["file"]  # as the backlog view names it: an id could name a ticket in the initiative
    # Once a leaf file exists it holds the text and the entry's copy is no longer kept up. Unlisted and stray
    # leaves have no entry.
    entry = {} if t["file"] else t
    container = home / f"{home.name}.md"  # an epic's file, or the initiative's for a leaf directly under one
    return {
        **row,
        "folder": home.name,
        "description": entry.get("description", ""),
        "verify": entry.get("verify", ""),
        "references": entry.get("references", []),
        "notes": entry.get("notes", []),
        "unknown": t.get("unknown", ""),
        "epic_file": str(container) if container.is_file() else None,
        "story_file": str(tickets_dir(home) / t["file"]) if t["file"] else None,
        "plan": str(plan_path(t, tree)),
    }


def ref_name(t: dict, tree: dict) -> str:
    return f"{t['file'] or t['id']} in {t['epic']}"


def cmd_pull(args) -> dict:
    folder = _folder(args)
    tree = load_tree(folder)
    n = _id(args.id)
    t = next((t for t in in_scope(tree) if t["epic"] == folder.name and t["id"] == n), None)
    if n is None or t is None:
        raise TicketError(f"{shown(folder)}/{BREAKDOWN} has no entry {args.id}")
    if t["file"]:
        raise TicketError(f"entry {args.id} is already pulled: {t['file']}")
    path = write_leaf(t, tree, project_root_for(args, folder))
    return {"file": path.name, "refine": t["refine"]}


def parent_reference(folder: Path, root: Path | None) -> str:
    """The epic file as a leaf's `- parent —` reference names it."""
    epic_file = folder / f"{folder.name}.md"
    try:
        return Path(os.path.relpath(epic_file, root)).as_posix() if root else epic_file.as_posix()
    except ValueError:  # another drive on Windows
        return epic_file.as_posix()


def write_leaf(t: dict, tree: dict, root: Path | None) -> Path:
    """Write an entry's leaf file from the entry and name it on the row."""
    folder = tree["folders"][t["epic"]]
    path = tickets_dir(folder) / f"{leaf_stem(t, tree)}.md"
    if path.exists():
        raise TicketError(f"{path.name} exists already; change entry {t['id']}'s title")
    after = [str(ref(b, t["epic"], tree)) for b in t["after"]]
    parent = parent_reference(folder, root)
    notes = ([f"Unknown: {t['unknown']}"] if t["unknown"] else []) + t["notes"]
    # Other empty fields are left out; `after` and `hitl` stay because `status` compares them with the entry.
    # No status: the build writes it when it starts.
    fields = [
        ("id", str(t["id"])),
        ("type", t["type"]),
        ("title", json.dumps(t["title"], ensure_ascii=False)),
        ("parent", t["epic"]),
        ("covers", f"[{', '.join(t['covers'])}]" if t["covers"] else ""),
        ("after", f"[{', '.join(after)}]"),
        ("refined", "false" if t["refine"] else ""),
        ("hitl", "true" if t["hitl"] else "false"),
        ("risk", t["risk"]),
        ("estimate", json.dumps(str(t["estimate"])) if t["estimate"] != "" else ""),
    ]
    path.write_text(
        PULLED.format(
            frontmatter="\n".join(f"{k}: {v}" for k, v in fields if v != ""),
            heading=t["title"],
            parent=parent,
            description=t["description"],
            verify=t["verify"],
            references="".join(f"- {r}\n" for r in t["references"]),
            notes="\n## Notes\n\n" + "".join(f"- {n}\n" for n in notes) if notes else "",
        ),
        encoding="utf-8",
    )
    t["file"] = path.name
    return path


def quoted(value: str) -> str:
    """A double-quoted scalar that `parse_frontmatter` reads back exactly."""
    # parse_frontmatter cuts a value at "   #" and splits lines on these characters, so they go in as escapes.
    text = json.dumps(value, ensure_ascii=False).replace("   #", "   \\u0023")
    return text.translate({c: f"\\u{c:04x}" for c in (0x85, 0x2028, 0x2029)})


def edit_frontmatter(path: Path, values: dict[str, str]) -> str:
    """Set frontmatter values in a file, keeping its line endings and byte-order mark; an empty value removes
    the line. Returns the new text."""
    raw = path.read_bytes()
    text = raw.decode("utf-8-sig").replace("\r\n", "\n")
    for key, value in values.items():
        text = set_frontmatter_value(text, key, value)
    data = text.encode("utf-8")  # before the file is opened, so a failure leaves the file as it was
    if b"\r\n" in raw:
        data = data.replace(b"\n", b"\r\n")
    if raw.startswith(codecs.BOM_UTF8):
        data = codecs.BOM_UTF8 + data
    path.write_bytes(data)
    return text


def cmd_mark(args) -> dict:
    """Write status, assignee, and the blocked fields to the ticket's plan; never to its leaf file."""
    folder = _folder(args)
    store = store_name(project_root_for(args, folder))
    if store != "repo":
        raise StoreRefusal(f"store is {store}: change status through the store's write verb, not this script")
    t, tree = locate(args, folder, args.ref)
    path = plan_path(t, tree)
    blocked = {"blocked_at": "", "blocked_reason": ""}
    if args.blocked is not None:
        blocked = {"blocked_at": quoted(date.today().isoformat()), "blocked_reason": quoted(args.blocked)}
    created = not t.get("plan")
    if created:
        assignee = args.assignee if args.assignee is not None else t["assignee"]
        fields = [
            ("title", quoted(t["title"])),
            ("ticket", str(t["id"]) if t["id"] is not None else quoted(t["file"][:-3])),
            ("status", args.status),
            ("assignee", quoted(assignee) if assignee else ""),
            *blocked.items(),
        ]
        text = "---\n" + "".join(f"{k}: {v}\n" for k, v in fields if v != "") + "---\n"
        data = text.encode("utf-8")  # before the file is opened, so a failure leaves no empty plan
        try:
            with path.open("xb") as f:
                f.write(data)
        except FileExistsError:
            raise TicketError(f"{path.name} exists already and is not the plan for {ref_name(t, tree)}") from None
    else:
        values = {"status": args.status, **blocked}
        if args.assignee is not None:
            values["assignee"] = quoted(args.assignee)
        text = edit_frontmatter(path, values)
    fm = parse_frontmatter(text, lenient=True)
    return {
        "plan": str(path),
        "created": created,
        **{k: fm.get(k, "") for k in PLAN_FIELDS},
    }


MIRROR_KEYS = ("ref", "tracker_id", "remote", "tracker_status", "assignee", "after")


def _mirror_values(item: dict, where: str) -> dict[str, str]:
    """The frontmatter lines one mirrored ticket sets, each validated."""
    values = {}
    for key in ("tracker_id", "remote", "assignee"):
        if key in item:
            value = item[key]
            if key == "tracker_id" and isinstance(value, int) and not isinstance(value, bool):
                value = str(value)  # a tracker that numbers its items returns a number
            if not isinstance(value, str):
                raise TicketError(f"{where}: {key} must be a string")
            values[key] = quoted(value) if value else ""
    if "tracker_status" in item:
        values["tracker_status"] = _one_of(item["tracker_status"], STATES, where, "tracker_status")
    if "after" in item:
        if not isinstance(item["after"], list):
            raise TicketError(f"{where}: after must be a list")
        parts = []
        for b in item["after"]:
            if isinstance(b, bool) or not isinstance(b, (int, str)) or (isinstance(b, str) and ("," in b or not b)):
                raise TicketError(f"{where}: after takes ids and names without commas, not {b!r}")
            # A number stays bare, a sibling's id; a string is quoted, so a tracker id of digits stays one.
            parts.append(str(b) if isinstance(b, int) else quoted(b))
        values["after"] = f"[{', '.join(parts)}]"
    return values


def by_tracker_id(tree: dict, tracker_id: str) -> dict:
    """The one ticket that already carries a tracker id; nothing else about it is matched."""
    hits = [t for t in tree["tickets"] if t["tracker_id"] and t["tracker_id"].lower() == tracker_id.lower()]
    if not hits:
        raise NoMatch(f"no ticket carries tracker_id {tracker_id!r}")
    if len(hits) > 1:
        raise TicketError(f"tracker_id {tracker_id!r} is on more than one ticket")
    return hits[0]


def container_named(tree: dict, name: str, by_ref: bool) -> dict | None:
    """The container whose tracker id, or with a `ref` whose folder, is `name`: its folder and its file's status."""
    low = name.lower()
    for folder in [*tree["folders"].values(), *([tree["initiative"]] if tree["initiative"] else [])]:
        path = folder / f"{folder.name}.md"
        if not path.is_file():
            continue
        fm = parse_frontmatter(read_text(path), lenient=True)
        tracker_id = str(fm.get("tracker_id", "") or "")
        if fm.get("type") in CONTAINER_TYPES and (
            (tracker_id and tracker_id.lower() == low) or (by_ref and folder.name.lower() == low)
        ):
            return {"folder": folder.name, "status": str(fm.get("status", "") or "")}
    return None


def cmd_mirror(args) -> dict:
    """Write what the tracker returned into each ticket's leaf file; never `status`, which is the build's."""
    folder = _folder(args)
    root = project_root_for(args, folder)
    store = store_name(root)
    if store == "repo":
        raise StoreRefusal("store is repo: there is no tracker to mirror")
    items = json.load(sys.stdin)
    if not isinstance(items, list) or not all(isinstance(i, dict) for i in items):
        raise TicketError("mirror reads a JSON array of objects on stdin")
    folders, trees = [folder], [load_tree(folder)]
    if getattr(args, "backlog", None):
        try:
            trees.append(load_tree(args.backlog))
            folders.append(args.backlog)
        except (TicketError, OSError, ValueError):  # a backlog that cannot be read holds no match
            pass
    work, unmatched, containers, seen = [], [], [], set()
    for n, item in enumerate(items, 1):
        name = item.get("ref", item.get("tracker_id"))
        where = f"item {n} ({name!r})"
        unknown = [k for k in item if k not in MIRROR_KEYS]
        if unknown:
            hint = "; status is the build's and is never mirrored" if "status" in unknown else ""
            raise TicketError(f"{where}: unknown key {unknown[0]!r}{hint}")
        if isinstance(name, bool) or not isinstance(name, (int, str)) or name == "":
            raise TicketError(f"item {n}: give a ref, or the tracker_id of a ticket that already carries it")
        values = _mirror_values(item, where)
        hit = miss = container = None
        for tree in trees:
            try:
                # A tracker id alone never falls back to an entry id or a title that happens to match it.
                t = resolve_ticket(tree, str(name)) if "ref" in item else by_tracker_id(tree, str(name))
            except NoMatch as e:
                miss = miss or e
                container = container or container_named(tree, str(name), "ref" in item)
                continue
            hit = (t, tree)
            break
        if hit is None:
            if container:
                # A container's file is Sofia's to edit; the row shows what the tracker says against it.
                containers.append({"ref": name, **container, "tracker_status": item.get("tracker_status", "")})
            else:
                unmatched.append({"ref": name, "error": str(miss)})
            continue
        if id(hit[0]) in seen:
            raise TicketError(f"{where}: {ref_name(*hit)} is named twice")
        seen.add(id(hit[0]))
        work.append((*hit, values))
    undo, mirrored = [], []
    try:
        for t, tree, values in work:
            pulled = not t["file"]
            if pulled:
                path = write_leaf(t, tree, root)
                undo.append((path, None))
            else:
                path = tickets_dir(tree["folders"][t["epic"]]) / t["file"]
                undo.append((path, path.read_bytes()))
            edit_frontmatter(path, values)
            ref_out = row_ref(t, tree) if tree is trees[0] else path.name
            mirrored.append({"ref": ref_out, "file": path.name, "pulled": pulled, "set": sorted(values)})
        for f in folders:
            load_tree(f)
    except (TicketError, OSError, ValueError) as e:
        for path, raw in undo:
            try:
                if raw is None:
                    path.unlink(missing_ok=True)
                else:
                    path.write_bytes(raw)
            except OSError:  # keep restoring the rest; the first failure is the one reported
                pass
        # The discovery procedure needs `unmatched` even when a known ticket's `after` named one of them.
        raise TicketError(f"nothing was mirrored: {e}", **({"unmatched": unmatched} if unmatched else {})) from e
    return {
        "folder": folder.name,
        "store": store,
        "mirrored": mirrored,
        "containers": containers,
        "unmatched": unmatched,
    }


TABLE_RE = re.compile(r"^[ \t]*\[\[(entry|epic)\]\][ \t]*(?:#.*)?$", re.M)
ID_LINE_RE = re.compile(r"^([ \t]*id[ \t]*=[ \t]*)(\S+)(.*)$", re.M)
AFTER_LINE_RE = re.compile(r"^([ \t]*after[ \t]*=[ \t]*)\[(.*?)\](.*)$", re.M)
NEXT_ID_LINE_RE = re.compile(r"^[ \t]*next_id[ \t]*=.*$", re.M)


def split_tables(text: str) -> tuple[str, list[str]]:
    """A breakdown's text as the lines before its first table and one string per `[[entry]]` or `[[epic]]`."""
    starts = [m.start() for m in TABLE_RE.finditer(text)]
    if not starts:
        return text, []
    return text[: starts[0]], [text[a:b] for a, b in zip(starts, [*starts[1:], len(text)], strict=True)]


def parse_table(block: str) -> tuple[str, dict]:
    data = tomllib.loads(block)
    kind = "entry" if "entry" in data else "epic"
    row = data[kind][0]
    row["id"] = _id(row.get("id"))
    return kind, row


def list_text(values: list, toml: bool) -> str:
    """A list as `tickets.toml` or a leaf's frontmatter reads it back: numbers bare, strings quoted."""
    return (
        "["
        + ", ".join(
            str(v) if isinstance(v, int) else (json.dumps(v, ensure_ascii=False) if toml else quoted(v)) for v in values
        )
        + "]"
    )


def set_after_line(block: str, current: list, new: list, where: str) -> str:
    m = AFTER_LINE_RE.search(block)
    try:
        if not m or tomllib.loads(f"after = [{m.group(2)}]")["after"] != current:
            raise ValueError
    except ValueError:
        raise TicketError(f"{where}: `after` must sit on one line to be rewritten") from None
    return block[: m.start()] + m.group(1) + list_text(new, toml=True) + m.group(3) + block[m.end() :]


def cmd_move(args) -> dict:
    """Move an entry, with its leaf file and plan, to another folder, rewriting ids and every `after` that names it."""
    src, dest = locate_dir(args, args.src), locate_dir(args, args.to)
    root = project_root_for(args, src)
    tree = load_tree(src)
    t = resolve_ticket(tree, args.ref)
    old = tree["folders"][t["epic"]]
    if old.resolve() == dest.resolve():
        raise TicketError(f"{ref_name(t, tree)} is already in {dest.name}")
    toml_old = tickets_dir(old) / BREAKDOWN
    head, tables = split_tables(read_text(toml_old)) if toml_old.is_file() else ("", [])
    parsed = [parse_table(b) for b in tables]
    moved = next((i for i, (kind, row) in enumerate(parsed) if kind == "entry" and row["id"] == t["id"]), None)
    if moved is None:
        raise TicketError(f"{ref_name(t, tree)} has no entry in {shown(old)}/{BREAKDOWN}; move moves entries")
    dest_tree = load_tree(dest)
    if dest_tree["scope"] is None:
        raise TicketError(f"{dest.name} holds epics, not entries")
    new_id = next_id(dest_tree, dest.name)
    floor = next_id(tree, old.name)
    same = tree["initiative"] is not None and dest.resolve().parent == tickets_dir(tree["initiative"]).resolve()
    old_epic_id = tree["epic_ids"].get(old.name) if same else None
    dest_epic_id = tree["epic_ids"].get(dest.name) if same else None
    slugs = {i: s for s, i in tree["epic_ids"].items()}
    mates = [o for o in tree["tickets"] if o["epic"] == old.name]

    def is_sibling(e, o) -> bool:
        if isinstance(e, int) and not isinstance(e, bool):
            return o["id"] == e
        return isinstance(e, str) and (
            (isinstance(o["id"], str) and o["id"] == e) or bool(o["file"] and e in (o["file"], o["file"][:-3]))
        )

    def names_moved(e, folder: Path) -> bool:
        if folder.resolve() == old.resolve() and is_sibling(e, t):
            return True
        if all(f.resolve() != folder.resolve() for f in tree["folders"].values()):
            return False  # another initiative numbers its epics its own way
        m = CROSS_RE.match(e) if isinstance(e, str) else None
        return bool(m and slugs.get(_id(m.group(1))) == old.name and _id(m.group(2)) == t["id"])

    def new_ref(folder: Path, where: str):
        if folder.resolve() == dest.resolve():
            return new_id
        if dest_epic_id is None:
            raise TicketError(
                f"{where}: names the moved entry, but {dest.name} has no id in the initiative's {BREAKDOWN} to name it by"
            )
        return f"{dest_epic_id}.{new_id}"

    def own_ref(e, where: str):
        mate = next((o for o in mates if o is not t and is_sibling(e, o)), None)
        if mate is not None:
            if old_epic_id is None or mate["id"] is None:
                raise TicketError(f"{where}: after {e!r} names a ticket in {old.name} that {dest.name} cannot name")
            return f"{old_epic_id}.{mate['id']}"
        m = CROSS_RE.match(e) if isinstance(e, str) else None
        if m and same and slugs.get(_id(m.group(1))) == dest.name:
            return _id(m.group(2))
        return e

    # Every change is computed first; files are written only when all of them are known.
    toml_writes: dict[Path, str] = {}
    fm_writes: list[tuple[Path, dict]] = []
    rewritten = []
    folders = list(tree["folders"].values())
    if all(f.resolve() != dest.resolve() for f in folders):
        folders.append(dest)
    for folder in folders:
        is_old = folder.resolve() == old.resolve()
        home, at = tickets_dir(folder), shown(folder)
        path = home / BREAKDOWN
        if path.is_file():
            f_head, f_tables = (head, list(tables)) if is_old else split_tables(read_text(path))
            changed = False
            for i, block in enumerate(f_tables):
                kind, row = parse_table(block)
                if kind != "entry" or (is_old and i == moved):
                    continue
                where = f"{at}/{BREAKDOWN} entry {row['id']}"
                after = _list(row.get("after"), where)
                new = [new_ref(folder, where) if names_moved(e, folder) else e for e in after]
                if new != after:
                    f_tables[i] = set_after_line(block, after, new, where)
                    rewritten.append({"where": where, "after": new})
                    changed = True
            if is_old:
                f_tables.pop(moved)
                if NEXT_ID_LINE_RE.search(f_head):
                    f_head = NEXT_ID_LINE_RE.sub(f"next_id = {floor}", f_head, count=1)
                else:
                    f_head = f_head.rstrip("\n") + ("\n\n" if f_head.strip() else "") + f"next_id = {floor}\n\n"
                changed = True
            if changed:
                toml_writes[path] = f_head + "".join(f_tables)
        own = folder / f"{folder.name}.md"  # an epic file's own `after` can name the entry too
        for path in sorted({*home.glob("*.md"), *([own] if own.is_file() else [])}):
            if is_old and path.parent == home and path.name == t["file"]:
                continue
            fm = parse_frontmatter(read_text(path), lenient=True)
            if fm.get("type") not in LEAF_TYPES and not (fm.get("type") in CONTAINER_TYPES and path == own):
                continue
            where = f"{at}/{path.name}" if path != own else f"{folder.name}/{own.name}"
            after = _list(fm.get("after"), where)
            new = [new_ref(folder, where) if names_moved(e, folder) else e for e in after]
            if new != after:
                fm_writes.append((path, {"after": list_text(new, toml=False)}))
                rewritten.append({"where": where, "after": new})
    block = tables[moved]
    old_home, dest_home = tickets_dir(old), tickets_dir(dest)
    where = f"{shown(dest)}/{BREAKDOWN} entry {new_id}"
    block = ID_LINE_RE.sub(lambda m: f"{m.group(1)}{new_id}{m.group(3)}", block, count=1)
    after = _list(parsed[moved][1].get("after"), where)
    new = [own_ref(e, where) for e in after]
    if new != after:
        block = set_after_line(block, after, new, where)
        rewritten.append({"where": where, "after": new})
    toml_dest = dest_home / BREAKDOWN
    if toml_dest in toml_writes:
        existing = toml_writes[toml_dest]
    else:
        existing = read_text(toml_dest) if toml_dest.is_file() else ""
    toml_writes[toml_dest] = (existing.rstrip("\n") + "\n\n" if existing.strip() else "") + block.strip("\n") + "\n"
    moves = []
    for name, values in (
        (t["file"], {"id": str(new_id), "parent": dest.name}),
        (t.get("plan"), {"ticket": str(new_id)}),
    ):
        if not name:
            continue
        if (dest_home / name).exists():
            raise TicketError(f"{shown(dest)}/{name} exists already")
        if name == t["file"]:
            fm = parse_frontmatter(read_text(old_home / name))
            after = _list(fm.get("after"), f"{shown(old)}/{name}")
            new = [own_ref(e, f"{shown(old)}/{name}") for e in after]
            if new != after:
                values["after"] = list_text(new, toml=False)
                rewritten.append({"where": f"{shown(dest)}/{name}", "after": new})
        moves.append((old_home / name, dest_home / name, values))

    backup: dict[Path, bytes | None] = {}
    made = None if dest_home.is_dir() else dest_home

    def keep(path: Path) -> None:
        if path not in backup:
            backup[path] = path.read_bytes() if path.exists() else None

    try:
        if made is not None:
            made.mkdir()
        for path, text in toml_writes.items():
            keep(path)
            path.write_text(text, encoding="utf-8")
        for path, values in fm_writes:
            keep(path)
            edit_frontmatter(path, values)
        for source, target, values in moves:
            keep(source)
            keep(target)
            raw = source.read_bytes()
            if source.name == t["file"]:
                raw = re.sub(
                    rb"(?m)^- parent \xe2\x80\x94 [^\r\n]*",
                    lambda _: f"- parent — {parent_reference(dest, root)}".encode(),
                    raw,
                    count=1,
                )
            target.write_bytes(raw)
            edit_frontmatter(target, values)
            source.unlink()
        for folder in (dest, old):
            load_tree(folder)
    except (TicketError, OSError, ValueError) as e:
        for path, raw in backup.items():
            try:
                if raw is None:
                    path.unlink(missing_ok=True)
                else:
                    path.write_bytes(raw)
            except OSError:  # keep restoring the rest; the first failure is the one reported
                pass
        if made is not None and made.is_dir() and not any(made.iterdir()):
            made.rmdir()
        raise TicketError(f"nothing was moved: {e}") from e
    return {
        "from": old.name,
        "to": dest.name,
        "old_id": t["id"],
        "id": new_id,
        "ref": f"{dest_epic_id}.{new_id}" if dest_epic_id is not None else str(new_id),
        "file": t["file"],
        "plan": t.get("plan"),
        "next_id": {old.name: floor},
        "rewritten": rewritten,
    }


REF_HELP = """<ref> is `<epic id>.<entry id>`, an entry id inside an epic folder (one with a letter, from any
folder), a tracker id, a file name, or an unbroken phrase from the title that matches one ticket."""

DIR_HELP = """<dir> is an epic folder, a backlog folder, or an initiative folder (all its epics): a path, or a
folder's name under the store, the active initiative, or its `tickets/` folder. A `tickets/` folder is
read as the folder that holds it. Left out, it is the active initiative."""

BACKLOG_HELP = "With no <dir>, a ticket the initiative does not hold is looked for in the backlog folder."

PARENT_HELP = """A parent initiative, whose `tickets.toml` lists `[[initiative]]` tables, works only with status
and next; the other commands want one of its initiatives."""

HELP = {
    "next": f"""Tickets grouped by what can happen next, in build order.

{DIR_HELP} With no <dir>, `backlog` holds the same view of the backlog folder, each row's `ref`
its file name.

Output: `ready_to_refine` (needs full criteria first), `ready_to_start`, `in_progress`, and `blocked`; a
`blocked` row has `waiting_on`, its unmet prerequisites, `blocked_reason`, or `unknown`, a question to
settle before it starts. A prerequisite is met when it is done or in review; an epic's own gate
(`gated_by`) waits for that epic to be done. `unpinned_after`, `undeclared_after`, `order_conflict`, and
`problems` report a tree that needs fixing. Every row carries `ref`, which find resolves, `state`, and
`risk`; a row has `drift` when the leaf file's `after` or `hitl` differs from the entry's, and `plan_checkpoint` or
`done_checkpoint` when the entry sets it.

On a parent initiative: `initiatives`, as status gives them, `order_conflict`, and `ready`, the
initiatives waiting on none and not done or dropped.""",
    "status": f"""Every ticket in build order.

{DIR_HELP} With no <dir>, `backlog` holds the same view of the backlog folder, each row's `ref`
its file name.

Output: `tickets` (each with `status`, `tracker_status`, `state`, `blocks`, and `drift` when the leaf file's
`after` or `hitl` differs from the entry's, with both values), `counts` by state, `next_id` (the id a new
ticket in the folder takes: one past the highest number used, or `next_id` at the top of the folder's
`tickets.toml` when that is higher; on an initiative, in each `epics` row), `longest_remaining_chain`, and
on an initiative `epics`: one row per epic its `tickets.toml` lists, in build order, then any epic folder
it does not list, each with its declared `after`, and `refined`, false only when the epic file says
`refined: false`. A listed epic with no folder yet has `folder` null and no `status`, `refined`,
`gated_by`, `blocks`, or `next_id`.

On a parent initiative: `initiatives`, one row per initiative its `tickets.toml` lists, in build order,
each with `id`, `slug`, `title`, `after` (as `[{{initiative, needs}}]`), `folder` (null until the folder
exists beside the parent), and `waiting_on`, the initiatives it is after whose file is not `status:
done`; a row with a folder also has that file's `status` and `counts`, its tickets by state.
`order_conflict` lists an initiative after one later in the build order.""",
    "find": f"""The one ticket a reference names.

{DIR_HELP}
{PARENT_HELP}
{BACKLOG_HELP}
{REF_HELP}

Output: the ticket's row; its entry's `description`, `verify`, `references`, and `notes`, all empty once
the entry is pulled, when `story_file` holds them; `unknown`; its `folder`; and the absolute paths
`epic_file` (the file of the container it is under, null in a backlog folder), `story_file` (null until
the entry is pulled), and `plan` (where its plan is or goes; it may not exist yet).""",
    "pull": f"""Write an entry's leaf file from its entry.

<dir> is the epic folder and <id> the entry's id. The file is `<type>-<slug of the title>.md` in the
epic's `tickets/` folder: `after` and `hitl` always, other fields only when the entry sets them, no status.
{PARENT_HELP}""",
    "mark": f"""Set a ticket's status in its plan; never in its leaf file. Repo store only.

{DIR_HELP}
{PARENT_HELP}
{BACKLOG_HELP}
{REF_HELP}

A ticket with no plan gets one holding only frontmatter. `--blocked` sets `blocked_at` (today) and
`blocked_reason`; without it both are cleared.""",
    "mirror": f"""Write what a tracker returned into leaf files. Tracker stores only.

{DIR_HELP}
{PARENT_HELP}
{BACKLOG_HELP}

Stdin is a JSON array, one object per ticket: `ref` (or `tracker_id` alone, which matches only a ticket
that already carries that id) and any of `tracker_id`, `remote`, `tracker_status` ({", ".join(STATES)}),
`assignee`, and `after` (a list: a number is a sibling's id, a string any other prerequisite form, a
tracker id included). Only the keys given are written, an empty string removes the line, and `status` is
never written. An entry with no leaf file is pulled first. A published container, named by its folder or
its tracker id, is listed under `containers` with its file's `status` and the `tracker_status` given, and
nothing is written to it. A ticket the tree does not hold is listed under `unmatched` and the rest are
written; values that would leave the tree unreadable write nothing, and the error then still lists
`unmatched`.""",
    "move": f"""Move an entry to another folder, with its leaf file and plan.

<from> and <to> are folders as <dir> is for the other commands; <ref> names an entry in <from>.
{PARENT_HELP}
{REF_HELP}

The entry takes <to>'s `next_id` as its id and goes last in <to>'s `tickets.toml`; reorder it by hand when
build order needs it. <from> keeps `next_id` atop its `tickets.toml` so the old id never returns. The leaf
file and plan move with it, the file's `id` and `parent` and the plan's `ticket` set. Every `after` naming
the entry is rewritten (a former sibling now says `<epic id>.<id>`) and its own `after` follows; `covers`
is untouched. When any of it fails, nothing is written.

Output: `from`, `to`, `old_id`, `id`, `ref` (as find on the initiative resolves it), `file`, `plan`,
`next_id` (what <from> now keeps), and `rewritten`, one `where` and `after` per line rewritten.""",
}


def main() -> int:
    raw = argparse.RawDescriptionHelpFormatter
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=raw)
    parser.add_argument(
        "--project-root",
        help="project holding _bmad/; default: walk up from the ticket folder, or the working directory with no folder",
    )
    sub = parser.add_subparsers(dest="command", required=True)
    p = sub.add_parser(
        "next",
        help="tickets whose prerequisites are done or in review, by state",
        description=HELP["next"],
        formatter_class=raw,
    )
    p.add_argument("dir", nargs="?", help="default: the active initiative")
    p.add_argument("--synced", action="store_true", help="tracker status was mirrored just now")
    p.set_defaults(func=cmd_next)
    p = sub.add_parser("status", help="every ticket resolved", description=HELP["status"], formatter_class=raw)
    p.add_argument("dir", nargs="?", help="default: the active initiative")
    p.set_defaults(func=cmd_status)
    p = sub.add_parser("find", help="the one ticket a reference names", description=HELP["find"], formatter_class=raw)
    p.add_argument("dir", nargs="?", help="default: the active initiative")
    p.add_argument("ref")
    p.set_defaults(func=cmd_find)
    p = sub.add_parser("pull", help="write an entry's leaf file", description=HELP["pull"], formatter_class=raw)
    p.add_argument("dir")
    p.add_argument("id")
    p.set_defaults(func=cmd_pull)
    p = sub.add_parser(
        "mark",
        help="set a ticket's status in its plan (repo store only)",
        description=HELP["mark"],
        formatter_class=raw,
    )
    p.add_argument("dir", nargs="?", help="default: the active initiative")
    p.add_argument("ref")
    p.add_argument("status", choices=STATUSES)
    p.add_argument("--assignee")
    p.add_argument(
        "--blocked", metavar="REASON", help="set blocked_at to today and blocked_reason; else both are cleared"
    )
    p.set_defaults(func=cmd_mark)
    p = sub.add_parser(
        "mirror",
        help="write what a tracker returned into leaf files (tracker stores only)",
        description=HELP["mirror"],
        formatter_class=raw,
    )
    p.add_argument("dir", nargs="?", help="default: the active initiative")
    p.set_defaults(func=cmd_mirror)
    p = sub.add_parser(
        "move",
        help="move an entry, its leaf file, and its plan to another folder",
        description=HELP["move"],
        formatter_class=raw,
    )
    p.add_argument("src", metavar="from", help="the folder holding the entry")
    p.add_argument("ref")
    p.add_argument("to", help="the folder it goes to")
    p.set_defaults(func=cmd_move)
    args = parser.parse_args()
    try:
        print(json.dumps(args.func(args), ensure_ascii=False, default=str))
        return 0
    except StoreRefusal as e:
        print(json.dumps({"error": str(e)}), file=sys.stderr)
        return 2
    except (TicketError, OSError, ValueError) as e:
        print(json.dumps({"error": str(e), **getattr(e, "data", {})}, ensure_ascii=False), file=sys.stderr)
        return 1


if __name__ == "__main__":
    if sys.platform == "win32":
        # Piped output on Windows defaults to a legacy code page, not UTF-8.
        sys.stdout.reconfigure(encoding="utf-8")
        sys.stderr.reconfigure(encoding="utf-8")
    raise SystemExit(main())

---
name: bmad-agent-po
description: 'Sofia, the Product Owner: turns any input (an idea, brief, PRD, transcript, notes) into a ticket''s requirements, slices initiatives into epics and epics into stories, writes and refines tickets, runs the board, validates the tree, replans when a change lands, and sets up the ticket store. Use when the user asks for Sofia or the product owner, or says ''create a spec'', ''distill this PRD'', ''write the initiative from'', ''slice this'', ''split this epic'', ''break this into stories'', ''make a ticket'', ''refine this ticket'', ''what''s ready'', ''status of'', ''publish'', ''validate these tickets'', ''this changed, what now'', ''set up the ticket store''.'
---

Run the following command exactly once without changing the current working directory.

Replace `{project-root}` with the nearest folder containing `_bmad/`, from the working directory upward.
Replace `{skill-root}` with the absolute path to this skill's directory.

```bash
uv run --no-cache "{project-root}/_bmad/scripts/render_skill.py" --project-root "{project-root}" --skill "{skill-root}"
```

- The command prints the skill's workflow: follow it. `HALT: <reason>`: report the reason and stop.
- If the script does not exist, BMad is not set up in this project yet. Offer to set it up using `bmad` skill, then run the command again. If you do not have the `bmad` skill, offer to install it first.
- On any other output or failure, including `uv` being unavailable, report the command output and stop. Do not run any workflow source directly.

{% if workflow.publication not in ("at_inception", "on_start") %}{{ halt("workflow.publication must be at_inception or on_start, not " ~ workflow.publication) }}{% endif %}
# Sofia — Product Owner

You are Sofia, the Product Owner: turn the user's intent into requirements and tickets a coding agent can build from. The user decides scope and split.

## Conventions

- When a line directs you to another file, read all of it and follow it: the Read tool with no offset or limit, or one `cat`; never a line range, `head`, or `grep`.
- Prefix every message with {{ agent.icon }}. The `bmad` skill answers questions about BMad itself.
- Speak to the user in the ticket's words: a story, an epic, a question to settle, a file to write; say what a script key means rather than naming it.
- A yes approves what was shown and, on the repo store, is the yes to commit it; say "commit" to the user, never "publish". Starting work is a separate yes, and so is a tracker publish, unless asked for together.

## On activation

Run in order (`_None._` means skip):

{{ agent.activation_steps_prepend }}

Stay in persona until dismissed, inside invoked skills too. Role: {{ agent.role }} Identity: {{ agent.identity }} Style: {{ agent.communication_style }} Principles:

{{ agent.principles }}

Session facts (read `file:` globs under `{project-root}`):

{{ agent.persistent_facts }}

{% if initiative_folder == config.core.output_folder %}
No initiative is active. A request that writes an initiative is the yes to create `initiative-<slug>/` under `{{ config.core.output_folder }}` and record it as `active_initiative` under `[core]` in `{project-root}/_bmad/custom/config.user.toml`; on any other request, offer to.
{% else %}
Tickets live under `{{ initiative_folder }}`.
{% endif %}

Then run in order:

{{ agent.activation_steps_append }}

## Dispatch

{{ agent.menu }}

A request or code that fits an item runs its `prompt` or `skill` (skill not installed: offer `npx skills add <repo> --skill <name>`, `<repo>` from the entry's `source` in `{skill-root}/bmod.toml`). No request: greet in one line that says what you do (requirements, slicing, the board, a change mid-flight) and wait. Never show the menu. Unless headless, your first message in a session also tells the user they can invoke `bmad-party-mode` for multi-agent perspectives or `bmad-advanced-elicitation` for deeper exploration at any point.

The ticket at the altitude asked about (the initiative's, an epic's, a leaf's) is the spec; a leaf is its file when pulled, else its entry and its epic file. Read it first and in full on every request about it, before `tickets.py next` or `status` and before any file but the item's reference. Its sources were distilled into it (Source, under Terms); open a companion only when the step needs it. A status or "what's next" answer also lists the initiative folder (a `<type>-<slug>/` absent from `sources:` and `companions:` is an input not yet folded in; do not open them) and reports the file's state (open questions and assumptions, not sliced, epics not refined, entries ready, blocked), naming the next step without starting it. A start request is exempt from the full read: `find` it, confirm it is unassigned, hand off; the builder reads the ticket.

`tickets.py` is `uv run {skill-root}/scripts/tickets.py --project-root {project-root}`. Resolve a leaf with `tickets.py find [<folder>] <ref>`; an epic by its `id` or `slug` in the `epics` of `tickets.py status`. On the repo store, readiness comes from `tickets.py next` or `status`. Open `{{ rendered("references/board.md") }}` to act on a row, when `problems`, `drift`, `unpinned_after`, `undeclared_after`, or `order_conflict` is non-empty, or first on a tracker.

### Intake

Size new work and say which path; a user override is a `Decision:` line. SPEC writes containers; a change after an entry has started is REPLAN, before that SPEC.

- **Standalone** — one bug or story, whatever its input: `{{ rendered("references/ticket.md") }}`.
- **Epic** — several tickets in one epic: Epic into stories in `{{ rendered("references/slice.md") }}`.
- **Initiative** — more than one epic: split into epics per `{{ rendered("references/slice.md") }}`.
- **Several initiatives** — more than one initiative's worth: a parent initiative per `{{ rendered("references/slice.md") }}`.

## How Sofia works

Headless or autonomous: do not stop to ask; gaps become `Open question:` lines and choices `Assumption:` lines, the checks still run (Self-validate in `{{ rendered("references/spec.md") }}` for requirements, the Set or Tree check in `{{ rendered("references/validate.md") }}` for a slice), and the result is committed (on a tracker, published per `publication`) and reported.

### Terms

- Entry: a planned leaf in a `tickets.toml`; with no file and no plan its state is `planned`
- Pull: `tickets.py pull` writes an entry's leaf file, needed to review, refine, or publish it, not to start it
- Refine: for a leaf, full acceptance criteria before start for bugs, entries with `refine = true`, and any leaf with no entry; the user's "refine <ref>" asks for them (set `refine = true`), "review <ref>" is a review; other stories build as they are. For an epic, settling its detail before its stories are planned, per Refining an epic in `slice.md`
- Source: the ticket above (a story's epic, an epic's initiative); an initiative's are its `sources:` files, opened only by SPEC and the checks in `validate.md`
- hitl: boolean leaf frontmatter; part needs a person
- State: `planned`, `backlog`, `in-progress`, `review`, `done`, `dropped`; take it from the `tickets.py` row, never derive it
- repo store: the tree is the store; a configured tracker is a remote that `write` pushes to and `query` reads from

### Store operations

`uv run {skill-root}/scripts/read_store.py --project-root {project-root} -k verbs.<name>` (repeat `-k`) returns a verb to follow. Before the first verb in a session, run it with `-k tickets`; `tickets.py` rows carry `store`. Shared rules: `{{ rendered("references/board.md") }}`.

| Verb | For |
|---|---|
| `setup` | connect the tool; create what the maps name |
| `write` | create or change a ticket — body, state, assignee, parent, blocking, fields |
| `query` | one ticket, a container's children, a search |

### The ticket tree

```
{{ config.core.output_folder }}/
  {active_initiative}/                        # example active_initiative=initiative-checkout
    initiative-checkout.md
    .memlog.md                                # one log per ticket, written by SPEC
    glossary.md
    ux-checkout/                              # another skill's run, beside its ticket
    tickets/
      tickets.toml                            # the epics in build order
      epic-cart-rules/
        epic-cart-rules.md
        .memlog.md
        tickets/
          tickets.toml                        # every planned entry, pulled or not
          story-cart-service-scaffold.md        # pulled; its id is in its frontmatter
          story-cart-service-scaffold-plan.md   # the build's plan, with the ticket's status
          story-apply-discount-codes-plan.md    # a plan for an entry with no file
          spike-discount-engine-latency.md
  backlog/
    bug-checkout-total-ignores-discount-codes.md
```

- A container's breakdown, child epics, leaf files and plans sit in its `tickets/` folder; its own file, memlog and other skills' runs, named after it, sit above. `tickets.py` takes the container's folder, not its `tickets/` folder.
- An initiative's `tickets.toml` lists child initiatives, epics, or entries, one kind only; with epics, every leaf is under one. A parent initiative's children are initiative folders beside it.
- `id` is a number (or an existing id like `6a`), never reused; new tickets take `next_id` from `tickets.py status <folder>`. Refer to it as `3` in its epic, `2.3` from another. File names carry no id except the `-<id>` `pull` adds on a collision; a title change renames the file.
- Table order in `tickets.toml` is the build order, not `id`.
- A leaf file is truth once it exists: keep the entry's `id`, `type`, `title`, `after`, and `hitl` equal to it.
- A leaf's `status`, blocked fields, and repo-store assignee live in its plan file, never the leaf; change them only per `board.md`. Never send the plan to a tracker.

Run read-heavy, small-output work in subagents; if blocked, say so and continue inline.