diff --git a/autopilot/README.md b/autopilot/README.md
index 093ba87..a79cc65 100644
--- a/autopilot/README.md
+++ b/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
diff --git a/autopilot/config_loader.py b/autopilot/config_loader.py
new file mode 100644
index 0000000..d00bae4
--- /dev/null
+++ b/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
diff --git a/autopilot/decisions/ledger.py b/autopilot/decisions/ledger.py
index d5d378c..513940a 100644
--- a/autopilot/decisions/ledger.py
+++ b/autopilot/decisions/ledger.py
@@ -8,6 +8,11 @@
 
 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
@@ -15,11 +20,18 @@
 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:
@@ -27,18 +39,7 @@ def _today_iso() -> str:
     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)
diff --git a/autopilot/message_formatter.py b/autopilot/message_formatter.py
index 7d22557..6462aed 100644
--- a/autopilot/message_formatter.py
+++ b/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 (
diff --git a/autopilot/recommendation_contract.py b/autopilot/recommendation_contract.py
index fb83f3d..8cd3887 100644
--- a/autopilot/recommendation_contract.py
+++ b/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,7 +52,19 @@ 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
@@ -55,23 +72,23 @@ def validate(r: Recommendation) -> tuple[bool, list[str]]:
         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)")
 
diff --git a/autopilot/tests/test_recommendation_contract.py b/autopilot/tests/test_recommendation_contract.py
index ca4ba15..87474de 100644
--- a/autopilot/tests/test_recommendation_contract.py
+++ b/autopilot/tests/test_recommendation_contract.py
@@ -143,6 +143,128 @@ def test_formatter_rejects_invalid():
     assert "REJECTED" in msg
 
 
+# --- New: config-driven behaviour ---
+
+
+def test_config_overrides_max_step_len():
+    """validate() should honour max_step_len from a custom config file."""
+    import config_loader
+
+    config_loader.reset_cache()
+    cfg = tempfile.NamedTemporaryFile(
+        mode="w", suffix=".yaml", delete=False, encoding="utf-8"
+    )
+    cfg.write("message_contract:\n  max_step_len: 20\n")
+    cfg.close()
+    try:
+        config_loader.DEFAULT_CONFIG_PATH = cfg.name
+        config_loader.reset_cache()
+        r = _good()
+        r.steps[0] = "x" * 25  # over the custom 20 limit, under default 120
+        ok, errs = validate(r)
+        assert not ok
+        assert any("step 1" in e and "20" in e for e in errs)
+    finally:
+        config_loader.reset_cache()
+        os.unlink(cfg.name)
+
+
+def test_config_overrides_debounce_window():
+    """should_post() should honour repost_policy from a custom config file."""
+    import config_loader
+
+    config_loader.reset_cache()
+    cfg = tempfile.NamedTemporaryFile(
+        mode="w", suffix=".yaml", delete=False, encoding="utf-8"
+    )
+    cfg.write(
+        "recommendation_debounce:\n"
+        "  repost_policy:\n"
+        "    open: {min_hours_between: 100000, never_repost: false}\n"
+    )
+    cfg.close()
+    ledger = tempfile.NamedTemporaryFile(suffix=".jsonl", delete=False)
+    ledger.close()
+    try:
+        config_loader.DEFAULT_CONFIG_PATH = cfg.name
+        config_loader.reset_cache()
+        r = _good()
+        from decisions.ledger import record, should_post
+
+        record(r, ledger.name)
+        allow, reason = should_post(r, ledger.name)
+        assert allow is False, reason  # 100000h window always suppresses
+    finally:
+        config_loader.reset_cache()
+        os.unlink(cfg.name)
+        os.unlink(ledger.name)
+
+
+# --- New: status validation ---
+
+
+def test_transition_rejects_unknown_status():
+    from decisions.ledger import transition
+
+    ledger = tempfile.NamedTemporaryFile(suffix=".jsonl", delete=False)
+    ledger.close()
+    try:
+        r = _good()
+        from decisions.ledger import record
+
+        record(r, ledger.name)
+        try:
+            transition(r.signature(), "assiged", path=ledger.name)  # typo
+            assert False, "expected ValueError for unknown status"
+        except ValueError as e:
+            assert "assiged" in str(e)
+    finally:
+        os.unlink(ledger.name)
+
+
+# --- New: concurrent-writer safety ---
+
+
+def test_concurrent_writers_do_not_lose_entries():
+    """Parallel record() calls must all land — file locking prevents races."""
+    import threading
+    from decisions.ledger import record
+
+    ledger = tempfile.NamedTemporaryFile(suffix=".jsonl", delete=False)
+    ledger.close()
+    try:
+        r = _good()
+        n_writers = 8
+        n_each = 25
+        errors = []
+
+        def writer():
+            try:
+                for _ in range(n_each):
+                    record(r, ledger.name)
+            except Exception as e:  # noqa
+                errors.append(e)
+
+        threads = [threading.Thread(target=writer) for _ in range(n_writers)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        assert errors == [], errors
+        from decisions.ledger import _load
+
+        entries = _load(ledger.name)
+        assert (
+            len(entries) == n_writers * n_each
+        ), f"expected {n_writers * n_each}, got {len(entries)} (lost entries)"
+        # seq values should be unique 0..N-1 (no collisions from the race)
+        seqs = sorted(e.get("seq", -1) for e in entries)
+        assert seqs == list(range(n_writers * n_each)), "seq collisions detected"
+    finally:
+        os.unlink(ledger.name)
+
+
 if __name__ == "__main__":
     fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
     passed = 0