feat: config-driven contract + concurrent-writer safety

autopilot/README.md

@@ -14,13 +14,14 @@ into a triage queue.
 
 ```
 autopilot/
-├── recommendation_contract.py   # dataclass + validate()
+├── recommendation_contract.py   # dataclass + validate() (config-driven)
 ├── message_formatter.py         # render to the new Slack template
-├── config.yaml                  # recommendation_debounce + status_tags block
+├── config_loader.py             # reads config.yaml, stdlib fallback
+├── config.yaml                   # recommendation_debounce + status_tags block
 ├── decisions/
-│   └── ledger.py                # append-only ledger, should_post() debounce
+│   └── ledger.py                # append-only ledger, should_post() debounce (file-locked)
 └── tests/
-    └── test_recommendation_contract.py
+    └── test_recommendation_contract.py   # 14 tests
 ```
 
 ## The contract

autopilot/config_loader.py

@@ -0,0 +1,109 @@
+"""Config loader: reads autopilot/config.yaml, falls back to built-in defaults.
+
+Uses PyYAML if installed; otherwise returns DEFAULTS so the module stays
+stdlib-runnable for tests. This closes the "declared but never read" gap
+flagged in the PR #180 review — ledger/contract/formatter now derive their
+constants from config rather than hardcoding them.
+"""
+
+from __future__ import annotations
+
+import copy
+import os
+from typing import Any, Optional
+
+DEFAULT_CONFIG_PATH = os.environ.get("AUTOPILOT_CONFIG_PATH", "autopilot/config.yaml")
+
+DEFAULTS: dict[str, Any] = {
+    "recommendation_debounce": {
+        "enabled": True,
+        "ledger_path": "autopilot/decisions/recommendations.jsonl",
+        "repost_policy": {
+            "open": {"min_hours_between": 72, "never_repost": False},
+            "assigned": {"min_hours_between": 168, "never_repost": False},
+            "inflight": {"min_hours_between": 168, "never_repost": False},
+            "done": {"never_repost": True},
+            "dropped": {"never_repost": True},
+        },
+    },
+    "status_tags": {
+        "open": "🟡 Open",
+        "assigned": "🔵 Assigned",
+        "inflight": "🟠 In-flight",
+        "done": "✅ Done",
+        "dropped": "❌ Dropped",
+    },
+    "message_contract": {
+        "max_step_len": 120,
+        "headline_is_outcome": True,
+        "require_impact_for": ["P0", "P1"],
+        "require_due_for": ["P0", "P1"],
+        "require_owner_for": ["P0"],
+    },
+}
+
+_cache: Optional[dict[str, Any]] = None
+
+
+def _deep_merge(base: dict[str, Any], over: dict[str, Any]) -> dict[str, Any]:
+    for k, v in over.items():
+        if k in base and isinstance(base[k], dict) and isinstance(v, dict):
+            _deep_merge(base[k], v)
+        else:
+            base[k] = v
+    return base
+
+
+def load_config(path: Optional[str] = None) -> dict[str, Any]:
+    """Load and cache the merged config. Falls back to DEFAULTS if PyYAML
+    is unavailable or the file is missing.
+
+    `path=None` reads the module-level DEFAULT_CONFIG_PATH at call time so
+    tests can swap the config file by reassigning the global + reset_cache().
+    """
+    global _cache
+    if _cache is not None:
+        return _cache
+    if path is None:
+        path = DEFAULT_CONFIG_PATH
+    cfg = copy.deepcopy(DEFAULTS)
+    try:
+        import yaml  # optional dependency
+    except ImportError:
+        pass
+    else:
+        if os.path.exists(path):
+            with open(path, "r", encoding="utf-8") as f:
+                loaded = yaml.safe_load(f) or {}
+            if isinstance(loaded, dict):
+                _deep_merge(cfg, loaded)
+    _cache = cfg
+    return cfg
+
+
+def get_debounce_hours(status: str) -> Optional[int]:
+    """Hours to suppress a re-raise, or None for never-repost."""
+    policy = load_config()["recommendation_debounce"]["repost_policy"].get(status, {})
+    if policy.get("never_repost"):
+        return None
+    return policy.get("min_hours_between", 72)
+
+
+def get_status_tag(status: str) -> str:
+    return load_config()["status_tags"].get(status, "🟡 Open")
+
+
+def get_contract() -> dict[str, Any]:
+    return load_config()["message_contract"]
+
+
+def get_ledger_path() -> str:
+    return load_config()["recommendation_debounce"].get(
+        "ledger_path", "autopilot/decisions/recommendations.jsonl"
+    )
+
+
+def reset_cache() -> None:
+    """Clear the config cache (for tests that swap config files)."""
+    global _cache
+    _cache = None

autopilot/decisions/ledger.py

@@ -8,37 +8,38 @@
 
 Status transitions (open -> assigned -> inflight -> done|dropped) are written
 by the executor agent, NOT the recommender — keeps concerns separate.
