fix: reconcile declared vs derived in the sizing faculty (#274)

agents/conductors/bug/BUG_TAXONOMY.md

@@ -86,9 +86,13 @@ or acknowledged YELLOW), never a check re-run here.
 
 ## Difficulty & selection (reused, severity-weighted)
 
-Difficulty is the Feature Agent's heuristic verbatim (`F.estimate_difficulty` — repos
-affected, prompt size, scientific complexity, architectural risk, test burden, memory
-context; thresholds `≤2 small · ≤5 medium · ≤9 large · >9 too-large`). Selection scans
+Difficulty is the sizing faculty's heuristic verbatim, read through its precedence
+rule (`F.effective_difficulty` — repos affected, prompt size, scientific complexity,
+architectural risk, test burden, memory context; thresholds `≤2 small · ≤5 medium ·
+≤9 large · >9 too-large`), so a `Difficulty:` the prompt DECLARES wins and the derived
+level is reported alongside it. A declared `Type: bug` likewise blocks a prose-keyword
+re-home suggestion: the author already answered that question, and a defect report
+legitimately says "refactor" or "documentation" while describing a crash. Selection scans
 `bug/**` (excluding `README.md`), down-ranks paths referenced in `active.md` /
 `planned.md`, and ranks **severity-first** by default and under `--impact` (a bug list is
 a triage queue); `--difficulty easy` / `--budget` / `--model weak` flip to smallest-first

agents/conductors/bug/_bug.py

@@ -242,10 +242,19 @@ def health_validation(workflow: str, factors: dict) -> str:
 
 
 def re_home_check(p: dict) -> str | None:
-    """If a bug/ prompt is really a feature/refactor/docs/research task, say so."""
+    """If a bug/ prompt is really a feature/refactor/docs/research task, say so.
+
+    A prompt that DECLARES `Type: bug` is never re-homed on prose keywords: the
+    author already answered this question, and prose is the weaker signal (a bug
+    report legitimately says "refactor", "documentation" or "design decision"
+    while describing a defect). Same family as the difficulty leg — derive from
+    prose, but never over a declared header key (PyAutoBrain#217, #274).
+    """
     wt = p["work_type"]
     if wt != "bug":
         return None
