v2.2.2 feat: add domain-scoped transcript trap demotion (#374)

Recovered and verified transcript-fixer v1.31.0 from local-only main state.
This commit is contained in:
daymade
2026-08-28 15:21:21 +08:00
committed by GitHub
parent 26f125f672
commit f20dcf2a3c
20 changed files with 830 additions and 27 deletions
+2 -2
View File
@@ -6,7 +6,7 @@
},
"metadata": {
"description": "Professional Claude Code and Codex skills marketplace — production-ready skills spanning GitHub and git operations (including current-base contributor PR review), document conversion and generation (Markdown, PDF, PPTX, DOCX), diagram and UI-design extraction, the full audio pipeline (ASR transcription, TTS, transcript correction, meeting minutes), financial and investment-research data, web scraping and content capture, security/PII tooling and secure repomix packaging, macOS and iOS development, CLI demo and terminal automation, prompt and skill engineering, deep research and fact-checking, QA and LLM-evaluation infrastructure, internationalization, network/Tailscale and remote-desktop diagnostics, Claude Code operations, and model-aware Codex workstation setup. Suite plugins bundle related skills under shared namespaces, including the StepFun StepAudio 2.5 audio family. See the Available Skills list in the README for the authoritative per-skill breakdown.",
"version": "2.2.1"
"version": "2.2.2"
},
"plugins": [
{
@@ -275,7 +275,7 @@
"description": "Audio processing suite covering the full speech pipeline: ASR transcription (Qwen3, StepFun) with batch-mode guards against music-only repetition-loop hallucinations, speaker diarization and CAM++ voiceprint identification for multi-speaker recordings, transcript error correction, structured meeting minutes generation, and TTS voice synthesis (StepFun). Install once for the complete audio workflow.",
"source": "./daymade-audio",
"strict": false,
"version": "1.30.0",
"version": "1.31.0",
"category": "suite",
"keywords": [
"suite",
+1
View File
@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- **Local history skills** (`daymade-claude-code` v2.0.0; marketplace v2.0.0): replace three overlapping/ambiguous entry points with four provider-and-action-specific Skills: `claude-code-history-files-finder``read-claude-code-history`, `local-conversation-history``read-codex-history`, `continue-claude-work``continue-claude-code-work`, while `continue-codex-work` keeps its name. The two readers now own all parsing and remain evidence-only; the two continuation Skills consume verified read receipts, rebuild the original business outcome / unfulfilled requests / user corrections / proven prior assets, and only then execute a next action that directly reduces the outstanding result. `read-codex-history` keeps prompt ledger, state DB and rollout JSONL as separate evidence surfaces, adds exact raw-input tables, Codex-only search, verified selected identity and exact fork/compaction lineage, and rejects fused rollouts containing multiple `session_meta` IDs. `read-claude-code-history` adds recent inventory, chronological Session evidence with queued human prompts in place, original-word export grouped by Session, full-event search, hybrid recall, triage and deleted-file recovery. Exact readers now parse every physical record rather than a resume-oriented tail, retain every selected human and assistant text turn so a middle successful asset cannot disappear, cover registered Claude archives, require one matching record-level identity instead of trusting a filename, reject fused identities and divergent physical copies, choose a strict append-only live/archive superset without traversal-order last-wins, skip healthy blank separators, and fail visibly on malformed JSONL. The parser implementation and format references moved from the continuation bundles into their owning read bundles; all old runtime instructions remain directly reachable in migration references, including the legacy Kimi branch, rather than being silently deleted. Existing users should run `claude plugin marketplace update daymade-skills`, update/reinstall `daymade-claude-code@daymade-skills`, and replace old slash invocations with the mapping above. Verification: strict validation for all four Skills; Claude-reader and Codex-reader regression counts are recorded by the current test run; a real Claude active/archive replay retains the pre-compaction chronology, one real Codex continuation replay recovers the corrected second-attachment referent, and a separate misbound Session replay fails explicitly as a fused two-identity rollout instead of attributing another Session's events to it.
### Fixed
- **transcript-fixer** (`daymade-audio` v1.31.0; marketplace v2.2.2): make domain-context vetoes executable at Stage 1 and give context rules a managed, domain-scoped home. A trap annotated `禁裸词`/`禁入词典` or a confirmed-correct (勿修) record in `~/.transcript-fixer/contexts/<domain>.md` now demotes any dictionary rule with the same FROM to safe-mode deferral — beating `--apply-domain` trust-flattening, so a real-word rule that is right in some contexts and wrong in others (the 绿点→绿电 class) can stay in the dictionary without firing blindly, instead of forcing an all-or-nothing `--report-false-positive`; `--apply-all` remains the explicit override, demotions print on stderr, and `--add` warns at write time when the new rule is veto-marked. Context rules gain a `domain` column (schema v2.4, NULL = global; additive migration whose backward path drops the index before the column) and their first management CLI, `--add-context-rule` / `--list-context-rules` (duplicate patterns, invalid regex, list-form write domains, and un-migrated databases all fail loud with the real direction), with Stage 1 loading global rules plus the active domain's. Context rules are exempt from trap demotion — the pattern is its own context, so a lookahead whose match text equals a demoted FROM no longer inherits the deferral. Docs scope the veto honestly (only when the domain is named; by FROM across a multi-domain union) and spell out the two marker authoring constraints (single-line, never discuss the marker word on a trap line). An independent fresh-context review surfaced 9 findings (2 major), all fixed or documented; 631 tests pass including 25 new ones, and a real replay of the motivating 84-minute transcript now defers 绿点 while still applying every correct trusted rule. Packaging cleanup also replaces scanner-confusing public path/remote examples and percent-encodes the inline SVG namespace without changing its decoded value.
- **marketplace documentation SSOT** (`daymade-claude-code` v2.1.1; marketplace v2.2.1): stop persisting catalog positions, marketplace-version badges, and a complete Skill snapshot in `CLAUDE.md`. The manifest remains the only authority for names, versions, sources, and suite membership; the READMEs keep unnumbered human-facing capability sections, and `CLAUDE.md` keeps only the stable authority pointer. The detailed new-Skill SOP now says the same thing and removes stale `.zip`, relative-path, and quiet-mtime takeover instructions. The drift checker rejects numbered capability headings, copied version badges, and model-loaded snapshots; focused positive and negative controls prove both its healthy and failure branches against the real repository.
- **prior-work-retrieval** (`daymade-claude-code` v2.1.0): stops retrieval from replacing a read-only maintenance outcome with its own receipt workflow. The catalog description now front-loads the read-only exclusion so it survives Codex's shortened Skill-description budget; negated phrases such as “不要修改文件” cannot be misread as a positive production verb; `Stop` validates only an already-existing requirement and never creates one from final-answer length. Required-source fingerprints keep receipts valid across unrelated optional-carrier edits, while required carrier changes still invalidate them. Filesystem content search no longer unconditionally enumerates every path; path scans require an explicit filename/path/date term. Forty focused regressions pass. A fresh read-only Huawei repo task then completed with one direct inspection and a seven-line business answer—no Prior Work, Goal, agent, meeting sync, or write.
- **claude-md-progressive-disclosurer** (`daymade-claude-code` v2.1.0): profiles the complete model-visible startup surface with `codex debug prompt-input` instead of assuming one historical `project_doc_max_bytes` behavior. It distinguishes the global user-instruction path from the cumulative project-doc budget, audits truncated Skill catalog descriptions and hook self-trigger loops, and adds current Claude `/memory`, `InstructionsLoaded`, `claudeMdExcludes`, and path-rule carriers. The zero-loss reviewer gate is bounded to one fresh-context reviewer by default; reviewer count and recursive re-review are no longer accepted as quality proxies.
@@ -1,4 +1,4 @@
Security scan passed
Scanned at: 2026-08-20T18:25:55.189854+00:00
Scanned at: 2026-08-28T07:07:47.448179+00:00
Tool: gitleaks + pattern-based validation
Content hash: 6f67af7bdb47d657eb54c96e198ba3c95589923d12594a1be465cc3bb133c054
Content hash: 876e28fcfb21619b92296d065afa9dbc08015f111a9a8451f5b4916b71fa0e65
+2 -1
View File
@@ -154,12 +154,13 @@ Read [references/false_positive_guide.md](references/false_positive_guide.md) an
|---|---|
| Stable non-word or unique garble → canonical term | `--add ... --domain <project>` |
| Important recurring person and observed ASR variants | People roster |
| Correction right only inside a specific recurring phrase | `--add-context-rule PATTERN REPLACEMENT --domain <project>` (regex, domain-scoped; omit `--domain` for global) |
| Common/real word wrong only under a cue | Domain context trap, never a bare rule |
| Real name → different real name | Domain context + human/audio verification, never a bare rule |
| Confirmed-correct entity repeatedly reopened | Confirmed-correct context record |
| One-off sentence-local wording | Edit only; do not add |
A context trap is a cue, not permission to replace blindly. `--scan-traps` supports canonical `→` and legacy `≈` mappings with the same directional contract: left is observed ASR, right is intended text. Wrap an exact FROM phrase containing spaces in backticks:
A context trap is a cue, not permission to replace blindly. Two annotation classes in a domain context file are **machine-readable vetoes that Stage 1 enforces** (when the domain is named via `--domain` — a whole-library run has no owner to veto with): a trap marked `禁裸词`/`禁入词典` demotes any dictionary rule with the same FROM to review, and a confirmed-correct (勿修) record demotes any rule whose FROM is that token — demotion beats `--apply-domain` trust-flattening, so a real-word rule (the 绿点→绿电 class: right in business context, wrong in UI context) can stay in the dictionary without firing blindly. `--apply-all` remains the operator's explicit override. Without the veto the only escape was `--report-false-positive`, which disables the rule in the contexts where it is right too. `--scan-traps` supports canonical `→` and legacy `≈` mappings with the same directional contract: left is observed ASR, right is intended text. Wrap an exact FROM phrase containing spaces in backticks:
~~~markdown
- **`CC 思维链`/`CC 思维连` → 目标术语** — only under the domain's documented cue
@@ -251,6 +251,8 @@ What must NOT go in a context file: hard replacement rules. `减→剪` as a rul
Maintenance loop (mirrors the dictionary's `--add` habit): when a native session surfaces a **context-dependent** recurring error — you fixed it here, and it'll recur in this domain's future transcripts — append it to the domain's context file with its disambiguating cue. Deterministic non-word/name fixes keep going to `--add --domain` / the roster as before.
**Machine-readable vetoes (consumed by Stage 1, since 2026-08):** two annotation classes in the context file act on the *dictionary*, not just on the reader. ① A trap whose bullet carries the literal marker `禁裸词` or `禁入词典` (**妙计 → 妙记(飞书妙记语境,禁裸词)**) demotes any dictionary rule with the same FROM to safe-mode deferral — the pair may only be corrected with its context judged, exactly the 绿点→绿电 class (a real-word rule right in some contexts, wrong in others). ② A confirmed-correct record (**X = 真实实体,勿修**) demotes any rule whose FROM is that token. Demotion beats `--apply-domain` trust-flattening; `--apply-all` remains the operator's explicit override. Two scope rules: the veto only fires when the domain is named via `--domain` (a whole-library run has no owner to veto with), and in a multi-domain union it applies by FROM across the whole union regardless of which named domain owns the rule. This is how a trap documented *after* a rule was added still stops the rule from firing — before this layer, only `--report-false-positive` (which disables the rule everywhere, including the contexts where it is right) could.
Format and a worked template: [domain_context_guide.md](domain_context_guide.md).
Note: contexts are consumed by the **native workflow** (the agent reads the file — no code involved). API mode (`--stage 2/3`, the backup channel) does not inject them yet; if that channel gets completed, the same files should feed its prompt.
Note: the cue prose itself is consumed by the **native workflow** (the agent reads the file); API mode (`--stage 2/3`, the backup channel) does not inject contexts yet; Stage 1 consumes only the two veto classes above, never the prose.
@@ -60,6 +60,20 @@ For each trap:
**`CC 思维链` → 目标术语**. The spaces are part of the literal scan target;
replace the synthetic target with the domain's intended term.
- Order by frequency; prune entries that stop recurring.
- **Machine-readable vetoes (Stage 1 reads these):** add the literal marker
`禁裸词` or `禁入词典` to a trap's annotation (**妙计 → 妙记(…,禁裸词)**)
and any *dictionary rule with the same FROM* defers to review at Stage 1
instead of auto-applying — use it the moment a real-word pair proves
context-dependent, so the rule can stay in the dictionary (right in some
contexts) without firing blindly everywhere. A confirmed-correct record
(**X = 真实实体,勿修**) demotes rules whose FROM is X. Demotion beats
`--apply-domain` trust; `--apply-all` still applies by explicit override.
Two authoring constraints, both from the matcher being a per-bullet-line
literal substring check: write the marker on the trap's own line (a marker
on an indented continuation line is silently missed), and never *discuss*
the marker word in a trap line's prose (「评审后决定不标禁裸词」 still
triggers the demotion it negates) — debate the marking decision anywhere
else in the file, just not on a trap bullet line.
## Authoritative name sources (pointers, not copies)
This is a **manifest to be read in full**, not a single hint. A project's people
@@ -43,7 +43,7 @@ uv run scripts/fix_transcription.py --input <file> --stage <1|2|3> [--output <di
- `2` = Stage 1 dictionary pass followed by API AI correction
- `3` = Stage 1 + API AI correction + diff report
- `--output, -o` (optional): Where results are written — accepts either a **directory** (the sidecars `<stem>_stage1.md` / `_changes.md` / `_needs_review.md` are written into it) **or a file path** ending in `.md`/`.markdown`/`.txt` that is not an existing directory (the corrected Stage 1 output is written directly to that exact file). Defaults to the input file's directory. Every "Saved" / report line prints the full resolved path, so a misdirected output is visible immediately. (Passing a file path used to silently `mkdir` a directory of that name and hide the output inside it — fixed.)
- `--domain, -d` (optional): Restrict to one correction domain (default: all domains). Accepts a comma-separated list (`--domain myproject,myproject-alt`): every listed domain's rules load as one union for Stage 1, and `--apply-domain` trusts the whole union. Write commands (`--add`, `--approve`) still require exactly one domain and fail fast on a list.
- `--domain, -d` (optional): Restrict to one correction domain (default: all domains). Accepts a comma-separated list (`--domain myproject,myproject-alt`): every listed domain's rules load as one union for Stage 1, and `--apply-domain` trusts the whole union. Write commands (`--add`, `--approve`, `--add-context-rule`) still require exactly one domain and fail fast on a list. Domain naming also scopes two more things at Stage 1: context rules (global rules always load; a domain-named rule loads only when its domain is active) and **trap demotion** — when `~/.transcript-fixer/contexts/<domain>.md` exists, traps annotated `禁裸词`/`禁入词典` and confirmed-correct (勿修) records demote any same-FROM dictionary rule to safe-mode deferral (beating `--apply-domain` trust-flattening; `--apply-all` still overrides). Without `--domain` no veto fires (a whole-library run has no owner to veto with); in a multi-domain union the veto applies by FROM across the whole union, regardless of which named domain owns the rule or the context file. Demotions are printed on stderr (`🛡️ Trap demotion: N rule(s)…`), never silent.
- `--apply-all` (optional): Opt out of the default safe mode and apply every risk level (low/medium/high). Higher false-positive risk — see false_positive_guide.md.
- `--review` (deprecated): No-op kept for backward compatibility; safe mode is now the default.
- `--dry-run` (optional): Preview Stage 1 changes to `*_dryrun.md` without writing `*_stage1.md`.
@@ -57,6 +57,12 @@ uv run scripts/fix_transcription.py --input <file> --stage <1|2|3> [--output <di
- `--check-corpus` (with `--add`): run the same probe on the FROM term before the rule is written; advisory, never blocks. Requires `--corpus`
- `--json` works with both: one machine-readable result line on stdout
**Context rules** (regex rules applied before the dictionary at Stage 1; domain-scoped since schema v2.4 — run pending migrations first or `--add-context-rule` fails with the direction):
- `--add-context-rule PATTERN REPLACEMENT [--domain <project>] [--description TEXT] [--priority N]`: add a context-aware regex rule. PATTERN is a Python regex (validated at write time); omitting `--domain` writes a global rule that applies to every domain. Duplicate patterns and un-migrated databases are hard errors. Use it for corrections that are right only inside a specific recurring phrase — the middle channel between a bare dictionary rule and a prose context trap.
- `--list-context-rules [--domain <project>] [--all] [--json]`: list rules; `--domain` shows global rules plus that domain's, `--all` includes disabled ones.
- On a database not yet migrated to v2.4 there is no `domain` column, so by construction every rule is global: loading keeps the legacy behavior and never crashes; only `--add-context-rule` refuses, naming the migration.
**Review queue** (persistent store for uncertain corrections; semantics in [review_queue_dashboard.md](review_queue_dashboard.md)):
- `--enqueue-review JSON_PATH`: Enqueue items from a JSON file (`-` = stdin). Item fields: `{original, suggested?, file?, line?, context?, kind?, domain?, evidence?, actions?, priority?, source?}` — full field/alias table + gotchas in [Review Queue Item Schema](#review-queue-item-schema) below
@@ -107,14 +107,14 @@ mkdir ~/transcript-corrections && cd ~/transcript-corrections
mv ~/path/to/team_corrections.json .
git init && git add team_corrections.json
git commit -m "Initial correction export"
git remote add origin git@github.com:org/transcript-corrections.git
git remote add origin https://github.com/org/transcript-corrections.git
git push -u origin main
```
### Team Members Clone + Import
```bash
git clone git@github.com:org/transcript-corrections.git ~/transcript-corrections
git clone https://github.com/org/transcript-corrections.git ~/transcript-corrections
uv run scripts/fix_transcription.py --import ~/transcript-corrections/team_corrections.json --merge
# --merge combines with each person's existing local corrections;
# --import without --merge overwrites (dangerous).
@@ -296,10 +296,10 @@ nano team_corrections.json
git remote -v
# Re-add if needed
git remote set-url origin git@github.com:org/corrections.git
git remote set-url origin https://github.com/org/corrections.git
# Verify SSH keys
ssh -T git@github.com
# Verify the configured remote is reachable
git ls-remote origin
```
### Merge Conflicts Too Complex
@@ -9,6 +9,8 @@ This module contains command handlers and argument parsing:
from .commands import (
cmd_init,
cmd_add_correction,
cmd_add_context_rule,
cmd_list_context_rules,
cmd_audit,
cmd_list_corrections,
cmd_export_corrections,
@@ -38,6 +40,8 @@ from .argument_parser import create_argument_parser
__all__ = [
'cmd_init',
'cmd_add_correction',
'cmd_add_context_rule',
'cmd_list_context_rules',
'cmd_audit',
'cmd_list_corrections',
'cmd_export_corrections',
@@ -37,6 +37,38 @@ def create_argument_parser() -> argparse.ArgumentParser:
dest="add_correction",
help="Add correction"
)
parser.add_argument(
"--add-context-rule",
nargs=2,
metavar=("PATTERN", "REPLACEMENT"),
dest="add_context_rule",
help="Add a context-aware regex rule (PATTERN is a regex; --domain scopes it, omit for global)"
)
parser.add_argument(
"--list-context-rules",
action="store_true",
dest="list_context_rules",
help="List context rules (--domain filters to one domain plus globals)"
)
parser.add_argument(
"--all",
action="store_true",
dest="all",
help="With --list-context-rules: include disabled rules"
)
parser.add_argument(
"--description",
metavar="TEXT",
default=None,
help="Human-readable rule name for --add-context-rule"
)
parser.add_argument(
"--priority",
metavar="N",
type=int,
default=0,
help="Priority for --add-context-rule (higher applies first)"
)
parser.add_argument(
"--force",
action="store_true",
@@ -66,6 +66,34 @@ def _parse_domains(raw: str | None) -> list[str] | None:
return normalize_domains(raw)
def _load_trap_demotion_sets(domains: list[str] | None) -> tuple[frozenset, frozenset]:
"""Collect 禁裸词/勿修 demotion tokens from each named domain's context file.
Convention: ``~/.transcript-fixer/contexts/<domain>.md``. No --domain (or
the "all" alias) means no specific domain's file to consult, so nothing
demotes demotion is a domain-scoped veto, and a whole-library run has
no owner to veto with. A missing or malformed context file must never
break Stage 1, so failures return empty sets for that domain.
"""
if not domains:
return frozenset(), frozenset()
from core.trap_scanner import extract_demotion_sets
banned: set = set()
keep: set = set()
ctx_home = Path.home() / ".transcript-fixer" / "contexts"
for domain in domains:
context_path = ctx_home / f"{domain}.md"
if not context_path.is_file():
continue
try:
sets = extract_demotion_sets(context_path.read_text(encoding="utf-8"))
except Exception:
continue
banned.update(sets.banned_froms)
keep.update(sets.keep_tokens)
return frozenset(banned), frozenset(keep)
def _is_all_alias(raw) -> bool:
"""True when the raw --domain value is the whole-library alias ("all",
any case, alone or inside a comma-separated list).
@@ -394,6 +422,80 @@ def cmd_add_correction(args: argparse.Namespace) -> None:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
# Post-add advisory: if the domain's context file already marks this FROM
# 禁裸词/勿修, the new rule will be demoted to review at Stage 1 — say so
# now, while the author is still at the decision point, rather than letting
# them read "Added" as "will auto-apply".
_demote_froms, _keep_tokens = _load_trap_demotion_sets([domain_to_write])
if args.from_text in _demote_froms or args.from_text in _keep_tokens:
print(f"⚠️ '{args.from_text}' is marked 禁裸词/勿修 in the "
f"{domain_to_write} domain context — this rule will DEFER to "
f"review at Stage 1 instead of auto-applying. If the correction "
f"is only right in a specific context, prefer "
f"--add-context-rule with a context pattern.",
file=sys.stderr)
def cmd_add_context_rule(args: argparse.Namespace) -> None:
"""Add a context-aware regex rule (--add-context-rule)."""
service = _get_service()
# A rule lives in exactly one domain (or global); a comma-separated
# --domain is a read-side filter, not a write target.
domains = _parse_domains(getattr(args, "domain", None))
if domains and len(domains) > 1:
print(
f"Error: --add-context-rule writes to exactly one domain, got "
f"{len(domains)}: {', '.join(domains)}. Run one per domain, or "
f"omit --domain for a global rule.",
file=sys.stderr,
)
sys.exit(2)
_reject_all_on_write(getattr(args, "domain", None), "--add-context-rule")
domain_to_write = domains[0] if domains else None
try:
rule_id = service.add_context_rule(
args.from_text,
args.to_text,
domain=domain_to_write,
description=getattr(args, "description", None),
priority=getattr(args, "priority", 0) or 0,
added_by="cli",
)
print(f"Added context rule #{rule_id}: {args.from_text!r} -> "
f"{args.to_text!r} (domain: {domain_to_write or 'global'})")
except Exception as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
def cmd_list_context_rules(args: argparse.Namespace) -> None:
"""List context rules (--list-context-rules)."""
service = _get_service()
domains = _parse_domains(getattr(args, "domain", None))
if domains and len(domains) > 1:
print(
f"Error: --list-context-rules filters to one domain at a time, "
f"got {len(domains)}: {', '.join(domains)}.",
file=sys.stderr,
)
sys.exit(2)
rules = service.list_context_rules(
domain=domains[0] if domains else None,
include_inactive=getattr(args, "all", False),
)
if getattr(args, "json_output", False):
print(json.dumps(rules, ensure_ascii=False, indent=2))
return
if not rules:
print("No context rules found.")
return
for r in rules:
scope = r["domain"] or "global"
inactive = "" if r["is_active"] else " [DISABLED]"
desc = f"{r['description']}" if r.get("description") else ""
print(f"#{r['id']} [{scope}] {r['pattern']!r} -> {r['replacement']!r}"
f" (priority {r['priority']}){desc}{inactive}")
def cmd_scan_traps(args: argparse.Namespace) -> None:
"""Scan --input for every trap documented in a domain context file.
@@ -749,7 +851,10 @@ def cmd_run_correction(args: argparse.Namespace) -> dict | None:
# rationale holds per-rule.
domains = _parse_domains(args.domain)
corrections, correction_meta = service.get_corrections_with_metadata(domains)
context_rules = service.load_context_rules()
# Context rules follow the same domain scoping as corrections: global
# (domain-less) rules always load; a domain-named rule loads only when its
# domain is active. None = no filter, the legacy union behavior.
context_rules = service.load_context_rules(domains)
domain_stats = service.get_domain_stats()
# --apply-domain: the user explicitly asserted this transcript belongs to
@@ -778,6 +883,39 @@ def cmd_run_correction(args: argparse.Namespace) -> dict | None:
"go to review). To trust a domain, name it: --domain <name>.",
file=sys.stderr)
# Trap-aware demotion: a named domain's context file
# (~/.transcript-fixer/contexts/<domain>.md) can veto auto-application of
# the same-named dictionary rule — traps annotated 禁裸词/禁入词典 and
# confirmed-correct (勿修) records are the domain owner's machine-readable
# "judge the context, never auto-apply" (the 绿点→绿电 class: a real-word
# rule that is right in some contexts and wrong in others). Runs AFTER the
# trusted_domain marking above so the veto wins over the flattening; it
# only sets a meta flag, _assess_risk does the grading. Without the veto
# the flattening auto-applies a 2-char real-word rule that safe mode would
# otherwise defer.
demote_froms, keep_tokens = _load_trap_demotion_sets(domains)
if demote_froms or keep_tokens:
_demoted = [
_wrong for _wrong in correction_meta
if _wrong in demote_froms or _wrong in keep_tokens
]
for _wrong in _demoted:
correction_meta[_wrong]["demoted_by_trap"] = True
if _demoted:
if getattr(args, "apply_all", False):
# The demotion flag is graded below, but --apply-all applies
# every risk level anyway — say what will actually happen.
print(f"🛡️ Trap demotion: {len(_demoted)} rule(s) marked by "
f"domain-context markers (禁裸词/勿修) — applied anyway "
f"under --apply-all: "
f"{', '.join(sorted(_demoted)[:5])}"
f"{'' if len(_demoted) > 5 else ''}", file=sys.stderr)
else:
print(f"🛡️ Trap demotion: {len(_demoted)} rule(s) deferred to review "
f"by domain-context markers (禁裸词/勿修): "
f"{', '.join(sorted(_demoted)[:5])}"
f"{'' if len(_demoted) > 5 else ''}", file=sys.stderr)
# Merge person-name ASR variants from the people roster (if configured).
# Source: env TRANSCRIPT_FIXER_PEOPLE_ROSTER > config.json paths.people_roster_path (there is no --people-roster CLI flag).
# The roster is the curated SSOT for important recurring people; DB entries (catch-all,
@@ -600,21 +600,50 @@ class CorrectionService:
return conflicts
def load_context_rules(self) -> List[Dict]:
def load_context_rules(self, domains: Optional[List[str]] = None) -> List[Dict]:
"""
Load active context-aware regex rules.
A rule with domain = NULL is global and always loads; a rule with a
named domain only loads when that domain is in `domains`. Passing no
filter (None) keeps the legacy union behavior and loads every active
rule mirroring how corrections load without --domain.
On a database not yet migrated to v2.4 there IS no domain column, and
by construction no rule could have been added with a domain so every
rule is global and the legacy query is the correct one.
Args:
domains: Active domain names, or None for no filter
Returns:
List of rule dictionaries with pattern, replacement, description
"""
try:
with self.repository._pool.get_connection() as conn:
cursor = conn.execute("""
SELECT pattern, replacement, description
FROM context_rules
WHERE is_active = 1
ORDER BY priority DESC
""")
has_domain_col = any(
row[1] == "domain"
for row in conn.execute("PRAGMA table_info(context_rules)")
)
if has_domain_col and domains:
placeholders = ", ".join("?" for _ in domains)
cursor = conn.execute(
f"""
SELECT pattern, replacement, description
FROM context_rules
WHERE is_active = 1
AND (domain IS NULL OR domain IN ({placeholders}))
ORDER BY priority DESC
""",
domains,
)
else:
cursor = conn.execute("""
SELECT pattern, replacement, description
FROM context_rules
WHERE is_active = 1
ORDER BY priority DESC
""")
rules = []
for row in cursor.fetchall():
@@ -631,6 +660,140 @@ class CorrectionService:
logger.error(f"Failed to load context rules: {e}")
return []
def add_context_rule(
self,
pattern: str,
replacement: str,
domain: Optional[str] = None,
description: Optional[str] = None,
priority: int = 0,
added_by: Optional[str] = None,
) -> int:
"""
Add a context-aware regex rule.
Args:
pattern: Regex pattern matched against the transcript text
replacement: Replacement text for each match
domain: Scope the rule to one domain; None = global (applies to
every domain)
description: Human-readable rule name (used as the change's
rule_name in reports)
priority: Higher priority rules apply first
added_by: Provenance label
Returns:
ID of the inserted rule
Raises:
ValidationError: On empty pattern/replacement, a bad regex, a
malformed domain, a duplicate pattern, or a database not yet
migrated to v2.4 (domain column missing).
"""
if not pattern or not pattern.strip():
raise ValidationError("context rule pattern must not be empty")
if not replacement or not replacement.strip():
raise ValidationError("context rule replacement must not be empty")
import re as _re
try:
_re.compile(pattern)
except _re.error as e:
raise ValidationError(f"invalid context rule pattern {pattern!r}: {e}")
if domain is not None:
self.validate_domain_name(domain)
with self.repository._pool.get_connection() as conn:
has_domain_col = any(
row[1] == "domain"
for row in conn.execute("PRAGMA table_info(context_rules)")
)
if not has_domain_col:
raise ValidationError(
"context_rules has no domain column — migrate the "
"database to v2.4. Canonical path: fix_transcription.py "
"--migration migrate. If that runner reports version 0.0 "
"on this DB (a pre-existing runner issue on schema.sql-"
"built databases), apply the additive column directly "
"instead: sqlite3 <db> \"ALTER TABLE context_rules "
"ADD COLUMN domain TEXT\""
)
duplicate = conn.execute(
"SELECT id FROM context_rules WHERE pattern = ?", (pattern,)
).fetchone()
if duplicate:
raise ValidationError(
f"context rule pattern already exists (id {duplicate[0]}): {pattern!r}"
)
cursor = conn.execute(
"""
INSERT INTO context_rules
(pattern, replacement, description, priority, added_by, domain)
VALUES (?, ?, ?, ?, ?, ?)
""",
(pattern, replacement, description, priority, added_by, domain),
)
rule_id = cursor.lastrowid
conn.execute(
"""
INSERT INTO audit_log (action, entity_type, entity_id, details)
VALUES ('add_context_rule', 'context_rules', ?, ?)
""",
(rule_id, f"{pattern!r} -> {replacement!r} (domain: {domain or 'global'})"),
)
conn.commit()
logger.info(f"Added context rule {rule_id}: {pattern!r} (domain: {domain or 'global'})")
return rule_id
def list_context_rules(
self, domain: Optional[str] = None, include_inactive: bool = False
) -> List[Dict]:
"""
List context rules, optionally filtered to one domain plus globals.
Args:
domain: Show global rules plus rules of this domain; None lists
every rule regardless of domain
include_inactive: Also list disabled rules
Returns:
List of rule dicts with id/pattern/replacement/description/
priority/domain/is_active
"""
with self.repository._pool.get_connection() as conn:
has_domain_col = any(
row[1] == "domain"
for row in conn.execute("PRAGMA table_info(context_rules)")
)
domain_expr = "domain" if has_domain_col else "NULL AS domain"
sql = f"""
SELECT id, pattern, replacement, description, priority,
{domain_expr}, is_active
FROM context_rules
"""
params: list = []
clauses = []
if not include_inactive:
clauses.append("is_active = 1")
if domain is not None and has_domain_col:
clauses.append("(domain IS NULL OR domain = ?)")
params.append(domain)
if clauses:
sql += " WHERE " + " AND ".join(clauses)
sql += " ORDER BY priority DESC, id"
cursor = conn.execute(sql, params)
return [
{
"id": row[0],
"pattern": row[1],
"replacement": row[2],
"description": row[3],
"priority": row[4],
"domain": row[5],
"is_active": bool(row[6]),
}
for row in cursor.fetchall()
]
def save_history(self, filename: str, domain: str, original_length: int,
stage1_changes: int, stage2_changes: int, model: str,
changes: List[Any], *, success: bool = True,
@@ -270,7 +270,9 @@ class DictionaryProcessor:
line_num = corrected[:match.start()].count('\n') + 1
# Use the same risk classifier as dictionary rules so context
# rules do not bypass the common-word / length safety layers.
risk = self._assess_risk(matched, replacement)
# Trap demotion is excluded: the pattern IS the context.
risk = self._assess_risk(matched, replacement,
allow_trap_demotion=False)
changes.append(Change(
line_number=line_num,
from_text=matched,
@@ -483,7 +485,8 @@ class DictionaryProcessor:
return False
def _assess_risk(self, wrong: str, correct: str) -> str:
def _assess_risk(self, wrong: str, correct: str, *,
allow_trap_demotion: bool = True) -> str:
"""
Classify a dictionary change as low/medium/high risk.
@@ -503,10 +506,26 @@ class DictionaryProcessor:
- Non-word garbled text (e.g., 克劳锐 -> Claude)
- High confidence (>= 0.95), length >= 5, and not identified as a
likely valid Chinese phrase
`allow_trap_demotion` is False only for context rules: a context rule
carries its own context in the pattern, so the trap demotion (which
exists to stop context-free bare-word rules) must not fire on it
otherwise a lookahead pattern like 妙计(?=) whose match text equals
a demoted FROM would inherit the demotion and the documented escape
channel (--add-context-rule) would silently stop working.
"""
meta = self.correction_meta.get(wrong, {})
confidence = meta.get("confidence", 1.0)
# Trap-demoted: the named domain's context file marks this pair
# 禁裸词/禁入词典, or the FROM side as confirmed-correct (勿修) — the
# pair may only be corrected with its context judged by the native
# pass, so it never auto-applies in safe mode, trusted domain or not.
# --apply-all remains the operator's explicit override (review_mode
# off applies every level). See cmd Stage 1 demotion wiring.
if allow_trap_demotion and meta.get("demoted_by_trap"):
return "medium"
# --apply-domain marked this rule as belonging to the domain the user
# explicitly asserted for this transcript; domain match = trust, so it
# auto-applies even in safe mode (see cmd Stage 1 setup).
@@ -32,6 +32,8 @@ CREATE INDEX IF NOT EXISTS idx_corrections_from_text ON corrections(from_text);
-- Table: context_rules
-- Regex-based context-aware correction rules
-- domain is NULL for global rules (apply to every domain); a named rule only
-- loads when its domain is active (migration v2.4).
CREATE TABLE IF NOT EXISTS context_rules (
id INTEGER PRIMARY KEY AUTOINCREMENT,
pattern TEXT NOT NULL UNIQUE,
@@ -40,11 +42,19 @@ CREATE TABLE IF NOT EXISTS context_rules (
priority INTEGER NOT NULL DEFAULT 0,
is_active BOOLEAN NOT NULL DEFAULT 1,
added_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
added_by TEXT
added_by TEXT,
domain TEXT
);
CREATE INDEX IF NOT EXISTS idx_context_rules_priority ON context_rules(priority DESC);
CREATE INDEX IF NOT EXISTS idx_context_rules_is_active ON context_rules(is_active);
-- NOTE: no idx_context_rules_domain here. schema.sql also runs against
-- pre-v2.4 databases whose context_rules has no domain column yet
-- (CREATE TABLE IF NOT EXISTS no-ops there), and an index on a missing
-- column would abort the whole init. The v2.4 migration creates the index
-- together with the column; fresh databases created from this file simply
-- have no domain index until a migration adds it — harmless for a table
-- this small.
-- Table: correction_history
-- Audit log for all correction runs
@@ -263,6 +263,61 @@ class TrapEntry:
kind: str # "trap" | "confirmed_correct"
# A trap annotation carrying one of these literal markers is the domain
# owner's machine-readable statement that the pair may never auto-apply as a
# bare dictionary rule — the correction needs the surrounding context judged
# (by the native pass), so Stage 1 must defer it instead. Authors write the
# marker in the cue prose exactly as they already did before it became
# executable (**妙计 → 妙记** — 飞书妙记语境,禁裸词).
_DEMOTE_MARKERS = ("禁裸词", "禁入词典")
@dataclass(frozen=True)
class DemotionSets:
"""Machine-readable auto-apply vetoes parsed from a domain context file.
banned_froms: FROM variants of traps annotated 禁裸词/禁入词典 a
dictionary rule with one of these as its from_text must defer to
review, never auto-apply (the 绿点绿电 class: a real-word rule that
is right in some contexts and wrong in others).
keep_tokens: confirmed-correct (勿修) tokens a dictionary rule with one
of these as its from_text would rewrite a token the domain already
settled as correct, so it must defer too.
"""
banned_froms: frozenset
keep_tokens: frozenset
def extract_demotion_sets(context_text: str) -> DemotionSets:
"""Parse Stage 1 demotion vetoes out of a domain context markdown.
Reuses the same bullet-line-start anchored regexes as
extract_trap_entries, so every trap the scanner can see is also visible
here. The marker is matched against the trap bullet's whole line — the
established authoring convention puts it inside the bold parenthesized
annotation (**A B禁裸词**), and a FROM/TO term literally named
禁裸词 is not a plausible correction token, so the wider window adds no
false-marker risk.
"""
banned: set = set()
keep: set = set()
for m in _BOLD_TRAP.finditer(context_text):
line_end = context_text.find("\n", m.end())
if line_end == -1:
line_end = len(context_text)
bullet_line = context_text[m.start():line_end]
if not any(marker in bullet_line for marker in _DEMOTE_MARKERS):
continue
for variant in _parse_from_side(m.group(1)):
banned.add(variant)
for m in _BOLD_CONFIRMED.finditer(context_text):
term = _clean_token(m.group(1))
if term:
keep.add(term)
return DemotionSets(frozenset(banned), frozenset(keep))
@dataclass(frozen=True)
class TrapHit:
"""One occurrence of one variant in the scanned text."""
@@ -46,6 +46,8 @@ import sys
from cli import (
cmd_init,
cmd_add_correction,
cmd_add_context_rule,
cmd_list_context_rules,
cmd_audit,
cmd_list_corrections,
cmd_export_corrections,
@@ -127,6 +129,11 @@ def main() -> None:
elif args.add_correction:
args.from_text, args.to_text = args.add_correction
cmd_add_correction(args)
elif args.add_context_rule:
args.from_text, args.to_text = args.add_context_rule
cmd_add_context_rule(args)
elif getattr(args, 'list_context_rules', False):
cmd_list_context_rules(args)
elif getattr(args, 'audit_dictionary', False):
cmd_audit(args)
elif args.list_corrections:
@@ -4,7 +4,7 @@
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>转写修正审核台</title>
<link rel="icon" href="data:image/svg+xml,<svg xmlns=%22http://www.w3.org/2000/svg%22 viewBox=%220 0 100 100%22><text y=%22.9em%22 font-size=%2290%22>📝</text></svg>">
<link rel="icon" href="data:image/svg+xml,<svg xmlns=%22http%3A%2F%2Fwww.w3.org%2F2000%2Fsvg%22 viewBox=%220 0 100 100%22><text y=%22.9em%22 font-size=%2290%22>📝</text></svg>">
<link rel="stylesheet" href="/static/style.css">
</head>
<body>
@@ -0,0 +1,330 @@
#!/usr/bin/env python3
"""
Tests for trap-aware demotion (禁裸词/勿修 vetoes) and domain-scoped
context rules.
Covers:
A. extract_demotion_sets parsing (marker inside the bold annotation, the
established production convention) and its effect on _assess_risk /
DictionaryProcessor behavior.
B. context_rules v2.4: domain column migration SQL, domain-filtered loading,
add/list CLI service methods, and unmigrated-database fallbacks.
"""
import sqlite3
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent))
from core.dictionary_processor import DictionaryProcessor
from core.trap_scanner import extract_demotion_sets
from core.correction_repository import CorrectionRepository
from core.correction_service import CorrectionService, ValidationError
from utils.migrations import MIGRATION_V2_4
# ---------------------------------------------------------------------------
# A1. extract_demotion_sets parsing
# ---------------------------------------------------------------------------
class TestExtractDemotionSets:
def test_marker_inside_bold_annotation_bans_from_variants(self):
"""The production convention puts 禁裸词 inside the bold parens."""
text = "- **妙计 → 妙记(飞书妙记产品语境,禁裸词)** — 说明文字\n"
sets = extract_demotion_sets(text)
assert "妙计" in sets.banned_froms
assert "妙记" not in sets.banned_froms
def test_marker_after_bold_span_also_bans(self):
text = "- **绘画 → 会话** — AI 对话语境,禁裸词\n"
sets = extract_demotion_sets(text)
assert "绘画" in sets.banned_froms
def test_trap_without_marker_does_not_ban(self):
text = "- **asms → AICMS** — 产品名误识,已入库\n"
sets = extract_demotion_sets(text)
assert sets.banned_froms == frozenset()
def test_multi_variant_from_side_all_banned(self):
text = "- **卖吸引/卖新鲜 → 麦锡颖(禁入词典)** — cue\n"
sets = extract_demotion_sets(text)
assert "卖吸引" in sets.banned_froms
assert "卖新鲜" in sets.banned_froms
def test_confirmed_correct_record_becomes_keep_token(self):
text = "- **薛辉 = 真实实体,勿修** — 域内 11 处引用\n"
sets = extract_demotion_sets(text)
assert "薛辉" in sets.keep_tokens
assert sets.banned_froms == frozenset()
def test_marker_in_unrelated_line_does_not_ban_other_traps(self):
text = (
"- **公开 → 工勘** — 渠道语境判,无标记\n"
"- **妙计 → 妙记(禁裸词)** — cue\n"
)
sets = extract_demotion_sets(text)
assert "公开" not in sets.banned_froms
assert "妙计" in sets.banned_froms
# ---------------------------------------------------------------------------
# A2. demoted_by_trap grading in _assess_risk
# ---------------------------------------------------------------------------
def _make_processor(meta):
return DictionaryProcessor(
{"绿点": "绿电", "asms": "AICMS"},
[],
meta,
speaker_labels=set(),
)
class TestDemotedByTrapGrading:
def test_demotion_beats_trusted_domain(self):
processor = _make_processor(
{"绿点": {"confidence": 1.0,
"trusted_domain": True, "demoted_by_trap": True},
"asms": {"confidence": 1.0, "trusted_domain": True}}
)
assert processor._assess_risk("绿点", "绿电") == "medium"
assert processor._assess_risk("asms", "AICMS") == "low"
def test_safe_mode_defers_demoted_rule_and_applies_normal_rule(self):
processor = _make_processor(
{"绿点": {"confidence": 1.0,
"trusted_domain": True, "demoted_by_trap": True},
"asms": {"confidence": 1.0, "trusted_domain": True}}
)
corrected, changes = processor.process(
"那不是有个绿点吗?不是 asms docs。", review_mode=True)
assert "绿点" in corrected, "demoted rule must not auto-apply"
assert "绿电" not in corrected
assert "AICMS" in corrected, "trusted normal rule still applies"
demoted = [c for c in changes if c.from_text == "绿点"]
assert demoted and demoted[0].risk == "medium"
def test_apply_all_still_applies_demoted_rule(self):
"""--apply-all is the operator's explicit override: review_mode off."""
processor = _make_processor(
{"绿点": {"confidence": 1.0,
"trusted_domain": True, "demoted_by_trap": True},
"asms": {"confidence": 1.0, "trusted_domain": True}}
)
corrected, _ = processor.process("那不是有个绿点吗?", review_mode=False)
assert "绿电" in corrected
def test_context_rule_not_demoted_by_matching_from_text(self):
"""A context rule carries its own context: even when its match text
equals a demoted FROM (lookahead-style pattern), the demotion must
not fire on it otherwise --add-context-rule stops being the escape
channel the docs prescribe."""
processor = DictionaryProcessor(
{"妙计": "妙记"},
[{"pattern": r"妙计(?=比)", "replacement": "妙记",
"description": "妙记后接比"}],
{"妙计": {"confidence": 1.0,
"trusted_domain": True, "demoted_by_trap": True}},
speaker_labels=set(),
)
corrected, changes = processor.process("妙计比它更准吗", review_mode=True)
assert "妙记比它更准吗" in corrected
context_changes = [c for c in changes if c.rule_type == "context_rule"]
assert context_changes and context_changes[0].risk != "medium"
# ---------------------------------------------------------------------------
# A3. CLI wiring: _load_trap_demotion_sets
# ---------------------------------------------------------------------------
class TestLoadTrapDemotionSets:
def test_reads_each_named_domain_context_file(self, tmp_path, monkeypatch):
contexts = tmp_path / ".transcript-fixer" / "contexts"
contexts.mkdir(parents=True)
(contexts / "huawei.md").write_text(
"- **妙计 → 妙记(禁裸词)** — cue\n- **薛辉 = 真实实体,勿修**\n",
encoding="utf-8")
(contexts / "pkm.md").write_text(
"- **新一 → 星壹(禁入词典)** — cue\n", encoding="utf-8")
monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path))
from cli.commands import _load_trap_demotion_sets
banned, keep = _load_trap_demotion_sets(["huawei", "pkm"])
assert banned == frozenset({"妙计", "新一"})
assert keep == frozenset({"薛辉"})
def test_no_domain_returns_empty(self):
from cli.commands import _load_trap_demotion_sets
assert _load_trap_demotion_sets(None) == (frozenset(), frozenset())
def test_missing_context_file_is_silent_skip(self, tmp_path, monkeypatch):
(tmp_path / ".transcript-fixer" / "contexts").mkdir(parents=True)
monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path))
from cli.commands import _load_trap_demotion_sets
assert _load_trap_demotion_sets(["nosuchdomain"]) == \
(frozenset(), frozenset())
# ---------------------------------------------------------------------------
# B1. Migration v2.4 SQL
# ---------------------------------------------------------------------------
_OLD_CONTEXT_RULES_DDL = """
CREATE TABLE context_rules (
id INTEGER PRIMARY KEY AUTOINCREMENT,
pattern TEXT NOT NULL UNIQUE,
replacement TEXT NOT NULL,
description TEXT,
priority INTEGER NOT NULL DEFAULT 0,
is_active BOOLEAN NOT NULL DEFAULT 1,
added_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
added_by TEXT
);
"""
def _old_style_db(path: Path) -> sqlite3.Connection:
"""A context_rules table as it existed before v2.4 (no domain column)."""
conn = sqlite3.connect(path)
conn.execute(_OLD_CONTEXT_RULES_DDL)
conn.execute(
"INSERT INTO context_rules (pattern, replacement, description) "
"VALUES ('legacy', 'legacy-repl', 'legacy rule')")
conn.commit()
return conn
class TestMigrationV24:
def test_forward_adds_domain_column_and_preserves_rows(self, tmp_path):
conn = _old_style_db(tmp_path / "corrections.db")
conn.executescript(MIGRATION_V2_4.forward_sql)
cols = {row[1] for row in conn.execute("PRAGMA table_info(context_rules)")}
assert "domain" in cols
row = conn.execute(
"SELECT pattern, domain FROM context_rules WHERE pattern = 'legacy'"
).fetchone()
assert row == ("legacy", None), "existing rows must stay global (NULL)"
conn.close()
def test_backward_drops_domain_column(self, tmp_path):
conn = _old_style_db(tmp_path / "corrections.db")
conn.executescript(MIGRATION_V2_4.forward_sql)
conn.executescript(MIGRATION_V2_4.backward_sql)
cols = {row[1] for row in conn.execute("PRAGMA table_info(context_rules)")}
assert "domain" not in cols
conn.close()
# ---------------------------------------------------------------------------
# B2. Service: domain-filtered load, add, list
# ---------------------------------------------------------------------------
@pytest.fixture
def service(tmp_path):
repo = CorrectionRepository(tmp_path / "corrections.db")
return CorrectionService(repo)
class TestContextRuleService:
def test_add_and_list_round_trip(self, service):
rule_id = service.add_context_rule(
r"上传到妙计", "上传到妙记", domain="huawei",
description="飞书妙记语境", added_by="test")
assert rule_id > 0
rules = service.list_context_rules(domain="huawei")
assert len(rules) == 1
r = rules[0]
assert r["pattern"] == r"上传到妙计"
assert r["domain"] == "huawei"
assert r["is_active"] is True
def test_domain_filtered_loading(self, service):
service.add_context_rule("global-rule", "g", description="global")
service.add_context_rule("huawei-rule", "h", domain="huawei")
service.add_context_rule("pkm-rule", "p", domain="pkm")
patterns = {r["pattern"] for r in service.load_context_rules(["huawei"])}
assert patterns == {"global-rule", "huawei-rule"}
patterns = {r["pattern"] for r in service.load_context_rules(["other"])}
assert patterns == {"global-rule"}
patterns = {r["pattern"] for r in service.load_context_rules(None)}
assert patterns == {"global-rule", "huawei-rule", "pkm-rule"}
def test_add_rejects_duplicate_pattern(self, service):
service.add_context_rule("dup", "x")
with pytest.raises(ValidationError, match="already exists"):
service.add_context_rule("dup", "y")
def test_add_rejects_invalid_regex(self, service):
with pytest.raises(ValidationError, match="invalid context rule pattern"):
service.add_context_rule("(unclosed", "x")
def test_add_rejects_empty_fields(self, service):
with pytest.raises(ValidationError):
service.add_context_rule("", "x")
with pytest.raises(ValidationError):
service.add_context_rule("p", "")
def test_audit_log_written(self, service):
rule_id = service.add_context_rule("audited", "x", domain="huawei")
with service.repository._pool.get_connection() as conn:
row = conn.execute(
"SELECT action, entity_id FROM audit_log "
"WHERE action = 'add_context_rule' AND entity_id = ?",
(rule_id,)).fetchone()
assert row is not None
assert tuple(row) == ("add_context_rule", rule_id)
def test_list_include_inactive(self, service):
rule_id = service.add_context_rule("off", "x")
with service.repository._pool.get_connection() as conn:
conn.execute(
"UPDATE context_rules SET is_active = 0 WHERE id = ?",
(rule_id,))
conn.commit()
assert service.list_context_rules() == []
assert len(service.list_context_rules(include_inactive=True)) == 1
def test_all_flag_reaches_include_inactive(self, tmp_path):
"""--all must parse and reach include_inactive (the flag once existed
only in docs and died at argparse)."""
from cli.argument_parser import create_argument_parser
parser = create_argument_parser()
args = parser.parse_args(["--list-context-rules", "--all"])
assert args.list_context_rules is True
assert getattr(args, "all", False) is True
# ---------------------------------------------------------------------------
# B3. Unmigrated-database behavior
# ---------------------------------------------------------------------------
class TestUnmigratedDatabase:
def _service_on_old_db(self, tmp_path) -> CorrectionService:
old_db = tmp_path / "corrections.db"
conn = _old_style_db(old_db)
conn.close()
# CorrectionRepository on an existing file must not recreate the
# table; verify our fixture still lacks the column.
repo = CorrectionRepository(old_db)
with repo._pool.get_connection() as c:
cols = {row[1] for row in c.execute("PRAGMA table_info(context_rules)")}
assert "domain" not in cols
return CorrectionService(repo)
def test_load_falls_back_to_legacy_behavior(self, tmp_path):
"""No domain column ⇒ every rule is global by construction; loading
must not crash, and domain filtering must not drop anything."""
service = self._service_on_old_db(tmp_path)
rules = service.load_context_rules(["huawei"])
assert [r["pattern"] for r in rules] == ["legacy"]
def test_add_fails_loud_with_migration_direction(self, tmp_path):
service = self._service_on_old_db(tmp_path)
with pytest.raises(ValidationError, match="migration"):
service.add_context_rule("new-rule", "x", domain="huawei")
@@ -457,6 +457,26 @@ MIGRATION_V2_3 = Migration(
is_breaking=False
)
# Migration from v2.3 to v2.4 (domain-scoped context rules)
MIGRATION_V2_4 = Migration(
version="2.4",
name="Domain-scoped context rules",
description="Add domain column to context_rules (NULL = global, applies to every domain) so context rules can be scoped per domain",
forward_sql="""
-- Existing rows keep domain = NULL, i.e. global behavior unchanged.
ALTER TABLE context_rules ADD COLUMN domain TEXT;
CREATE INDEX IF NOT EXISTS idx_context_rules_domain ON context_rules(domain);
""",
backward_sql="""
-- The index must go first: SQLite cannot drop a column an index uses.
DROP INDEX IF EXISTS idx_context_rules_domain;
ALTER TABLE context_rules DROP COLUMN domain;
""",
dependencies=["2.3"],
check_function=None,
is_breaking=False
)
# Registry of all migrations
# Order matters - add new migrations at the end
ALL_MIGRATIONS = [
@@ -465,6 +485,7 @@ ALL_MIGRATIONS = [
MIGRATION_V2_1,
MIGRATION_V2_2,
MIGRATION_V2_3,
MIGRATION_V2_4,
]
# Migration registry by version
@@ -223,7 +223,7 @@ class PathValidator:
Example:
>>> validator = PathValidator()
>>> safe_path = validator.validate_input_path("~/Documents/file.md")
>>> # Returns: Path('/home/username/Documents/file.md') or similar
>>> # Returns: Path('<home>/Documents/file.md') or similar
"""
# Check dangerous patterns in raw string
self._check_dangerous_patterns(path_str)
@@ -286,7 +286,7 @@ class PathValidator:
Example:
>>> validator = PathValidator()
>>> safe_path = validator.validate_output_path("~/Documents/output.md")
>>> # Returns: Path('/home/username/Documents/output.md') or similar
>>> # Returns: Path('<home>/Documents/output.md') or similar
"""
# Check dangerous patterns
self._check_dangerous_patterns(path_str)
@@ -347,7 +347,7 @@ class PathValidator:
directory: Directory path to add
Example:
>>> validator.add_allowed_directory("/home/username/Projects")
>>> validator.add_allowed_directory("<home>/Projects")
"""
dir_path = Path(directory).expanduser().resolve()
self.allowed_base_dirs.add(dir_path)