+
+Concurrency: all reads+writes are guarded by an fcntl.flock on POSIX so the
+recommender (record) and executor (transition) agents can't interleave appends
+or race on _next_seq. Falls back to no-op locking on non-POSIX platforms.
+Debounce windows and status tags are read from config.yaml via config_loader.
 """
 
 from __future__ import annotations
 
 import json
 import os
 import time
-from dataclasses import asdict
 from datetime import datetime, timezone
 from typing import Optional
 
-from recommendation_contract import Recommendation
+from recommendation_contract import Recommendation, VALID_STATUSES
+from config_loader import get_debounce_hours, get_ledger_path
+
+try:
+    import fcntl  # POSIX only
+
+    _HAS_FCNTL = True
+except ImportError:
+    _HAS_FCNTL = False
 
 
 def _today_iso() -> str:
     """Current UTC date as YYYY-MM-DD (for the first_raised ledger field)."""
     return datetime.now(timezone.utc).strftime("%Y-%m-%d")
 
 
-DEFAULT_LEDGER_PATH = os.environ.get(
-    "DECISIONS_LEDGER_PATH", "autopilot/decisions/recommendations.jsonl"
-)
-
-# Debounce windows in hours, keyed by status of the existing ledger entry.
-DEBOUNCE_HOURS = {
-    "open": 72,
-    "assigned": 168,
-    "inflight": 168,
-    "done": None,  # never repost
-    "dropped": None,  # never repost (superseded/rejected)
-}
+DEFAULT_LEDGER_PATH = os.environ.get("DECISIONS_LEDGER_PATH", get_ledger_path())
 
 
 def _load(path: str = DEFAULT_LEDGER_PATH) -> list[dict]:
@@ -48,11 +49,14 @@ def _load(path: str = DEFAULT_LEDGER_PATH) -> list[dict]:
     with open(path, "r", encoding="utf-8") as f:
         for line in f:
             line = line.strip()
-            if line:
-                try:
-                    out.append(json.loads(line))
-                except json.JSONDecodeError:
-                    continue
+            if not line:
+                continue
+            try:
+                out.append(json.loads(line))
+            except json.JSONDecodeError:
+                # Partial/corrupt line (e.g. a write was interrupted). Skip
+                # rather than crash — locking makes this rare, not impossible.
+                continue
     return out
 
 
@@ -62,10 +66,53 @@ def _next_seq(path: str = DEFAULT_LEDGER_PATH) -> int:
     return len(_load(path))
 
 
+def _lock(fd):
+    """Exclusive lock the ledger file handle (POSIX). No-op elsewhere."""
+    if _HAS_FCNTL:
+        fcntl.flock(fd.fileno(), fcntl.LOCK_EX)
+
+
+def _unlock(fd):
+    if _HAS_FCNTL:
+        fcntl.flock(fd.fileno(), fcntl.LOCK_UN)
+
+
 def _append(entry: dict, path: str = DEFAULT_LEDGER_PATH) -> None:
+    """Append a JSON line under an exclusive file lock.
+
+    Holds the lock across _next_seq + write when called via _append_locked
+    so concurrent writers can't collide on sequence numbers or interleave
+    large entries.
+    """
     os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
     with open(path, "a", encoding="utf-8") as f:
-        f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+        _lock(f)
+        try:
+            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+            f.flush()
+            os.fsync(f.fileno())
+        finally:
+            _unlock(f)
+
+
+def _append_with_seq(entry: dict, path: str = DEFAULT_LEDGER_PATH) -> dict:
+    """Atomic: assign seq under lock, then append. Prevents _next_seq races."""
+    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
+    with open(path, "a+", encoding="utf-8") as f:
+        _lock(f)
+        try:
+            f.seek(0, os.SEEK_END)
+            # count existing lines for seq
+            f.seek(0)
+            seq = sum(1 for line in f if line.strip())
+            entry["seq"] = seq
+            f.seek(0, os.SEEK_END)
+            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+            f.flush()
+            os.fsync(f.fileno())
+        finally:
+            _unlock(f)
+    return entry
 
 
 def latest_match(sig: str, path: str = DEFAULT_LEDGER_PATH) -> Optional[dict]:
@@ -100,7 +147,7 @@ def should_post(r: Recommendation, path: str = DEFAULT_LEDGER_PATH) -> tuple[boo
         return True, "first raise of this signature"
 
     status = existing.get("status", "open")
-    window = DEBOUNCE_HOURS.get(status, 72)
+    window = get_debounce_hours(status)
 
     if window is None:
         return False, f"existing entry is {status} — never repost"
@@ -132,12 +179,10 @@ def record(r: Recommendation, path: str = DEFAULT_LEDGER_PATH) -> dict:
         "headline": r.headline,
         "first_raised": _today_iso(),
         "first_raised_ts": int(time.time()),
-        "seq": _next_seq(path),
         "run_id": r.run_id,
         "prior_run_id": r.prior_run_id,
     }
-    _append(entry, path)
-    return entry
+    return _append_with_seq(entry, path)
 
 
 def transition(
@@ -150,14 +195,17 @@ def transition(
     """Append a status transition for an existing work item.
 
     Used by the executor agent when work starts / merges / is dropped.