+    if (p.get("declared_type") or "") == "bug":
+        return None
     text = p["text"].lower()
     # A genuine defect signal keeps it a bug even if other words also fire.
     defect = F._hits(text, ["bug", "regression", "crash", "traceback", "wrong",
@@ -264,7 +273,7 @@ def re_home_check(p: dict) -> str | None:
 
 
 def analyse_bug(p: dict, heart_check: str | None = None) -> dict:
-    level, score, factors = F.estimate_difficulty(p)
+    level, score, factors, derived_level = F.effective_difficulty(p)
     cls = classify(p, factors)
     workflow = recommended_workflow(p, factors)
     mishome = re_home_check(p)
@@ -283,6 +292,11 @@ def analyse_bug(p: dict, heart_check: str | None = None) -> dict:
         "recommended_workflow": workflow,
         "rehome_suggestion": mishome,
         "difficulty": level,
+        "difficulty_declared": p.get("declared_difficulty"),
+        "difficulty_derived": derived_level,
+        "difficulty_disagreement": (
+            p.get("declared_difficulty") is not None and derived_level != level
+        ),
         "difficulty_score": score,
         "difficulty_factors": factors,
         "health_validation": health_validation(workflow, factors),
@@ -329,7 +343,7 @@ def select_bug(mind: Path, constraint: dict, limit: int):
     rows = []
     for path in prompts:
         p = F.parse_prompt(path, mind)
-        level, score, factors = F.estimate_difficulty(p)
+        level, score, factors, _derived = F.effective_difficulty(p)
         cls = classify(p, factors)
         rows.append({
             "path": p["path"], "difficulty": level, "score": score,
@@ -385,7 +399,11 @@ def emit_human(mode: str, d: dict):
     print(f"Fix strategy:         {d['fix_strategy']}")
     print(f"Recommended workflow: {d['recommended_workflow']}", end="")
     print(f"  [re-home as {d['rehome_suggestion']}/]" if d["rehome_suggestion"] else "")
-    print(f"Difficulty:           {d['difficulty']} (score {d['difficulty_score']})")
+    if d.get("difficulty_disagreement"):
+        print(f"Difficulty:           {d['difficulty']} (declared; heuristic derived "
+              f"{d['difficulty_derived']}, score {d['difficulty_score']})")
+    else:
+        print(f"Difficulty:           {d['difficulty']} (score {d['difficulty_score']})")
     print(f"Health validation:    {d['health_validation']}")
     print("Risks:")
     for r in d["risks"]:

agents/conductors/feature/_feature.py

@@ -35,7 +35,7 @@
     KNOWN_REPOS, MEMORY_WIKIS, SCIENCE_KEYWORDS, RISK_KEYWORDS, AMBIGUITY_KEYWORDS,
     policy as _sizing_policy,
     TEST_KEYWORDS, normalise_repo, parse_prompt, discover_prompts,
-    empty_discovery_reason, estimate_difficulty, _hits, _within,
+    effective_difficulty, empty_discovery_reason, _hits, _within,
     declared_blocked, priority_rank,
 )
 
@@ -159,20 +159,6 @@ def risks(level: str, factors: dict, workflow: str):
     return out
 
 
-def effective_difficulty(p: dict):
-    """(level, score, factors, derived_level) — the DECLARED level wins.
-
-    REFERENCE.md promises that the `Difficulty:` Intake persists is "the value
-    the Feature Agent later acts on", so a declared level overrides the
-    re-derived one. The derived score is kept: it still orders prompts within a
-    level, and the derived LEVEL is returned alongside so a disagreement can be
-    reported rather than silently resolved — the disagreement is evidence about
-    the heuristic and is worth seeing.
-    """
-    derived_level, score, factors = estimate_difficulty(p)
-    return p.get("declared_difficulty") or derived_level, score, factors, derived_level
-
-
 def analyse(p: dict):
     level, score, factors, derived_level = effective_difficulty(p)
     workflow, rehome = recommend_workflow(p, factors)

agents/conductors/intake/AGENTS.md

@@ -55,9 +55,17 @@ Proposed path · Header (the block written verbatim) · Risks · Next action
 - **Difficulty** — from the shared `agents/faculties/sizing/` faculty, **persisted
   into the header**, because scope is decided during the intake back-and-forth.
   The Feature Agent later *trusts* this number rather than recomputing a divergent
-  one.
+  one. A difficulty **declared in the raw input** — a pasted header block, or the
+  `ideas.md` house style `Difficulty large, supervised.` — **wins** over the
+  estimate, via the faculty's shared `effective_difficulty`: the author knows the
+  scope, and the heuristic's biggest input is length, which is a bad size proxy
+  (a prompt is long when it carries a design, not when the work is large). The
+  IntakeDecision reports which one it used (`difficulty_source`) and the level
+  the heuristic derived, and the declaration is kept out of the derived
+  title/slug. A `Difficulty:` quoted inside a code fence or backticks is
+  documentation, not a declaration.
 - **Autonomy** (`safe|supervised|human-required`) and **Priority**
-  (`low|normal|high`) — the "can an agent safely handle this?" and "how urgent?"
+  (`low|normal|high`) — the "can an agent safely handle this?" and "how urgent?" questions; a declared value wins here too, as does a declared `Type:` over the prose classifier.
   inputs, written into the header.
 
 **`too-large` means decompose, never dispatch.** A prompt sized `too-large` is

agents/conductors/intake/INTAKE_TAXONOMY.md

@@ -63,8 +63,10 @@ Primary-folder preference when several resolve: library repo → organism repo 
 ## 4. Difficulty, autonomy, priority (the header inputs)
 
 - **Difficulty** (`small|medium|large|too-large`) — the shared sizing faculty's
-  `estimate_difficulty`, the *same* function the Feature Agent uses. Persisted so
-  the number is decided once, at conception, when scope is freshest.
+  `effective_difficulty`: `estimate_difficulty` (the *same* heuristic the Feature
+  Agent uses) unless the input DECLARES a level, which wins. Persisted so the
+  number is decided once, at conception, when scope is freshest — and a
+  declaration is exactly the author correcting a heuristic that weights length.
 - **Autonomy** (`safe|supervised|human-required`) — `human-required` when the
   work needs a design decision and no repo resolved; `supervised` for
   architectural risk / large / multi-repo; `safe` otherwise.

agents/conductors/intake/_intake.py

@@ -16,7 +16,9 @@
 classification lands in `triage/` — the existing unclassified bucket, reused not
 reinvented. Difficulty is OWNED here (scope is decided during the intake
 back-and-forth) and persisted into the header via the shared sizing faculty, so
-the Feature Agent later trusts the same number.
+the Feature Agent later trusts the same number — unless the author already
+declared one in the raw text, which outranks the estimate (see "human-declared
+header fields").
 
 Stdlib only. Writes ONLY under --apply; every other path is read-only.
 """
@@ -37,7 +39,8 @@
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent / "faculties" / "sizing"))
 from _sizing import (  # noqa: E402
     WORK_TYPES, LIBRARY_REPOS, WORKSPACE_REPOS, ORGANISM_REPOS, KNOWN_REPOS,
-    RISK_KEYWORDS, AMBIGUITY_KEYWORDS, normalise_repo, estimate_difficulty, _hits,
+    RISK_KEYWORDS, AMBIGUITY_KEYWORDS, normalise_repo, declared_header,
+    declared_inline, effective_difficulty, strip_declarations, _hits,
     policy as _sizing_policy, BODY_MAP_PATH,
 )
 
@@ -262,19 +265,35 @@ def infer_autonomy(level: str, factors: dict) -> str:
 def analyse(text: str, source: str):
     """Classify raw text into a full IntakeDecision (never writes)."""
     repos = _repos_in(text)
+    # What the input DECLARES outranks what its prose merely suggests — the same
+    # rule the feature and bug conductors apply (the faculty owns it). Raw
+    # conception input may carry a full header block (a pasted prompt) or state
+    # a key mid-sentence, so both readers run; header lines win a tie.
+    inline, decl_spans = declared_inline(text)
+    header = declared_header(text)
+    declared = {k: v for k, v in {
+        "difficulty": header["declared_difficulty"] or inline.get("difficulty"),
+        "autonomy": header["declared_autonomy"] or inline.get("autonomy"),
+        "priority": header["priority"] or inline.get("priority"),
+        "type": header["declared_type"] or inline.get("type"),
+    }.items() if v}
+
     work_type, confidence, type_hits = classify_work_type(text)
+    if declared.get("type"):
+        work_type, confidence = declared["type"], "high"
     target, target_display, repos = infer_target(text, repos)
 
     # Build a prompt-shaped dict the shared sizing faculty understands.
     p = {"text": text, "repos": repos, "words": len(text.split()),
-         "target": target, "work_type": work_type}
-    level, score, factors = estimate_difficulty(p)
+         "target": target, "work_type": work_type,
+         "declared_difficulty": declared.get("difficulty")}
+    level, score, factors, estimated = effective_difficulty(p)
 
-    autonomy = infer_autonomy(level, factors)
-    priority = infer_priority(text)
+    autonomy = declared.get("autonomy") or infer_autonomy(level, factors)
+    priority = declared.get("priority") or infer_priority(text)
     workflow = infer_workflow(target, repos)
 
-    title = _title(text)
+    title = _title(strip_declarations(text, decl_spans))
     slug = _slug(title)
     folder = work_type if confidence != "low" else "triage"
     if folder == "triage":
@@ -295,19 +314,29 @@ def analyse(text: str, source: str):
         "title": title,
         "work_type": work_type,
         "classification_confidence": confidence,
+        "work_type_source": "declared" if declared.get("type") else "inferred",
         "type_signals": type_hits,
         "target": target,
         "target_display": target_display,
         "repos_affected": repos,
         "difficulty": level,
         "difficulty_score": score,
         "difficulty_factors": factors,
+        "difficulty_declared": declared.get("difficulty"),
+        "difficulty_derived": estimated,
+        "difficulty_disagreement": (
+            "difficulty" in declared and estimated != level
+        ),
+        "difficulty_source": "declared" if "difficulty" in declared else "estimated",
         "autonomy": autonomy,
+        "autonomy_source": "declared" if "autonomy" in declared else "inferred",
         "priority": priority,
+        "priority_source": "declared" if "priority" in declared else "inferred",
+        "declared_fields": declared,
         "workflow": workflow,
         "proposed_path": proposed,
         "header": header,
-        "risks": _risks(level, factors, confidence, target),
+        "risks": _risks(level, factors, confidence, target, declared, estimated),
         "next_action": _next_action(proposed, confidence),
     }
 
@@ -322,8 +351,16 @@ def _render_header(title, work_type, target_display, repos, level, autonomy, pri
     return "\n".join(lines)
 
 
-def _risks(level, factors, confidence, target):
+def _risks(level, factors, confidence, target, declared=None, estimated=None):
     out = []
+    declared = declared or {}
+    if "difficulty" in declared and declared["difficulty"] != estimated:
+        out.append(f"Difficulty {declared['difficulty']} declared in the raw text "
+                   f"— it overrides the heuristic estimate ({estimated}).")
+    for field in ("autonomy", "priority"):
+        if field in declared:
+            out.append(f"{field.capitalize()} {declared[field]} declared in the raw "
+                       f"text — taken as written, not inferred.")
     if confidence == "low":
         out.append("Low classification confidence — filed to triage/ for a human "
                    "to re-home once the work type is clear.")
@@ -1545,15 +1582,24 @@ def _derive_fields(text: str, work_type: str, target: str) -> dict:
     tgt = normalise_repo(target) if target != "-" else "?"
     if tgt in KNOWN_REPOS and tgt not in repos:
         repos = sorted(set(repos) | {tgt})
+    # Same precedence as conception: what the author already stated wins.
+    inline, _spans = declared_inline(text)
+    header = declared_header(text)
+    declared = {k: v for k, v in {
+        "difficulty": header["declared_difficulty"] or inline.get("difficulty"),
+        "autonomy": header["declared_autonomy"] or inline.get("autonomy"),
+        "priority": header["priority"] or inline.get("priority"),
+    }.items() if v}
     p = {"text": text, "repos": repos, "words": len(text.split()),
-         "target": target, "work_type": work_type}
-    level, _score, factors = estimate_difficulty(p)
+         "target": target, "work_type": work_type,
+         "declared_difficulty": declared.get("difficulty")}
+    level, _score, factors, _derived = effective_difficulty(p)
     return {
         "type": work_type,
         "target": REPO_DISPLAY.get(tgt, target if target != "-" else "?"),
         "difficulty": level,
-        "autonomy": infer_autonomy(level, factors),
-        "priority": infer_priority(text),
+        "autonomy": declared.get("autonomy") or infer_autonomy(level, factors),
+        "priority": declared.get("priority") or infer_priority(text),
         "status": "formalised",
     }
 
@@ -2215,12 +2261,19 @@ def emit_human(d: dict):
     print("== IntakeDecision ==")
     print(f"Source:               {d['source']}")
     print(f"Title:                {d['title']}")
-    print(f"Work-type:            {d['work_type']}  (confidence: {d['classification_confidence']})")
+    wt_mark = ("declared" if d.get("work_type_source") == "declared"
+               else f"confidence: {d['classification_confidence']}")
+    print(f"Work-type:            {d['work_type']}  ({wt_mark})")
     print(f"Target:               {d['target_display']}")
     print(f"Repos resolved:       {', '.join(d['repos_affected']) or '(none)'}")
-    print(f"Difficulty:           {d['difficulty']} (score {d['difficulty_score']})")
-    print(f"Autonomy:             {d['autonomy']}")
-    print(f"Priority:             {d['priority']}")
+    if d.get("difficulty_source") == "declared":
+        print(f"Difficulty:           {d['difficulty']} (declared; heuristic derived "
+              f"{d['difficulty_derived']}, score {d['difficulty_score']})")
+    else:
+        print(f"Difficulty:           {d['difficulty']} (score {d['difficulty_score']})")
+    for field in ("autonomy", "priority"):
+        mark = " (declared)" if d.get(f"{field}_source") == "declared" else ""
+        print(f"{field.capitalize() + ':':<22}{d[field]}{mark}")
     print(f"Workflow:             {d['workflow']}")
     print(f"Proposed path:        {d['proposed_path']}")
     print("Header to be written:")

agents/faculties/sizing/AGENTS.md

@@ -12,6 +12,21 @@ A PyAutoBrain read-only reasoning faculty. It owns the organism's single
 difficulty heuristic so the number is defined **once** and shared, never
 recomputed by a divergent copy.
 
+It also owns the **precedence rule** over that heuristic. `estimate_difficulty`
+derives; `effective_difficulty` reconciles — a **declared** `Difficulty:` wins,
+and the derived level comes back alongside it so a disagreement is reported
+rather than silently resolved (the disagreement is evidence about the
+heuristic). `declared_header` reads the keys a filed prompt declares
+(`Type:`/`Difficulty:`/`Autonomy:`/`Priority:`/`Status:`/`Blocked-by:`);
+`declared_inline` reads the same keys out of unheadered conception prose — the
+`ideas.md` idiom `Difficulty large, supervised.` included. Neither reads a
+fenced block: a prompt quoting a header is documenting it, not declaring it.
+
+The rule lives here rather than in each conductor because it was re-implemented
+per conductor three times and forgotten twice — the ranker (#217), the Bug Agent
+and Intake (#274) all derived over a declared key. One heuristic, one
+reconciliation, every conductor.
+
 ## Who consults it
 
 - The **Intake Agent** (`agents/conductors/intake/`) sizes a task at