feature: honour the prompt header keys the ranker is documented to act on

agents/conductors/feature/AGENTS.md

@@ -44,12 +44,36 @@ it reasons over and the PyAutoMemory routing it uses.
 | **selection** | no task given | Scan `draft/feature/**` (legacy flat `feature/**` still resolves), rank candidates, and recommend the best next task — **not** merely the first in a list; down-ranks in-flight work (from `active.md` / `planned.md`). |
 | **difficulty-constrained** | `--difficulty` / `--model` / `--budget` / `--ambitious` / `--impact` | Estimate difficulty per task and select to match the constraint (easy/weak-model/limited-token → small; ambitious/strong-model → large; impact → high-leverage). |
 
+## The declared header wins
+
+A prompt's metadata header (PyAutoMind `REFERENCE.md`, "Optional metadata
+header") is **read, not decoration**:
+
+| Key | Effect on the ranker |
+|---|---|
+| `Difficulty:` | Overrides the derived level. Intake persists it from this same sizing faculty, so it *is* the value this agent acts on; a declared/derived disagreement is reported, never silently resolved. |
+| `Priority:` | Orders the shortlist (`high` → `normal` → `low`), above the difficulty term. |
+| `Status: blocked` | Sinks the prompt below everything and bars it from being the recommended pick. |
+| `Blocked-by:` | Same, on its own — an unresolved gate reads as blocked. |
+
+Blocked prompts stay **listed, in their own band**, so a human can see and
+override; they are never recommended. Gate *state* is not resolved here — this
+agent is offline. `PyAutoMind/scripts/lifecycle.py issues --drafts` checks the
+refs against GitHub and is the tool that says a `Blocked-by:` has cleared.
+
+Keys inside fenced code blocks are documentation and are ignored, so a prompt may
+quote another's header without inheriting it.
+
 ## Difficulty & sizing
 
-Difficulty is a transparent heuristic (`small | medium | large | too-large`) over
+When nothing is declared, difficulty is a transparent heuristic
+(`small | medium | large | too-large`) over
 repos affected, prompt size, scientific complexity, architectural risk, test
 burden, and whether memory context / human judgement is required. The factor
-breakdown is in every decision so the reasoning layer can adjust.
+breakdown is in every decision so the reasoning layer can adjust. Note the
+prompt-size term: a prompt grows as it accumulates findings, so a long,
+well-documented prompt can derive `too-large` for work that is not — which is
+exactly what a declared `Difficulty:` is for.
 
 Sizing then drives the **phase decision**:
 

agents/conductors/feature/_feature.py

@@ -36,6 +36,7 @@
     policy as _sizing_policy,
     TEST_KEYWORDS, normalise_repo, parse_prompt, discover_prompts,
     empty_discovery_reason, estimate_difficulty, _hits, _within,
+    declared_blocked, priority_rank,
 )
 
 # Default sub-wiki to consult per library target when no keyword fires. Memory
@@ -158,8 +159,22 @@ 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 = estimate_difficulty(p)
+    level, score, factors, derived_level = effective_difficulty(p)
     workflow, rehome = recommend_workflow(p, factors)
     mem = memory_context(p)
     phase, stubs = phase_decision(level, factors, p)
@@ -169,6 +184,14 @@ def analyse(p: dict):
         "target": p["target"],
         "repos_affected": p["repos"],
         "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
+        ),
+        "priority": p.get("priority"),
+        "status": p.get("status"),
+        "blocked": declared_blocked(p),
         "difficulty_score": score,
         "difficulty_factors": factors,
         "recommended_workflow": workflow,
@@ -207,13 +230,18 @@ def select(mind: Path, constraint: dict, limit: int):
     rows = []
     for path in prompts:
         p = parse_prompt(path, mind)
-        level, score, factors = estimate_difficulty(p)
+        level, score, factors, derived_level = effective_difficulty(p)
         impact = score + (2 if factors["library_and_workspace"] else 0) \
             + len(factors["scientific_complexity"])
         rows.append({
             "path": p["path"], "difficulty": level, "score": score,
             "impact": impact, "repos": p["repos"],
             "in_flight": p["path"] in in_flight,
+            "blocked": declared_blocked(p),
+            "priority": p.get("priority"),
+            "priority_rank": priority_rank(p),
+            "difficulty_declared": p.get("declared_difficulty"),
+            "difficulty_derived": derived_level,
             "factors": factors,
         })
 
@@ -225,15 +253,19 @@ def select(mind: Path, constraint: dict, limit: int):
     impact_pref = constraint.get("impact")
 
     def keyfn(r):