+    Raises ValueError if new_status is not a known lifecycle status.
     """
+    if new_status not in VALID_STATUSES:
+        raise ValueError(
+            f"unknown status {new_status!r}; expected one of {sorted(VALID_STATUSES)}"
+        )
     entry = {
         "sig": sig,
         "status": new_status,
         "owner": owner or "",
         "due": due or "",
         "transitioned_ts": int(time.time()),
-        "seq": _next_seq(path),
     }
-    _append(entry, path)
-    return entry
+    return _append_with_seq(entry, path)

autopilot/message_formatter.py

@@ -16,8 +16,10 @@
 from __future__ import annotations
 
 from recommendation_contract import Recommendation, SEVERITY_EMOJI, validate
+from config_loader import get_status_tag
 
-STATUS_TAG = {
+# Fallback tags if config is unavailable; live values come from config.yaml.
+_STATUS_TAG_FALLBACK = {
     "open": "🟡 Open",
     "assigned": "🔵 Assigned",
     "inflight": "🟠 In-flight",
@@ -39,7 +41,7 @@ def format(r: Recommendation) -> str:
     steps = "\n".join(f"{i}. {s}" for i, s in enumerate(r.steps, 1))
     owner = r.owner
     due = r.due_date or "TBD"
-    status = STATUS_TAG.get(r.status, "🟡 Open")
+    status = get_status_tag(r.status) or _STATUS_TAG_FALLBACK.get(r.status, "🟡 Open")
     ref = r.prior_run_id if r.prior_run_id else "none"
 
     return (

autopilot/recommendation_contract.py

@@ -19,7 +19,12 @@
 Severity = Literal["P0", "P1", "P2"]
 Status = Literal["open", "assigned", "inflight", "done", "dropped"]
 
+# Lifecycle statuses a ledger entry may hold. Used by ledger.transition() to
+# reject typos that would silently break de-dup (e.g. "assiged").
+VALID_STATUSES = {"open", "assigned", "inflight", "done", "dropped"}
+
 SEVERITY_EMOJI = {"P0": "🔴", "P1": "🟠", "P2": "🟡"}
+# Fallback only — the live value is read from config.yaml via config_loader.
 MAX_STEP_LEN = 120
 RUN_ID_TOKEN = "run_"
 
@@ -47,31 +52,43 @@ def signature(self) -> str:
 
 
 def validate(r: Recommendation) -> tuple[bool, list[str]]:
-    """Return (ok, errors). ok=False => withhold from Slack."""
+    """Return (ok, errors). ok=False => withhold from Slack.
+
+    Severity/step/owner rules are read from config.yaml (message_contract)
+    via config_loader, with stdlib fallbacks if PyYAML/file is absent.
+    """
+    from config_loader import get_contract
+
+    contract = get_contract()
+    max_step_len = contract.get("max_step_len", MAX_STEP_LEN)
+    require_impact_for = set(contract.get("require_impact_for", ["P0", "P1"]))
+    require_due_for = set(contract.get("require_due_for", ["P0", "P1"]))
+    require_owner_for = set(contract.get("require_owner_for", ["P0"]))
+
     errs: list[str] = []
 
     # Headline must not be a raw run_id
     if r.headline.strip().startswith(RUN_ID_TOKEN) or len(r.headline) < 8:
         errs.append("headline must be an outcome, not a run_id")
 
     # Impact required for anything that claims urgency
-    if r.severity in ("P0", "P1") and not r.impact_if_ignored.strip():
+    if r.severity in require_impact_for and not r.impact_if_ignored.strip():
         errs.append(f"{r.severity} missing impact_if_ignored")
 
     # Due date required for P0/P1
-    if r.severity in ("P0", "P1") and not r.due_date.strip():
+    if r.severity in require_due_for and not r.due_date.strip():
         errs.append(f"{r.severity} missing due_date")
 
     # P0 must have an owner; else withhold + escalate
-    if r.severity == "P0" and r.owner == "unassigned":
+    if r.severity in require_owner_for and r.owner == "unassigned":
         errs.append("P0 with no owner — withhold, escalate to #morning-digest")
 
     # Steps: one-line, bounded length, no run_ids
     for i, s in enumerate(r.steps, 1):
         if not s.strip():
             errs.append(f"step {i} empty")
-        if len(s) > MAX_STEP_LEN:
-            errs.append(f"step {i} > {MAX_STEP_LEN} chars (split it)")
+        if len(s) > max_step_len:
+            errs.append(f"step {i} > {max_step_len} chars (split it)")
         if RUN_ID_TOKEN in s:
             errs.append(f"step {i} contains run_id (move to Ref: footer)")