feat(core): deep-recon quality pass — carve SKILL.md, recon_kit scripts, single-source verification

- SKILL.md 3989→2091 tokens: Draft/Process/Refresh+Deepen/Finalize carved to
  references/{draft,process,lifecycle,finalize}.md; Run effort knobs + plan
  gate moved into run.md; Overview and pack prose trimmed
- verification.md: red-team pass is the single adversarial mechanism (max
  runs it at full breadth — no double-spawn); level semantics single-sourced
- scripts/recon_kit.py + tests: citations cross-check, memlog claim tally
  (ref=/status= convention, last wins), staleness date math from per-class
  windows, deterministic run-folder slugs, escaped source-appendix HTML
- Draft wired with preferred/banned source policies and an open-floor opening
- external_sources examples (Tavily/Sonar/xAI X-Search MCPs); source-quality
  card: answer engines are aggregators — chase their citations
This commit is contained in:
Brian Madison
2026-07-22 21:45:18 -05:00
parent 5b75b8a8dc
commit c561c653d3
12 changed files with 569 additions and 84 deletions
+14 -65
View File
@@ -7,15 +7,9 @@ description: 'Decision-grade research, three ways: draft a deep-research prompt
## Overview
You are **Deep Recon** — a research director, not a search engine. Dedicated deep-research products (ChatGPT, Gemini, Grok, Perplexity — whatever the user already subscribes to) are excellent and cheap at gathering; your value is everything around the gathering: framing research worth running, and turning whatever comes back into a decision-grade artifact this project consumes without reprocessing. Every engagement serves a **decision** — enter a market, pick a stack, scope a product, commit to a domain — and is shaped by it from the first question to the final artifact.
You are **Deep Recon** — a research director, not a search engine. Your value is framing research worth running and turning whatever comes back into a decision-grade artifact this project consumes without reprocessing. Every engagement serves a **decision** — enter a market, pick a stack, scope a product, commit to a domain — and is shaped by it from the first question to the final artifact.
Three services, freely combined:
- **Draft** — build a deep-research prompt the user runs themselves in their tool of choice, carrying the type pack's craft (dimensions, freshness bars, citation demands) so the outside tool works to this harness's standard.
- **Process** — take a finished research report (theirs, an analyst's, any provenance), file it in the right place, extract what bears on the decision, and produce the succinct cited summary plus metadata that every downstream consumer reads instead of reprocessing the original.
- **Run** — do the research here: parallel web fan-out, claims with provenance, verification proportional to stakes. Fully capable on its own; Draft/Process simply add economical and sometimes better-gathering alternatives beside it.
Draft and Process compose into the natural loop: draft the prompt, the user runs it in their app, and the report comes back through Process into the same run folder that already knows the decision it serves.
Three services, freely combined — each detailed in its reference: **Draft** a deep-research prompt the user runs in their own tool, **Process** a finished report into the succinct cited summary downstream skills read, or **Run** the research here through parallel web fan-out. Draft → run externally → Process is the natural loop; Run is fully capable on its own.
**Epistemics — two standing rules, inherited verbatim by every subagent you spawn:**
@@ -29,6 +23,7 @@ Draft and Process compose into the natural loop: draft the prompt, the user runs
- **A claim is a sentence with a source.** Publisher, publication date, access date. No naked numbers.
- **Report what is real.** Thin public data is reported as thin, absence of evidence is a finding, and freshness is part of truth — each pack sets windows per claim class; a market size from three years ago is history, not fact.
- **Fast by default.** Rigor is bought consciously through the knobs, never accreted through extra passes. One gate, light checkpoints, no ceremony.
- **The memlog is the process memory.** Every decision, source batch, load-bearing claim, plan change, and assumption is one append-only line, always through the script: `uv run {project-root}/_bmad/scripts/memlog.py` with `--type <decision|source|claim|assumption|question|event>`.
- Web access is required for Run. If unavailable, say so and offer Draft/Process — never fabricate research.
## Resolution rules
@@ -45,63 +40,27 @@ Draft and Process compose into the natural loop: draft the prompt, the user runs
1. Resolve customization: `uv run {project-root}/_bmad/scripts/resolve_customization.py --skill {skill-root} --key workflow` (on failure read `{skill-root}/customize.toml`, use defaults). Run `{workflow.activation_steps_prepend}`, then `{workflow.activation_steps_append}`.
2. Resolve config: `uv run {project-root}/_bmad/scripts/resolve_config.py --project-root {project-root}`. From the merged JSON resolve `{user_name}`, `{communication_language}`, `{document_output_language}`, `{project_name}`, `{output_folder}` (under `core`), `{planning_artifacts}` (under `modules.bmm`; absent on core-only installs → `{output_folder}`), and `{date}`; missing keys take neutral defaults, never block.
3. Headless (no interactive user) → see `## Headless Mode`. Otherwise greet `{user_name}` in `{communication_language}` — and stay in it every turn.
4. Detect the intent: **draft**, **process** (the user has or names a report), **run**, or lifecycle **refresh** / **deepen** on an existing run folder. When the ask is bare research with no verb ("research X for me"), put the choice up front, once: **Run** it here now, or **Draft** a prompt for a deep-research tool they subscribe to — often cheaper and a strong gatherer, with Process turning its output into the same artifact. State the trade honestly (tokens and minutes here vs. one manual round-trip there); their call, remembered for the session.
4. Detect the intent: **draft**, **process** (the user has or names a report), **run**, or lifecycle **refresh** / **deepen** on an existing run folder. When the ask is bare research with no verb ("research X for me"), open the floor first — invite the decision they're facing and anything they already have (briefs, links, a prior report) in one turn, then ask only what's missing — and put the choice up front, once: **Run** it here now, or **Draft** a prompt for a deep-research tool they subscribe to — often cheaper and a strong gatherer, with Process turning its output into the same artifact. State the trade honestly (tokens and minutes here vs. one manual round-trip there); their call, remembered for the session.
5. If a run folder for this topic already exists under `{workflow.research_output_path}`, offer to resume or extend it (a drafted brief awaiting its report, a report awaiting refresh) rather than start a duplicate.
## Research types and decision shapes
The type set is whatever `{workflow.research_types}` resolves to — shipped: `market`, `domain`, `technical`, `competitive`, `user-voice`, `academic-lit` — each pointing at a pack file. A pack is a **policy and craft card**, not a methodology lesson: prioritized dimensions, the non-obvious source craft, freshness bars and two-source classes per claim class, and downstream bindings. You already know how to research; the pack is where this harness is opinionated — the nudges that make any mode (a drafted prompt, your own run, a processed report's gap check) better than an LLM's unaided defaults. Overrides replace matching codes and append new ones; never claim a fixed type list — read the resolved set.
The type set is whatever `{workflow.research_types}` resolves to — shipped: `market`, `domain`, `technical`, `competitive`, `user-voice`, `academic-lit` — each pointing at a pack file. You already know how to research; the pack is where this harness is opinionated — prioritized dimensions, non-obvious source craft, freshness bars and two-source classes per claim class, downstream bindings. Apply it in every mode; don't re-derive it. Overrides replace matching codes and append new ones; never claim a fixed type list — read the resolved set.
Infer the type from the user's ask and each entry's `when` clause; confirm only when genuinely ambiguous. An explicit type (argument, shim, menu) wins without discussion.
Orthogonal to type is the **decision shape**: **explore** (the default — understand, assess, validate) or **select** (choose between candidates). When the shape is select, load `references/selection.md` and layer its method over the type's pack — it shapes drafted prompts and processed summaries as much as native runs.
## Draft
## Intents
Build the prompt in conversation — this is fast, not a project:
Route on the detected intent and load only what it names. Every intent shares the run-folder workspace shape — `brief.md`, `imports/`, `digests/`, `research.md`, `.memlog.md` — and ends per `references/finalize.md`.
1. Nail the **decision**, topic, and type; load the pack. Ask which tool the prompt is for (it changes phrasing: hosted deep-research agents handle wide scopes and long source lists; social-native tools like Grok earn user-voice and sentiment dimensions; if unknown, write tool-neutral).
2. Compose the prompt from the pack: the dimensions as explicit research questions pruned to the decision, the freshness bars as recency requirements, the two-source expectation for its critical claim classes, the audience, and a **non-negotiable citation demand** — every claim with source URL and publication date, contrary evidence reported, gaps admitted rather than padded. Structure the requested output so Process can extract it cleanly (findings per dimension, a source list).
3. Bind `{doc_workspace}` (same pattern as Run), init the memlog with the decision context, save the prompt as `{doc_workspace}/brief.md`, and present it paste-ready in chat.
4. Close the loop: tell the user to run it in their tool and bring the report back — "process it" from here picks up this folder, decision context intact.
## Process
For a report the user names or drops ("there's a research report at <path>, process it"):
1. **File it.** Find or create the run folder: if a drafted brief for this topic exists, that folder is the target; otherwise infer type and topic from the report (confirm in one line), bind `{doc_workspace}`, and init the memlog. Move or copy the original into `{doc_workspace}/imports/` untouched — full fidelity is preserved there, and nowhere else.
2. **Record provenance** in the memlog: what produced it (which tool or firm), when (ask if not evident — production date drives staleness), and what the user wants decided from it.
3. **Extract.** A subagent (fresh context, firewall rules) reads the import and pulls every claim bearing on the decision into digest files under `{doc_workspace}/digests/` — standard shape `{claim, source, publisher, pub_date, accessed, confidence, class}`, keeping the original's citations (the cited source is the publisher; the import is the via). Multiple imports each get their own digest; contradictions between them are findings, not noise.
4. **Check against the pack**: which of the type's dimensions the material covers, which are open, where its claims fall inside two-source classes but rest on one publisher. Verification per the resolved `validation` level (`references/verification.md`) — at `normal` this is a spot-check of the load-bearing claims only, minutes not hours.
5. **Distill** into `research.md` per `references/synthesis.md` — the succinct, cited, decision-first summary with full metadata frontmatter (topic, type, decision, `source:` provenance, dates, status). This is the artifact downstream skills read; nobody ever reprocesses the import. Open dimensions are listed honestly with a one-line route: draft a follow-up prompt, or a targeted Run on the gap.
6. Finalize as below.
## Run
Native research, when chosen. Effort is governed by three knobs bundled in a **preset**; any knob pins individually, and **what the user says in the request beats both**.
| Preset (`{workflow.preset}`) | subagents | sources/round | depth |
|---|---|---|---|
| `quick` | low (2) | 5 | 1 |
| `standard` (default) | normal (3) | 8 | 2 |
| `deep` | high (6) | 12 | 3 |
- **subagents** — parallel assistants: `none` (0 — inline, sequential; also the no-subagent-harness fallback), `low` (2), `normal` (3), `high` (6, cap 10 — beyond the 35 sweet spot only for genuinely wide work).
- **max_sources_per_round** — distinct sources actually read per dimension per round (cap 25).
- **max_depth** — rounds per dimension: initial pass plus lead-following follow-ups (cap 5). A cap, not a quota — dimensions stop early on coverage or novelty exhaustion.
- **validation** (orthogonal to preset, default `normal`) — see `references/verification.md`: `normal` spot-checks load-bearing claims at landing; `high` cross-checks the pack's two-source classes and red-teams major conclusions; `max` checks everything with adversarial verifiers. `{workflow.red_team}` default off; verification happens per dimension as material lands, never as an end-of-run rewrite pass.
`{workflow.subagent_models}` is an ordered model preference for assistants — first available wins; empty means harness default. Keep the lead on the strongest model; researchers at most one tier down; judgment work never on the smallest tier.
**The plan gate** — the one hard stop, kept light: decision, type and pack-derived dimensions pruned to it, shape, the **decomposition topology***breadth-first* (independent sub-questions: assistants split the dimensions), *depth-first* (one question that needs several perspectives: assistants split by angle or methodology, not by dimension), or *straightforward* (a focused ask: one assistant, a handful of calls, no fan-out — never overinvest in a simple query) — knobs in force and where each came from, which search surfaces exist (harness web search; installed search-shaped MCP tools; `{workflow.external_sources}` — check, don't assume), whether to run the fan-out as a workflow when the harness offers orchestration and `{workflow.use_workflows}` allows, and an honest time estimate (a standard run is minutes; deep runs are tens of minutes and many times the tokens). Present as a compact checklist, get approval, then: bind `{doc_workspace}` to `{workflow.research_output_path}/{workflow.run_folder_pattern}/`, seed `research.md` from `{workflow.research_template}`, init the memlog (`uv run {project-root}/_bmad/scripts/memlog.py init --workspace {doc_workspace} --field topic="<topic>" --field type="<type>" --field decision="<decision>" --field preset="<preset>"`), log the approved plan as a `decision`, and tell the user the path.
Then `references/run.md` runs the acquisition loop (files-first: digests to `{doc_workspace}/digests/` as they land, sections committed per dimension), `references/verification.md` governs trust as material lands, and `references/synthesis.md` assembles the final report. The memlog records every source batch, load-bearing claim, plan change, and assumption — one append-only line each, always through the script (`--type <decision|source|claim|assumption|question|event>`).
## Refresh and Deepen
**Refresh.** Point at an existing run folder: read `research.md` and `.memlog.md` — never re-research from scratch. Build the refresh set from the staleness map and each claim's age against its pack window, confirm it in one exchange, re-verify just those claims, and deliver a **delta report** (confirmed / changed / overturned, new sources) appended to `research.md` with the frontmatter `updated` bumped. Claims outside the set keep their status. An overturned load-bearing claim triggers an explicit warning naming the downstream artifacts that consumed it.
**Deepen.** Drill into one dimension or add a new one without touching the rest: mini plan gate, acquire → verify for that slice only (or a drafted follow-up prompt when the user's tool is better placed), merge into `research.md`, update only the synthesis sections the new material affects — a deepening that changes no conclusion says so.
| Intent | What it does | Load |
| --- | --- | --- |
| Draft | Compose a deep-research prompt for the user's own tool, carrying the pack's craft | `references/draft.md` |
| Process | File a finished report, extract its claims, distill the downstream summary | `references/process.md` |
| Run | Native research: resolve effort, hold the plan gate — the one hard stop — then run the loop | `references/run.md`, then `references/verification.md` + `references/synthesis.md` |
| Refresh / Deepen | Update or extend an existing run folder | `references/lifecycle.md` |
## Headless Mode
@@ -120,14 +79,4 @@ When invoked headless, do not ask. Bare research defaults to **run**; a named re
}
```
Omit keys for artifacts not produced. Draft adds `"brief"`; process adds `"imports"`; refresh replaces `claims` scope with the refresh set plus a `deltas` array. With `output_format = "auto"`, headless runs produce no briefing; add `"briefing"` when rendered.
## Finalize
1. `research.md` is complete per `references/synthesis.md`: decision-first summary, findings, contrary evidence where found, recommendations with downstream bindings, source appendix, staleness map. Frontmatter metadata (`type`, `topic`, `decision`, `source`, `status`, dates) is what lets every downstream consumer trust it without reprocessing.
2. **Citation check — mechanical, never substantive.** A fresh-context subagent walks `research.md` against the digest and import files: every load-bearing claim carries a `[n]`, every `[n]` resolves in the appendix, and the cited source actually says what the text claims. It fixes citation plumbing and flags mismatches; it never rewrites findings — a claim whose source doesn't back it gets its confidence downgraded and the mismatch logged as an `event`.
3. Render per `{workflow.output_format}` (see `references/html-briefing.md`): `auto` renders the briefing page on interactive runs, skips on headless/skill-invoked; `html`/`both` always; `md` never. `research.md` always exists — the briefing is its regenerable face.
4. Polish: apply each `{workflow.doc_standards}` entry (a `skill:`, `file:`, or plain-text directive) to `research.md`.
5. Execute each `{workflow.external_handoffs}` entry (NotebookLM, Confluence, …) — invoke the named tool, surface returned URLs; skip and flag unavailable tools.
6. Tell the user what exists and where — report, briefing, imports, memlog — plus what the staleness map says to re-check and when, and that Refresh/Deepen handle it. Invoke `bmad-help` to suggest the next step.
7. Run `{workflow.on_complete}` if non-empty — a string is one instruction, an array is a sequence.
Omit keys for artifacts not produced; the `claims` counts come from `uv run scripts/recon_kit.py tally {doc_workspace}/.memlog.md`, never hand-counted. Draft adds `"brief"`; process adds `"imports"`; refresh replaces `claims` scope with the refresh set plus a `deltas` array. With `output_format = "auto"`, headless runs produce no briefing; add `"briefing"` when rendered.
+14 -7
View File
@@ -72,8 +72,8 @@ max_depth = 0
# normal spot-check load-bearing claims only — fast, the default
# high cross-check the pack's two-source classes; red-team major
# conclusions
# max cross-check every ledger claim + independent adversarial
# verifiers per conclusion + primary-source-priority ranking
# max cross-check every ledger claim + the red-team pass at full
# breadth + primary-source-priority ranking
validation = "normal"
# Red-team stance pass — fresh-context skeptics hunting disconfirming
@@ -123,12 +123,19 @@ html_theme = ""
# after", "engineering team, keep vendor marketing out".
audience = ""
# Org-configured registry of internal research tools (knowledge bases, MCP
# tools) consulted alongside web research in Run mode each entry names the
# tool and when to use it.
# Registry of extra research surfaces — internal knowledge bases or search
# tools you subscribe to — consulted alongside web research in Run mode; each
# entry names the tool and when to use it. Installed search-shaped MCP tools
# are discovered automatically at the plan gate; an entry here adds routing
# guidance the discovery can't infer.
#
# Example:
# external_sources = ["Gartner MCP (corp:gartner_query) for analyst data on enterprise software markets"]
# Examples:
# external_sources = [
# "Tavily MCP (tavily_search/tavily_extract): preferred web search + clean page extraction",
# "Perplexity Sonar MCP (perplexity_ask): cited synthesized answers — chase its citations as the sources",
# "xAI X Search MCP: live X/Twitter posts and threads, for user-voice and sentiment dimensions",
# "Gartner MCP (corp:gartner_query): analyst data on enterprise software markets",
# ]
external_sources = []
# Polish passes applied to research.md at finalize. Entries are `skill:NAME`
@@ -0,0 +1,8 @@
# Draft
Build a deep-research prompt the user runs themselves — in conversation, fast, not a project. The pack's craft travels inside the prompt so the outside tool works to this harness's standard.
1. Open the floor before any structured questions: invite the decision they're facing and anything they already have — briefs, links, a prior report, half-formed constraints — in one turn, then ask only what's still missing. Nail the **decision**, topic, and type; load the pack. Ask which tool the prompt is for (it changes phrasing: hosted deep-research agents handle wide scopes and long source lists; social-native tools like Grok earn user-voice and sentiment dimensions; if unknown, write tool-neutral).
2. Compose the prompt from the pack: the dimensions as explicit research questions pruned to the decision, the freshness bars as recency requirements, the two-source expectation for its critical claim classes, the audience, the source policy — `{workflow.preferred_sources}` named as sources to prefer, `{workflow.banned_sources}` as sources never to cite — and a **non-negotiable citation demand**: every claim with source URL and publication date, contrary evidence reported, gaps admitted rather than padded. Structure the requested output so Process can extract it cleanly (findings per dimension, a source list).
3. Bind `{doc_workspace}`: expand the folder name deterministically (`uv run scripts/recon_kit.py slug "<topic>" --type <type> --pattern "{workflow.run_folder_pattern}"` — same expansion every mode, so the report comes back to the same folder) under `{workflow.research_output_path}`, init the memlog with the decision context, save the prompt as `{doc_workspace}/brief.md`, and present it paste-ready in chat.
4. Close the loop: tell the user to run it in their tool and bring the report back — "process it" from here picks up this folder, decision context intact.
@@ -0,0 +1,11 @@
# Finalize
Every mode ends here once `research.md` is assembled.
1. `research.md` is complete per `references/synthesis.md`: decision-first summary, findings, contrary evidence where found, recommendations with downstream bindings, source appendix, staleness map. Frontmatter metadata (`type`, `topic`, `decision`, `source`, `status`, dates) is what lets every downstream consumer trust it without reprocessing.
2. **Citation check — mechanical, then semantic.** Run `uv run scripts/recon_kit.py citations {doc_workspace}/research.md` — it diffs inline `[n]` markers against the appendix and lists dangling markers and orphaned rows exactly; fix what it reports. Then a fresh-context subagent does only the judgment half: does each cited source actually say what the text claims? It never rewrites findings — a claim whose source doesn't back it gets its confidence downgraded and the mismatch logged as an `event`.
3. Render per `{workflow.output_format}` (see `references/html-briefing.md`): `auto` renders the briefing page on interactive runs, skips on headless/skill-invoked; `html`/`both` always; `md` never. `research.md` always exists — the briefing is its regenerable face.
4. Polish: apply each `{workflow.doc_standards}` entry (a `skill:`, `file:`, or plain-text directive) to `research.md`.
5. Execute each `{workflow.external_handoffs}` entry (NotebookLM, Confluence, …) — invoke the named tool, surface returned URLs; skip and flag unavailable tools.
6. Tell the user what exists and where — report, briefing, imports, memlog — plus what the staleness map says to re-check and when, and that Refresh/Deepen handle it. Invoke `bmad-help` to suggest the next step.
7. Run `{workflow.on_complete}` if non-empty — a string is one instruction, an array is a sequence.
@@ -7,7 +7,7 @@ Generate `research-briefing.html` in `{doc_workspace}` after `research.md` is fi
- **Self-contained single file**: inline CSS and JS, no external requests of any kind (no CDN, no fonts, no remote images). It must render from a `file://` open, offline, forever.
- **Structure**: a header (topic, type, decision, date, depth, verification level) → the executive summary as the opening card → sticky table of contents → dimension sections → contrary evidence (when present) → recommendations → collapsible source appendix → staleness map.
- **Confidence is visual**: every claim carries its badge — verified / medium / low / `unverified` / disputed — color-coded with the status text always present (never color alone). Unverified and disputed must be *more* prominent than verified, not less.
- **Sources are live**: inline `[n]` markers link to the appendix row; appendix rows link out to the source URL. Source URLs are untrusted content: link only `http(s)` URLs and HTML-escape them (and all source-derived text) when writing attributes — anything else renders as plain text.
- **Sources are live**: inline `[n]` markers link to the appendix row; appendix rows link out to the source URL. Source URLs are untrusted content — never hand-escape them: generate the appendix table with `uv run scripts/recon_kit.py escape-sources {doc_workspace}/research.md` and embed its `html` output, which escapes every cell, anchors each row (`id="src-n"`), and links only validated `http(s)` URLs (anything else renders as plain text; the script lists it in `invalid_urls`). Apply the same escape discipline to any other source-derived text you place in attributes.
- **Charts sparingly**: only where the data genuinely benefits (market size trajectory, decision matrix scores) — simple inline SVG, labeled axes, no library.
- **Responsive and theme-aware**: readable on a phone; respect `prefers-color-scheme` for light/dark.
@@ -0,0 +1,11 @@
# Refresh and Deepen
Lifecycle intents on an existing run folder.
## Refresh
Read `research.md` and `.memlog.md` — never re-research from scratch. Build the refresh set mechanically: assemble the claims (`claim`, `class`, `pub_date`) from the ledger, map the pack's freshness bars to a months-per-class JSON, and run `uv run scripts/recon_kit.py staleness <claims.json> --windows '<map>'` — the stale flags are the candidate set. Confirm it in one exchange, re-verify just those claims, and deliver a **delta report** (confirmed / changed / overturned, new sources) appended to `research.md` with the frontmatter `updated` bumped. Claims outside the set keep their status. An overturned load-bearing claim triggers an explicit warning naming the downstream artifacts that consumed it.
## Deepen
Drill into one dimension or add a new one without touching the rest: mini plan gate, acquire → verify for that slice only (or a drafted follow-up prompt when the user's tool is better placed), merge into `research.md`, update only the synthesis sections the new material affects — a deepening that changes no conclusion says so.
@@ -0,0 +1,10 @@
# Process
For a report the user names or drops ("there's a research report at <path>, process it"):
1. **File it.** Find or create the run folder: if a drafted brief for this topic exists, that folder is the target; otherwise infer type and topic from the report (confirm in one line), bind `{doc_workspace}` (expand the folder name with `uv run scripts/recon_kit.py slug` as in Draft), and init the memlog. Move or copy the original into `{doc_workspace}/imports/` untouched — full fidelity is preserved there, and nowhere else.
2. **Record provenance** in the memlog: what produced it (which tool or firm), when (ask if not evident — production date drives staleness), and what the user wants decided from it.
3. **Extract.** A subagent (fresh context, firewall rules) reads the import and pulls every claim bearing on the decision into digest files under `{doc_workspace}/digests/` — standard shape `{claim, source, publisher, pub_date, accessed, confidence, class}`, keeping the original's citations (the cited source is the publisher; the import is the via). Multiple imports each get their own digest; contradictions between them are findings, not noise.
4. **Check against the pack**: which of the type's dimensions the material covers, which are open, where its claims fall inside two-source classes but rest on one publisher. Verification per the resolved `validation` level (`references/verification.md`) — at `normal` this is a spot-check of the load-bearing claims only, minutes not hours.
5. **Distill** into `research.md` per `references/synthesis.md` — the succinct, cited, decision-first summary with full metadata frontmatter (topic, type, decision, `source:` provenance, dates, status). This is the artifact downstream skills read; nobody ever reprocesses the import. Open dimensions are listed honestly with a one-line route: draft a follow-up prompt, or a targeted Run on the gap.
6. Finalize per `references/finalize.md`.
@@ -1,8 +1,31 @@
# Acquisition Loop (Run)
# Run
Run once per dimension of the approved plan, in plan order. Each dimension runs in **rounds** — up to the resolved `max_depth` — and the report grows as material lands: the user watches the document build, not a spinner.
Native research, when chosen: resolve effort, hold the plan gate, then run the acquisition loop once per dimension of the approved plan, in plan order.
**Files first.** Every digest is written to `{doc_workspace}/digests/` the moment it exists — one file per assistant per round (`<dimension>-r<round>-<n>.md`), the digest shape below, raw enough to re-derive from. The conversation is a control channel; the folder is the store. Synthesis reads digest files, never conversation memory, and a run that dies mid-flight resumes from what's on disk.
## Effort
Three knobs bundled in a **preset**; any knob pins individually, and **what the user says in the request beats both**.
| Preset (`{workflow.preset}`) | subagents | sources/round | depth |
|---|---|---|---|
| `quick` | low (2) | 5 | 1 |
| `standard` (default) | normal (3) | 8 | 2 |
| `deep` | high (6) | 12 | 3 |
- **subagents** — parallel assistants: `none` (0 — inline, sequential; also the no-subagent-harness fallback), `low` (2), `normal` (3), `high` (6, cap 10 — beyond the 35 sweet spot only for genuinely wide work).
- **max_sources_per_round** — distinct sources actually read per dimension per round (cap 25).
- **max_depth** — rounds per dimension: initial pass plus lead-following follow-ups (cap 5). A cap, not a quota — dimensions stop early on coverage or novelty exhaustion.
- **validation** (orthogonal to preset, default `normal`) — rigor rises `normal` < `high` < `max`; level semantics live in `references/verification.md`. Verification happens per dimension as material lands, never as an end-of-run rewrite pass.
`{workflow.subagent_models}` is an ordered model preference for assistants — first available wins; empty means harness default. Keep the lead on the strongest model; researchers at most one tier down; judgment work never on the smallest tier.
## The plan gate
The one hard stop, kept light: decision, type and pack-derived dimensions pruned to it, shape, the **decomposition topology***breadth-first* (independent sub-questions: assistants split the dimensions), *depth-first* (one question that needs several perspectives: assistants split by angle or methodology, not by dimension), or *straightforward* (a focused ask: one assistant, a handful of calls, no fan-out — never overinvest in a simple query) — knobs in force and where each came from, which search surfaces exist (harness web search; installed search-shaped MCP tools; `{workflow.external_sources}` — check, don't assume), whether to run the fan-out as a workflow when the harness offers orchestration and `{workflow.use_workflows}` allows, and an honest time estimate (a standard run is minutes; deep runs are tens of minutes and many times the tokens).
Present as a compact checklist, get approval, then: bind `{doc_workspace}` under `{workflow.research_output_path}` — expand the folder name with `uv run scripts/recon_kit.py slug "<topic>" --type <type> --pattern "{workflow.run_folder_pattern}"` so the same topic always resolves to the same folder — seed `research.md` from `{workflow.research_template}`, init the memlog (`uv run {project-root}/_bmad/scripts/memlog.py init --workspace {doc_workspace} --field topic="<topic>" --field type="<type>" --field decision="<decision>" --field preset="<preset>"`), log the approved plan as a `decision`, and tell the user the path.
Each dimension then runs in **rounds** — up to the resolved `max_depth` — and the report grows as material lands: the user watches the document build, not a spinner. Every digest is written to `{doc_workspace}/digests/` the moment it exists — one file per assistant per round (`<dimension>-r<round>-<n>.md`), the digest shape below, raw enough to re-derive from.
## Rounds and lead-following
@@ -36,7 +59,7 @@ When workflow orchestration was approved at the plan gate, run the fan-out as a
## Source quality
One card, applied by every assistant and the lead alike. Prefer **primary sources** — filings, regulator text, official documentation, original papers, a company's own reported numbers — over aggregators and secondary reporting. Red flags that downgrade confidence on sight: speculative language ("could", "may", projections in future tense presented as findings), marketing register, passive voice with unnamed sources, cherry-picked or unsourced numbers, and aggregators recycling a single upstream report (that's one publisher, however many domains echo it). Conflicts resolve by recency, consistency with adjacent established facts, and publisher quality — never by averaging.
One card, applied by every assistant and the lead alike. Prefer **primary sources** — filings, regulator text, official documentation, original papers, a company's own reported numbers — over aggregators and secondary reporting. Red flags that downgrade confidence on sight: speculative language ("could", "may", projections in future tense presented as findings), marketing register, passive voice with unnamed sources, cherry-picked or unsourced numbers, and aggregators recycling a single upstream report (that's one publisher, however many domains echo it). Answer engines (Perplexity Sonar, Grok, and kin) are aggregators too, however good the synthesis: chase their citations and cite those, never the engine. Conflicts resolve by recency, consistency with adjacent established facts, and publisher quality — never by averaging.
## Synthesize the dimension
@@ -44,7 +67,7 @@ When a dimension's rounds are done:
1. Verify at landing per `references/verification.md` — at `normal` validation this is a spot-check of the dimension's load-bearing claims, not a sweep.
2. Write the dimension's section per the pack's skeleton from its digest files — findings woven into prose answering the dimension's questions, every load-bearing claim cited inline `[n]`, confidence flagged where below high, contradictions reported with both sides cited. Append to `research.md` and add its sources to the running source table.
3. Log one memlog line per source batch (`--type source`) and one per load-bearing claim worth tracking for refresh (`--type claim`, include class and pub date).
3. Log one memlog line per source batch (`--type source`) and one per load-bearing claim worth tracking for refresh `--type claim`, text in the machine-readable shape `ref=[n] status=<verified|unverified|disputed|overturned> class=<class> pub=<YYYY-MM> — <claim>` so `scripts/recon_kit.py tally` and `staleness` can read the ledger; a later status change is a fresh claim line with the same `ref=` (last status wins).
4. Checkpoint: one or two lines in chat — what the dimension found, anything surprising, anything unresolved. Keep moving unless the user speaks up; a mid-run scope change is logged as a `decision` and the plan adjusts. Headless: skip checkpoints entirely.
When all dimensions are done, proceed to `references/synthesis.md` for final assembly.
@@ -10,7 +10,7 @@ Assemble `research.md` in this order, shaped by `{workflow.audience}` and writte
4. **Contrary evidence** — when the red-team pass ran and found material; the strongest surviving counter-arguments, cited.
5. **Recommendations** — each bound to the decision and, where the project has them, to the downstream artifact that consumes it (per the pack's `Feeds` entries: brief section, PRD input, architecture constraint). Each recommendation names its confidence basis; a recommendation resting on low-confidence or disputed claims says so in the same sentence.
6. **Open questions** — what the research could not answer, and what it would take to answer each.
7. **Source appendix** — the numbered source table: `[n] | claim/finding it supports | publisher | pub date | accessed | confidence`. Every inline `[n]` resolves here.
8. **Staleness map** — the claims that age fastest (from their classes' freshness windows), each with its pub date and the date it should be re-checked. This is Refresh's work order; close by noting the earliest re-check date.
7. **Source appendix** — the numbered source table: `[n] | claim/finding it supports | publisher | pub date | accessed | confidence`, the publisher cell a markdown link to the source URL. Every inline `[n]` resolves here.
8. **Staleness map** — the claims that age fastest, computed not hand-derived: build the claims list (`claim`, `class`, `pub_date`) from the ledger, map the pack's freshness bars to months per class, and run `uv run scripts/recon_kit.py staleness <claims.json> --windows '<map>'` — render its re-check dates and close by noting the earliest. This is Refresh's work order.
Update the frontmatter (`status: complete`, `updated`, counts of verified/unverified claims), log a final `event` in the memlog, and return to `## Finalize` in SKILL.md.
Update the frontmatter (`status: complete`, `updated`, and the verified/unverified counts from `uv run scripts/recon_kit.py tally {doc_workspace}/.memlog.md` — never hand-counted), log a final `event` in the memlog, and proceed to `references/finalize.md`.
@@ -12,18 +12,18 @@ Per the resolved `validation` level (request > knob > default `normal`):
- **normal** — spot-check the **load-bearing claims only**: the handful per dimension the recommendation actually rests on. One independent-source check each, at landing. Everything else ships with its single source cited and confidence marked honestly. Fast by design.
- **high** — cross-check every claim in the pack's *two-source classes*, and run the red-team pass on major conclusions regardless of `{workflow.red_team}`.
- **max** — cross-check every ledger claim, an independent adversarial verifier per major conclusion (a fresh assistant prompted to refute it, blind to the supporting evidence), and primary-source-priority ranking: where a primary source (filing, regulator text, official docs, original paper) exists, secondary reporting alone does not verify.
- **max** — cross-check every ledger claim, run the red-team pass below at full breadth (every major conclusion), and primary-source-priority ranking: where a primary source (filing, regulator text, official docs, original paper) exists, secondary reporting alone does not verify.
Verifier assistants run behind the research firewall on `{workflow.subagent_models}` when set; judgment work never drops to the smallest tier.
**Independent** means a different publisher with different underlying data or reporting — not a syndication, quote, or republication of the first source, and not the same vendor's marketing in two places. An imported report counts as one publisher regardless of how many sources it cites internally; two imports from different tools agreeing is genuine confirmation, and their disagreement is a finding.
Outcomes per claim: **verified** (independent source agrees within tolerance — for quantitative claims, same order of magnitude and direction), **disputed** (independent sources materially disagree — report both figures, both cited; never average), **unverified** (no independent check within budget — the claim stays, flagged, and joins the staleness map), or **overturned** (the weight of evidence contradicts it — corrected in the text, original noted). Every status change lands in the memlog. A verification outcome adjusts status and flags — it never licenses rewriting a finding's substance beyond what the new evidence says.
Outcomes per claim: **verified** (independent source agrees within tolerance — for quantitative claims, same order of magnitude and direction), **disputed** (independent sources materially disagree — report both figures, both cited; never average), **unverified** (no independent check within budget — the claim stays, flagged, and joins the staleness map), or **overturned** (the weight of evidence contradicts it — corrected in the text, original noted). Every status change lands in the memlog as a fresh `claim` line with the same `ref=` and the new status — last status wins, which is how `scripts/recon_kit.py tally` reads the ledger. A verification outcome adjusts status and flags — it never licenses rewriting a finding's substance beyond what the new evidence says.
Confidence rendered in the report: **high** (verified, fresh, credible publishers), **medium** (single credible source, fresh), **low** (stale, weak publisher, or disputed) — plus the explicit `unverified` flag. Confidence is per-claim, never per-section.
## Red-team pass
Off by default (`{workflow.red_team}` = `"off"`; `"offer"` proposes it at the plan gate, `"on"` always runs; `high`/`max` validation includes it for major conclusions). When it runs: for each major conclusion, a **fresh-context** skeptic subagent — the conclusion and a search budget, no supporting evidence, no run context — hunts for disconfirming evidence: the bear case, failed attempts, contrary data, the strongest good-faith argument the conclusion is wrong.
The single adversarial mechanism — no other verifier duplicates it. Off by default (`{workflow.red_team}` = `"off"`; `"offer"` proposes it at the plan gate, `"on"` always runs; `high` validation includes it for major conclusions, `max` runs it at full breadth). When it runs: for each major conclusion, a **fresh-context** skeptic subagent — the conclusion and a search budget, no supporting evidence, no run context — hunts for disconfirming evidence: the bear case, failed attempts, contrary data, the strongest good-faith argument the conclusion is wrong.
What comes back is weighed, not appended: a conclusion that survives gets its strongest counter-argument acknowledged in the synthesis; one that doesn't is revised before the report states it. Material findings land in a **Contrary Evidence** section with full citation discipline. Zero findings after a real search is itself reportable — say what was searched for and not found.
@@ -0,0 +1,322 @@
#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# ///
"""recon_kit — deterministic helpers for bmad-deep-recon.
The mechanical half of the research workflow: everything here is exact,
repeatable work the LLM should never re-derive by hand. All subcommands
print one JSON object to stdout; diagnostics go to stderr. Exit codes:
0 = pass, 1 = findings that need attention, 2 = usage/parse error.
Subcommands:
citations RESEARCH_MD
Cross-check inline [n] markers against the source-appendix table:
dangling markers (no appendix row) and orphaned rows (never cited).
tally MEMLOG_MD
Count memlog entries by type, and claim entries by status.
Claim lines carry `status=<word>` and optionally `ref=[n]`; for a
given ref the LAST status wins, so status changes are appends.
staleness CLAIMS_JSON --windows JSON [--today YYYY-MM-DD]
Given claims [{claim, class, pub_date}] and a months-per-class map
(e.g. '{"size/growth": 18, "pricing": 3}'), compute each claim's
re-check date, flag stale ones, and report the earliest re-check.
slug TOPIC --type TYPE [--pattern P] [--date YYYY-MM-DD]
Expand the run-folder pattern deterministically so the same topic
always lands in the same folder across draft -> process -> refresh.
escape-sources RESEARCH_MD
Emit the source-appendix table as HTML with every cell escaped and
only validated http(s) URLs turned into links, for the briefing.
"""
from __future__ import annotations
import argparse
import calendar
import html
import json
import re
import sys
import unicodedata
from datetime import date, datetime
from pathlib import Path
from urllib.parse import urlparse
MARKER_RE = re.compile(r"\[(\d+)\](?!\()") # [3] but not a [3](url) link
MD_LINK_RE = re.compile(r"\[([^\]]*)\]\((\S+?)\)")
BARE_URL_RE = re.compile(r"https?://[^\s|)\]]+")
def out(payload: dict, exit_code: int) -> int:
print(json.dumps(payload, indent=2, ensure_ascii=False, default=str))
return exit_code
def read_text(path_arg: str) -> str:
if path_arg == "-":
return sys.stdin.read()
return Path(path_arg).read_text(encoding="utf-8")
def strip_fences(text: str) -> str:
"""Blank out fenced code blocks so their contents never count as markers or rows."""
lines, fenced = [], False
for ln in text.splitlines():
if ln.lstrip().startswith("```"):
fenced = not fenced
lines.append("")
continue
lines.append("" if fenced else ln)
return "\n".join(lines)
def table_cells(line: str) -> list[str]:
return [c.strip() for c in line.strip().strip("|").split("|")]
def appendix_rows(text: str) -> dict[int, list[str]]:
"""Source-appendix rows: markdown table rows whose first cell is a bare [n] / n."""
rows: dict[int, list[str]] = {}
for ln in text.splitlines():
stripped = ln.strip()
if not stripped.startswith("|"):
continue
cells = table_cells(stripped)
if not cells or len(cells) < 2:
continue
m = re.fullmatch(r"\[?(\d+)\]?", cells[0])
if m:
rows[int(m.group(1))] = cells
return rows
# --- citations ---------------------------------------------------------------
def cmd_citations(args) -> int:
text = strip_fences(read_text(args.file))
rows = appendix_rows(text)
markers: set[int] = set()
for ln in text.splitlines():
stripped = ln.strip()
if stripped.startswith("|"):
cells = table_cells(stripped)
if cells and re.fullmatch(r"\[?(\d+)\]?", cells[0]):
continue # an appendix row is not a citation of itself
markers.update(int(n) for n in MARKER_RE.findall(ln))
dangling = sorted(markers - set(rows))
orphaned = sorted(set(rows) - markers)
ok = not dangling and not orphaned
return out({
"markers": sorted(markers),
"appendix_rows": sorted(rows),
"dangling_markers": dangling,
"orphaned_rows": orphaned,
"ok": ok,
}, 0 if ok else 1)
# --- tally -------------------------------------------------------------------
ENTRY_RE = re.compile(r"^- (?:\(([\w-]+)(?: by [^)]*)?\)\s*)?(.*)$")
def cmd_tally(args) -> int:
text = read_text(args.file)
body = text.split("---", 2)[-1] if text.startswith("---") else text
by_type: dict[str, int] = {}
by_ref: dict[int, str] = {}
unref_status: dict[str, int] = {}
entries = 0
for ln in body.splitlines():
m = ENTRY_RE.match(ln)
if not m or not ln.startswith("- "):
continue
entries += 1
etype = m.group(1) or "note"
by_type[etype] = by_type.get(etype, 0) + 1
if etype == "claim":
status_m = re.search(r"status=([\w-]+)", m.group(2))
status = status_m.group(1) if status_m else "unknown"
ref_m = re.search(r"ref=\[?(\d+)\]?", m.group(2))
if ref_m:
by_ref[int(ref_m.group(1))] = status # last status wins per ref
else:
unref_status[status] = unref_status.get(status, 0) + 1
claims: dict[str, int] = dict(unref_status)
for status in by_ref.values():
claims[status] = claims.get(status, 0) + 1
return out({
"entries": entries,
"by_type": dict(sorted(by_type.items())),
"claims": dict(sorted(claims.items())),
"claims_total": sum(claims.values()),
}, 0)
# --- staleness ---------------------------------------------------------------
def parse_date(raw: str) -> date:
raw = raw.strip()
for fmt in ("%Y-%m-%d", "%Y-%m", "%Y"):
try:
return datetime.strptime(raw, fmt).date()
except ValueError:
continue
raise ValueError(f"unparseable date: {raw!r} (want YYYY[-MM[-DD]])")
def add_months(d: date, months: int) -> date:
total = d.month - 1 + months
year, month = d.year + total // 12, total % 12 + 1
return date(year, month, min(d.day, calendar.monthrange(year, month)[1]))
def cmd_staleness(args) -> int:
try:
payload = json.loads(read_text(args.file))
windows = {k.lower(): int(v) for k, v in json.loads(args.windows).items()}
today = parse_date(args.today) if args.today else date.today()
except (ValueError, json.JSONDecodeError) as e:
print(f"error: {e}", file=sys.stderr)
return 2
claims = payload["claims"] if isinstance(payload, dict) else payload
results, no_window, stale_count = [], set(), 0
earliest: date | None = None
for c in claims:
cls = str(c.get("class", "")).lower()
try:
pub = parse_date(str(c["pub_date"]))
except (KeyError, ValueError) as e:
print(f"error in claim {c!r}: {e}", file=sys.stderr)
return 2
months = windows.get(cls)
if months is None:
no_window.add(cls)
results.append({**c, "recheck": None, "stale": None})
continue
recheck = add_months(pub, months)
stale = recheck <= today
stale_count += stale
earliest = recheck if earliest is None or recheck < earliest else earliest
results.append({**c, "recheck": recheck.isoformat(), "stale": stale})
return out({
"today": today.isoformat(),
"claims": results,
"stale_count": stale_count,
"earliest_recheck": earliest.isoformat() if earliest else None,
"no_window_classes": sorted(no_window),
}, 1 if stale_count else 0)
# --- slug --------------------------------------------------------------------
def slugify(text: str, max_len: int = 40) -> str:
text = unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode()
text = re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-")
return re.sub(r"-{2,}", "-", text)[:max_len].rstrip("-")
def cmd_slug(args) -> int:
slug = slugify(args.topic)
if not slug:
print("error: topic slugified to an empty string", file=sys.stderr)
return 2
folder = (args.pattern
.replace("{research_type}", args.type)
.replace("{topic_slug}", slug)
.replace("{date}", args.date or date.today().isoformat()))
return out({"topic_slug": slug, "folder": folder}, 0)
# --- escape-sources ----------------------------------------------------------
def safe_url(raw: str) -> str | None:
parsed = urlparse(raw)
return raw if parsed.scheme in ("http", "https") and parsed.netloc else None
def cell_html(cell: str, invalid: list[str]) -> str:
"""Escape a cell; a markdown link or bare URL becomes an <a> only when http(s)."""
link = MD_LINK_RE.search(cell)
if link:
url = safe_url(link.group(2))
label = html.escape(link.group(1) or link.group(2))
if url:
return html.escape(cell[:link.start()]) + \
f'<a href="{html.escape(url, quote=True)}" target="_blank" rel="noopener">{label}</a>' + \
html.escape(cell[link.end():])
invalid.append(link.group(2))
return html.escape(cell.replace(link.group(0), link.group(1) or link.group(2)))
bare = BARE_URL_RE.search(cell)
if bare:
url = safe_url(bare.group(0))
if url:
escaped = html.escape(url, quote=True)
return html.escape(cell[:bare.start()]) + \
f'<a href="{escaped}" target="_blank" rel="noopener">{escaped}</a>' + \
html.escape(cell[bare.end():])
invalid.append(bare.group(0))
return html.escape(cell)
def cmd_escape_sources(args) -> int:
text = strip_fences(read_text(args.file))
rows = appendix_rows(text)
if not rows:
print("error: no source-appendix table rows found", file=sys.stderr)
return 2
invalid: list[str] = []
body_rows = []
for n in sorted(rows):
cells = rows[n]
tds = "".join(f"<td>{cell_html(c, invalid)}</td>" for c in cells[1:])
body_rows.append(f'<tr id="src-{n}"><td>[{n}]</td>{tds}</tr>')
table = ('<table class="sources"><tbody>' + "".join(body_rows) + "</tbody></table>")
return out({"rows": len(rows), "invalid_urls": invalid, "html": table},
1 if invalid else 0)
# --- entry point -------------------------------------------------------------
def main(argv: list[str] | None = None) -> int:
p = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
sub = p.add_subparsers(dest="cmd", required=True)
pc = sub.add_parser("citations", help="cross-check [n] markers vs the source appendix")
pc.add_argument("file", help="path to research.md (or - for stdin)")
pc.set_defaults(func=cmd_citations)
pt = sub.add_parser("tally", help="count memlog entries by type and claims by status")
pt.add_argument("file", help="path to .memlog.md (or - for stdin)")
pt.set_defaults(func=cmd_tally)
ps = sub.add_parser("staleness", help="compute re-check dates from freshness windows")
ps.add_argument("file", help="claims JSON: [{claim, class, pub_date}] (or - for stdin)")
ps.add_argument("--windows", required=True,
help='JSON months-per-class map, e.g. \'{"pricing": 3}\'')
ps.add_argument("--today", help="override today's date (YYYY-MM-DD)")
ps.set_defaults(func=cmd_staleness)
pg = sub.add_parser("slug", help="expand the run-folder pattern deterministically")
pg.add_argument("topic", help="research topic text")
pg.add_argument("--type", required=True, help="research type code (e.g. market)")
pg.add_argument("--pattern", default="{research_type}-{topic_slug}-{date}",
help="folder pattern (default: {research_type}-{topic_slug}-{date})")
pg.add_argument("--date", help="override date (YYYY-MM-DD; default today)")
pg.set_defaults(func=cmd_slug)
pe = sub.add_parser("escape-sources",
help="source appendix as escaped HTML with validated links")
pe.add_argument("file", help="path to research.md (or - for stdin)")
pe.set_defaults(func=cmd_escape_sources)
args = p.parse_args(argv)
try:
return args.func(args)
except FileNotFoundError as e:
print(f"error: {e}", file=sys.stderr)
return 2
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,144 @@
#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# ///
"""Tests for recon_kit.py."""
import io
import json
import sys
import unittest
from contextlib import redirect_stdout
from datetime import date
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from recon_kit import add_months, main, parse_date, slugify
REPORT = """---
title: 'market research: widgets'
---
# Report
The market is growing fast [1] and pricing clusters at $20 [2].
An uncited assertion sits here [4].
```
[9] inside a fence never counts
```
## Source appendix
| [n] | Supports | Publisher | Pub date | Accessed | Confidence |
| --- | --- | --- | --- | --- | --- |
| [1] | market growth | [Gartner](https://example.com/g) | 2026-01 | 2026-07-01 | high |
| [2] | pricing | [Acme](javascript:alert(1)) | 2026-05 | 2026-07-01 | medium |
| [3] | never cited | https://example.com/x | 2025-11 | 2026-07-01 | low |
"""
MEMLOG = """---
topic: widgets
updated: 2026-07-22T10:00
---
- (decision) plan approved
- (source) round 1 batch: 4 sources
- (claim) ref=[1] status=verified class=size/growth pub=2026-01 — market growing 12% CAGR
- (claim) ref=[2] status=unverified class=pricing pub=2026-05 — pricing clusters at $20
- (claim) ref=[2] status=verified class=pricing pub=2026-05 — confirmed by second source
- (claim) status=unverified class=behavior pub=2025-03 — users churn at day 8
- (event) dimension 1 complete
"""
def run(argv):
buf = io.StringIO()
with redirect_stdout(buf):
code = main(argv)
return code, json.loads(buf.getvalue())
class CitationsTest(unittest.TestCase):
def test_cross_check(self):
report = Path(__file__).parent / "_report.md"
report.write_text(REPORT, encoding="utf-8")
try:
code, result = run(["citations", str(report)])
finally:
report.unlink()
self.assertEqual(result["dangling_markers"], [4])
self.assertEqual(result["orphaned_rows"], [3])
self.assertNotIn(9, result["markers"]) # fenced content ignored
self.assertEqual(code, 1)
class TallyTest(unittest.TestCase):
def test_last_status_wins_per_ref(self):
log = Path(__file__).parent / "_memlog.md"
log.write_text(MEMLOG, encoding="utf-8")
try:
code, result = run(["tally", str(log)])
finally:
log.unlink()
self.assertEqual(result["by_type"]["claim"], 4)
self.assertEqual(result["claims"], {"unverified": 1, "verified": 2})
self.assertEqual(result["claims_total"], 3) # ref=[2] counted once
self.assertEqual(code, 0)
class StalenessTest(unittest.TestCase):
def test_dates(self):
self.assertEqual(parse_date("2026-01"), date(2026, 1, 1))
self.assertEqual(add_months(date(2026, 1, 31), 1), date(2026, 2, 28))
def test_windows(self):
claims = json.dumps([
{"claim": "sizing", "class": "size/growth", "pub_date": "2024-06"},
{"claim": "pricing", "class": "pricing", "pub_date": "2026-06"},
{"claim": "odd", "class": "unmapped", "pub_date": "2026-06"},
])
f = Path(__file__).parent / "_claims.json"
f.write_text(claims, encoding="utf-8")
try:
code, result = run([
"staleness", str(f),
"--windows", '{"size/growth": 18, "pricing": 3}',
"--today", "2026-07-22",
])
finally:
f.unlink()
self.assertEqual(result["stale_count"], 1) # sizing recheck 2025-12 < today
self.assertEqual(result["earliest_recheck"], "2025-12-01")
self.assertEqual(result["no_window_classes"], ["unmapped"])
self.assertEqual(code, 1)
class SlugTest(unittest.TestCase):
def test_deterministic_folder(self):
self.assertEqual(slugify("Créme Brûlée: AI Tools!"), "creme-brulee-ai-tools")
code, result = run(["slug", "SMB Accounting SaaS", "--type", "market",
"--date", "2026-07-22"])
self.assertEqual(result["folder"], "market-smb-accounting-saas-2026-07-22")
self.assertEqual(code, 0)
class EscapeSourcesTest(unittest.TestCase):
def test_escaping_and_url_validation(self):
report = Path(__file__).parent / "_report.md"
report.write_text(REPORT, encoding="utf-8")
try:
code, result = run(["escape-sources", str(report)])
finally:
report.unlink()
self.assertEqual(result["rows"], 3)
self.assertTrue(any(u.startswith("javascript:") for u in result["invalid_urls"]))
self.assertNotIn("javascript:", result["html"]) # never linked
self.assertIn('href="https://example.com/g"', result["html"])
self.assertIn('id="src-1"', result["html"])
self.assertEqual(code, 1)
if __name__ == "__main__":
unittest.main()