-        # Down-rank in-flight work so we never just resurface active tasks.
-        penalty = 100 if r["in_flight"] else 0
+        # A prompt that declares itself blocked sinks below everything, so it can
+        # never be the recommended pick — it stays listed, in its own band, so a
+        # human can still see it and override.
+        # Then: in-flight work is down-ranked so we never just resurface active
+        # tasks; then declared Priority:, which is an ordering input and not
+        # merely display; then the constraint's own difficulty term.
+        head = (200 if r["blocked"] else 0) + (100 if r["in_flight"] else 0)
+        prio = r["priority_rank"]
         if impact_pref:
-            return (penalty, -r["impact"])
+            return (head, prio, -r["impact"])
         if model == "strong" or constraint.get("ambitious"):
-            return (penalty, -r["score"])
-        if model == "weak" or budget or want in ("easy", "small"):
-            return (penalty, r["score"])
-        return (penalty, r["score"])  # default: easiest-first, stable
+            return (head, prio, -r["score"])
+        return (head, prio, r["score"])  # default: easiest-first, stable
 
     candidates = rows
     if want and want not in ("easy",):
@@ -255,7 +287,19 @@ def emit_human(mode: str, decision: dict):
     print(f"Mode:                 {mode}")
     print(f"Work-type / target:   {d['work_type']} / {d['target']}")
     print(f"Repos affected:       {', '.join(d['repos_affected']) or '(none resolved)'}")
-    print(f"Difficulty:           {d['difficulty']} (score {d['difficulty_score']})")
+    src = "declared" if d.get("difficulty_declared") else "derived"
+    print(f"Difficulty:           {d['difficulty']} ({src}, score {d['difficulty_score']})")
+    if d.get("difficulty_disagreement"):
+        # Surfaced, not silently resolved: the declared value governs, but the
+        # gap is evidence about the sizing heuristic and someone should see it.
+        print(f"  ! declared {d['difficulty_declared']} but derived "
+              f"{d['difficulty_derived']} — declared wins; disagreement worth a look")
+    if d.get("priority"):
+        print(f"Priority:             {d['priority']} (declared)")
+    if d.get("blocked"):
+        print(f"BLOCKED (declared):   {d['blocked']}")
+        print("                      gate state is NOT resolved here — "
+              "`lifecycle.py issues --drafts` checks it against GitHub")
     print(f"Recommended workflow: {d['recommended_workflow']}", end="")
     print(f"  [re-home as {d['rehome_suggestion']}/]" if d["rehome_suggestion"] else "")
     if d["memory_context"]:
@@ -279,6 +323,10 @@ def emit_human(mode: str, decision: dict):
 
 
 def _next_action(d: dict):
+    if d.get("blocked"):
+        return (f"Do NOT start — the prompt declares {d['blocked']}. Clear the gate "
+                f"(or correct the header) first; `lifecycle.py issues --drafts` "
+                f"resolves gate state against GitHub.")
     if d["rehome_suggestion"]:
         return f"Re-home this prompt under {d['rehome_suggestion']}/ and scope it before development."
     if d["phase_decision"] == "split-into-phases":
@@ -347,10 +395,24 @@ def main(argv=None):
     print(f"== Feature task {mode} ({total} feature prompts considered) ==")
     print("Shortlist (recommendation — apply priorities/dependencies/health on top):")
     for i, r in enumerate(ranked):
-        flag = "  [in-flight, down-ranked]" if r["in_flight"] else ""
+        flags = []
+        if r["blocked"]:
+            flags.append(f"BLOCKED — {r['blocked']}")
+        if r["in_flight"]:
+            flags.append("in-flight, down-ranked")
+        if r.get("priority") and r["priority"] != "normal":
+            flags.append(f"priority {r['priority']}")
+        if r.get("difficulty_declared") and r["difficulty_derived"] != r["difficulty"]:
+            flags.append(f"declared {r['difficulty_declared']} vs derived "
+                         f"{r['difficulty_derived']}")
+        flag = f"  [{'; '.join(flags)}]" if flags else ""
         print(f"  {i+1}. {r['path']}  [{r['difficulty']}, score {r['score']}, "
               f"impact {r['impact']}]{flag}")
     print()
+    if ranked[0]["blocked"]:
+        # Every candidate is blocked — say so instead of recommending one anyway.
+        print("NOTE: every shortlisted prompt declares itself blocked; the pick "
+              "below is shown for context and should NOT be started as-is.\n")
     chosen = parse_prompt(mind / ranked[0]["path"], mind)
     decision = analyse(chosen)
     print("Recommended pick (not merely the first prompt — ranked by the constraint):")

agents/faculties/sizing/_sizing.py

@@ -224,6 +224,90 @@ def empty_discovery_reason(mind: Path, work_type: str) -> str:
     return f"{where} exists under {mind} but holds no prompts (backlog genuinely empty)"
 
 
+# --- the declared metadata header -------------------------------------------
+#
+# PyAutoMind/REFERENCE.md ("Optional metadata header") defines these keys and
+# states the contract this parser exists to honour: Intake persists `Difficulty:`
+# "so the value shown up front is the one the Feature Agent later acts on".
+# Parsing them here — beside the derivation — keeps declared and derived in one
+# place, and gives the bug/refactor conductors the same reading for free.
+DIFFICULTY_LEVELS = ("small", "medium", "large", "too-large")
+# `medium` is not a documented Priority: value but occurs in the live backlog;
+# read it as normal rather than dropping the prompt's stated intent.
+PRIORITY_RANK = {"high": 0, "normal": 1, "medium": 1, "low": 2}
+DEFAULT_PRIORITY_RANK = 1
+
+_HEADER_KEY_RE = re.compile(
+    r"^\s*(difficulty|status|priority|blocked-by|closes-when)\s*:\s*(.+?)\s*$", re.I
+)
+
+
+def _strip_trailing_comment(value: str) -> str:
+    """Header values may carry a trailing `# note` (the live backlog does, e.g.
+    `Blocked-by: PyAutoFit#1334   # WP1 gate (MERGED)`). Split on ` #` so a
+    `Repo#123` ref — which has no space before the hash — survives intact."""
+    return value.split(" #", 1)[0].strip()
+
+
+def declared_header(text: str) -> dict:
+    """The header keys a prompt *declares*, as opposed to what we infer.
+
+    Fenced blocks are documentation, not declarations — a prompt that quotes
+    another prompt's header in a ```-block (the bug prompt for this very fix
+    does exactly that) must not be read as declaring it. Same rule, and the
+    same reason, as PyAutoMind `lifecycle.py:draft_gate_refs`.
+    """
+    out = {"declared_difficulty": None, "status": None,
+           "priority": None, "blocked_by": [], "closes_when": []}
+    in_fence = False
+    for line in text.splitlines():
+        if line.lstrip().startswith("```"):
+            in_fence = not in_fence
+            continue
+        if in_fence:
+            continue
+        m = _HEADER_KEY_RE.match(line)
+        if not m:
+            continue
+        key, value = m.group(1).lower(), _strip_trailing_comment(m.group(2))
+        if not value:
+            continue
+        if key == "difficulty":
+            v = value.lower()
+            if v in DIFFICULTY_LEVELS and out["declared_difficulty"] is None:
+                out["declared_difficulty"] = v
+        elif key == "status" and out["status"] is None:
+            out["status"] = value.lower()
+        elif key == "priority" and out["priority"] is None:
+            out["priority"] = value.lower()
+        elif key == "blocked-by":
+            out["blocked_by"].append(value)
+        elif key == "closes-when":
+            out["closes_when"].append(value)
+    return out
+
+
+def priority_rank(p: dict) -> int:
+    return PRIORITY_RANK.get(p.get("priority") or "", DEFAULT_PRIORITY_RANK)
+
+
+def declared_blocked(p: dict):
+    """Why the prompt declares itself un-startable, or None.
+
+    Deliberately conservative: this faculty is offline, so it cannot resolve
+    whether a `Blocked-by:` gate has since closed — that is
+    `PyAutoMind/scripts/lifecycle.py issues --drafts`, which talks to GitHub. An
+    unresolved gate therefore reads as blocked. Being wrongly held back is cheap
+    and visible (the prompt is still listed, in its own band); being wrongly
+    recommended is the failure this exists to stop.
+    """
+    if (p.get("status") or "") == "blocked":
+        return "Status: blocked"
+    if p.get("blocked_by"):
+        return "Blocked-by: " + "; ".join(p["blocked_by"])
+    return None
+
+
 def parse_prompt(path: Path, mind: Path):
     """Read a prompt file and extract structure: work-type, target, repos, body."""
     text = path.read_text(encoding="utf-8", errors="replace")
@@ -282,6 +366,7 @@ def parse_prompt(path: Path, mind: Path):
         "text": text,
         "lines": text.count("\n") + 1,
         "words": len(text.split()),
+        **declared_header(text),
     }