From c836d57e80b1e7c5ee4ee490a3c5448a11141451 Mon Sep 17 00:00:00 2001 From: Perplexity Computer Date: Mon, 6 Jul 2026 04:04:10 +0000 Subject: [PATCH 1/2] feat: config-driven contract + concurrent-writer safety Addresses the deferred review findings from PR #180/#188: 1. [MAJOR] config.yaml blocks are now read (were declared but never used). New config_loader.py reads config.yaml via PyYAML with stdlib fallback, so debounce windows, status tags, max_step_len, and require_* sets are config-driven instead of hardcoded. Silent config drift eliminated. 2. [MAJOR] Concurrent-writer safety: ledger _append_with_seq() now holds an exclusive fcntl.flock across seq-assignment + write, preventing the _next_seq race and large-entry interleaving between the recommender (record) and executor (transition) agents. fsync per write. POSIX-only; no-op fallback on non-fcntl platforms. 3. [MINOR] Status validation: transition() now raises ValueError on unknown statuses (e.g. 'assiged' typo) via VALID_STATUSES, so a typo can't silently break de-dup. Tests: 14/14 passing (added 4: config-overrides-max_step_len, config-overrides-debounce, transition-rejects-unknown-status, concurrent-writers-do-not-lose-entries with 8x25 parallel writes). --- autopilot/README.md | 9 +- autopilot/config_loader.py | 112 ++++++++++++++++ autopilot/decisions/ledger.py | 122 ++++++++++++------ autopilot/message_formatter.py | 7 +- autopilot/recommendation_contract.py | 46 ++++--- .../tests/test_recommendation_contract.py | 120 ++++++++++++++++- 6 files changed, 346 insertions(+), 70 deletions(-) create mode 100644 autopilot/config_loader.py 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..bd96590 --- /dev/null +++ b/autopilot/config_loader.py @@ -0,0 +1,112 @@ +"""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..7103a4e 100644 --- a/autopilot/decisions/ledger.py +++ b/autopilot/decisions/ledger.py @@ -8,18 +8,28 @@ 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: @@ -28,18 +38,9 @@ def _today_iso() -> str: DEFAULT_LEDGER_PATH = os.environ.get( - "DECISIONS_LEDGER_PATH", "autopilot/decisions/recommendations.jsonl" + "DECISIONS_LEDGER_PATH", get_ledger_path() ) -# 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) -} - def _load(path: str = DEFAULT_LEDGER_PATH) -> list[dict]: if not os.path.exists(path): @@ -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]: @@ -77,13 +124,9 @@ def latest_match(sig: str, path: str = DEFAULT_LEDGER_PATH) -> Optional[dict]: matches = [e for e in _load(path) if e.get("sig") == sig] if not matches: return None - return max( - matches, - key=lambda e: ( - max(e.get("first_raised_ts", 0), e.get("transitioned_ts", 0)), - e.get("seq", 0), - ), - ) + return max(matches, key=lambda e: (max(e.get("first_raised_ts", 0), + e.get("transitioned_ts", 0)), + e.get("seq", 0))) def should_post(r: Recommendation, path: str = DEFAULT_LEDGER_PATH) -> tuple[bool, str]: @@ -100,7 +143,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" @@ -109,9 +152,8 @@ def should_post(r: Recommendation, path: str = DEFAULT_LEDGER_PATH) -> tuple[boo # window restarts when an item is reassigned/moved to in-flight. Without # this, transition entries (which only carry transitioned_ts) default to # first_raised_ts=0 -> ~495k hours elapsed -> always reposts. - latest_ts = max( - existing.get("first_raised_ts", 0), existing.get("transitioned_ts", 0) - ) + latest_ts = max(existing.get("first_raised_ts", 0), + existing.get("transitioned_ts", 0)) elapsed_h = (time.time() - latest_ts) / 3600.0 if elapsed_h < window: return False, ( @@ -132,32 +174,30 @@ 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( - sig: str, - new_status: str, - owner: Optional[str] = None, - due: Optional[str] = None, - path: str = DEFAULT_LEDGER_PATH, + sig: str, new_status: str, owner: Optional[str] = None, + due: Optional[str] = None, path: str = DEFAULT_LEDGER_PATH, ) -> dict: """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..960c70d 100644 --- a/autopilot/message_formatter.py +++ b/autopilot/message_formatter.py @@ -12,12 +12,13 @@ This replaces the old bot format that buried the subject behind a run_id. """ - 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 +40,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..bcbeed5 100644 --- a/autopilot/recommendation_contract.py +++ b/autopilot/recommendation_contract.py @@ -10,7 +10,6 @@ Wired into the DRC recommendation pipeline: a recommendation that fails validate() is withheld (P0) or warned (P1) before posting to Slack. """ - from __future__ import annotations from dataclasses import dataclass, field @@ -19,7 +18,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_" @@ -27,18 +31,18 @@ @dataclass class Recommendation: severity: Severity - headline: str # outcome, NOT run_id - impact_if_ignored: str # required for P0/P1 + headline: str # outcome, NOT run_id + impact_if_ignored: str # required for P0/P1 steps: list[str] - due_date: str # YYYY-MM-DD or "" - owner: str = "unassigned" # @user_id or "unassigned" + due_date: str # YYYY-MM-DD or "" + owner: str = "unassigned" # @user_id or "unassigned" status: Status = "open" - blocked_by: str = "" # prerequisite, "" if none + blocked_by: str = "" # prerequisite, "" if none target_repo: str = "" file_or_workflow_path: str = "" - action_verb: str = "" # "create" | "configure" | "rotate" | ... - prior_run_id: str = "" # for Ref: footer + de-dup - run_id: str = "" # current raise's id (footer only) + action_verb: str = "" # "create" | "configure" | "rotate" | ... + prior_run_id: str = "" # for Ref: footer + de-dup + run_id: str = "" # current raise's id (footer only) def signature(self) -> str: """Normalised work-item signature for de-dup matching.""" @@ -47,7 +51,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 +71,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..988ba88 100644 --- a/autopilot/tests/test_recommendation_contract.py +++ b/autopilot/tests/test_recommendation_contract.py @@ -3,7 +3,6 @@ Run: python -m pytest autopilot/tests/test_recommendation_contract.py -q or: python autopilot/tests/test_recommendation_contract.py """ - import os import sys import tempfile @@ -77,7 +76,7 @@ def test_ledger_debounce_suppresses_duplicate(): path = tf.name try: r = _good() - assert should_post(r, path)[0] is True # first raise allowed + assert should_post(r, path)[0] is True # first raise allowed record(r, path) # immediate re-raise of same signature -> suppressed @@ -95,7 +94,6 @@ def test_done_status_blocks_repost(): r = _good() record(r, path) from decisions.ledger import transition - transition(r.signature(), "done", path=path) allow, reason = should_post(r, path) @@ -115,10 +113,8 @@ def test_assigned_status_debounces_repost(): r = _good() record(r, path) from decisions.ledger import transition - - transition( - r.signature(), "assigned", owner="U0AKJK1J7GR", due="2026-07-05", path=path - ) + transition(r.signature(), "assigned", owner="U0AKJK1J7GR", + due="2026-07-05", path=path) # Immediate re-raise while assigned must be suppressed. allow, reason = should_post(r, path) @@ -143,6 +139,116 @@ 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 From 952922a49f436595aa6f232b34bd565d156df545 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Mon, 6 Jul 2026 04:04:53 +0000 Subject: [PATCH 2/2] style: auto-fix pre-commit issues [skip ci] --- autopilot/config_loader.py | 9 ++---- autopilot/decisions/ledger.py | 28 ++++++++++------ autopilot/message_formatter.py | 1 + autopilot/recommendation_contract.py | 17 +++++----- .../tests/test_recommendation_contract.py | 32 ++++++++++++++----- 5 files changed, 55 insertions(+), 32 deletions(-) diff --git a/autopilot/config_loader.py b/autopilot/config_loader.py index bd96590..d00bae4 100644 --- a/autopilot/config_loader.py +++ b/autopilot/config_loader.py @@ -5,15 +5,14 @@ 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" -) +DEFAULT_CONFIG_PATH = os.environ.get("AUTOPILOT_CONFIG_PATH", "autopilot/config.yaml") DEFAULTS: dict[str, Any] = { "recommendation_debounce": { @@ -84,9 +83,7 @@ def load_config(path: Optional[str] = None) -> dict[str, Any]: 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, {}) - ) + policy = load_config()["recommendation_debounce"]["repost_policy"].get(status, {}) if policy.get("never_repost"): return None return policy.get("min_hours_between", 72) diff --git a/autopilot/decisions/ledger.py b/autopilot/decisions/ledger.py index 7103a4e..513940a 100644 --- a/autopilot/decisions/ledger.py +++ b/autopilot/decisions/ledger.py @@ -14,6 +14,7 @@ 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 @@ -27,6 +28,7 @@ try: import fcntl # POSIX only + _HAS_FCNTL = True except ImportError: _HAS_FCNTL = False @@ -37,9 +39,7 @@ def _today_iso() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%d") -DEFAULT_LEDGER_PATH = os.environ.get( - "DECISIONS_LEDGER_PATH", get_ledger_path() -) +DEFAULT_LEDGER_PATH = os.environ.get("DECISIONS_LEDGER_PATH", get_ledger_path()) def _load(path: str = DEFAULT_LEDGER_PATH) -> list[dict]: @@ -124,9 +124,13 @@ def latest_match(sig: str, path: str = DEFAULT_LEDGER_PATH) -> Optional[dict]: matches = [e for e in _load(path) if e.get("sig") == sig] if not matches: return None - return max(matches, key=lambda e: (max(e.get("first_raised_ts", 0), - e.get("transitioned_ts", 0)), - e.get("seq", 0))) + return max( + matches, + key=lambda e: ( + max(e.get("first_raised_ts", 0), e.get("transitioned_ts", 0)), + e.get("seq", 0), + ), + ) def should_post(r: Recommendation, path: str = DEFAULT_LEDGER_PATH) -> tuple[bool, str]: @@ -152,8 +156,9 @@ def should_post(r: Recommendation, path: str = DEFAULT_LEDGER_PATH) -> tuple[boo # window restarts when an item is reassigned/moved to in-flight. Without # this, transition entries (which only carry transitioned_ts) default to # first_raised_ts=0 -> ~495k hours elapsed -> always reposts. - latest_ts = max(existing.get("first_raised_ts", 0), - existing.get("transitioned_ts", 0)) + latest_ts = max( + existing.get("first_raised_ts", 0), existing.get("transitioned_ts", 0) + ) elapsed_h = (time.time() - latest_ts) / 3600.0 if elapsed_h < window: return False, ( @@ -181,8 +186,11 @@ def record(r: Recommendation, path: str = DEFAULT_LEDGER_PATH) -> dict: def transition( - sig: str, new_status: str, owner: Optional[str] = None, - due: Optional[str] = None, path: str = DEFAULT_LEDGER_PATH, + sig: str, + new_status: str, + owner: Optional[str] = None, + due: Optional[str] = None, + path: str = DEFAULT_LEDGER_PATH, ) -> dict: """Append a status transition for an existing work item. diff --git a/autopilot/message_formatter.py b/autopilot/message_formatter.py index 960c70d..6462aed 100644 --- a/autopilot/message_formatter.py +++ b/autopilot/message_formatter.py @@ -12,6 +12,7 @@ This replaces the old bot format that buried the subject behind a run_id. """ + from __future__ import annotations from recommendation_contract import Recommendation, SEVERITY_EMOJI, validate diff --git a/autopilot/recommendation_contract.py b/autopilot/recommendation_contract.py index bcbeed5..8cd3887 100644 --- a/autopilot/recommendation_contract.py +++ b/autopilot/recommendation_contract.py @@ -10,6 +10,7 @@ Wired into the DRC recommendation pipeline: a recommendation that fails validate() is withheld (P0) or warned (P1) before posting to Slack. """ + from __future__ import annotations from dataclasses import dataclass, field @@ -31,18 +32,18 @@ @dataclass class Recommendation: severity: Severity - headline: str # outcome, NOT run_id - impact_if_ignored: str # required for P0/P1 + headline: str # outcome, NOT run_id + impact_if_ignored: str # required for P0/P1 steps: list[str] - due_date: str # YYYY-MM-DD or "" - owner: str = "unassigned" # @user_id or "unassigned" + due_date: str # YYYY-MM-DD or "" + owner: str = "unassigned" # @user_id or "unassigned" status: Status = "open" - blocked_by: str = "" # prerequisite, "" if none + blocked_by: str = "" # prerequisite, "" if none target_repo: str = "" file_or_workflow_path: str = "" - action_verb: str = "" # "create" | "configure" | "rotate" | ... - prior_run_id: str = "" # for Ref: footer + de-dup - run_id: str = "" # current raise's id (footer only) + action_verb: str = "" # "create" | "configure" | "rotate" | ... + prior_run_id: str = "" # for Ref: footer + de-dup + run_id: str = "" # current raise's id (footer only) def signature(self) -> str: """Normalised work-item signature for de-dup matching.""" diff --git a/autopilot/tests/test_recommendation_contract.py b/autopilot/tests/test_recommendation_contract.py index 988ba88..87474de 100644 --- a/autopilot/tests/test_recommendation_contract.py +++ b/autopilot/tests/test_recommendation_contract.py @@ -3,6 +3,7 @@ Run: python -m pytest autopilot/tests/test_recommendation_contract.py -q or: python autopilot/tests/test_recommendation_contract.py """ + import os import sys import tempfile @@ -76,7 +77,7 @@ def test_ledger_debounce_suppresses_duplicate(): path = tf.name try: r = _good() - assert should_post(r, path)[0] is True # first raise allowed + assert should_post(r, path)[0] is True # first raise allowed record(r, path) # immediate re-raise of same signature -> suppressed @@ -94,6 +95,7 @@ def test_done_status_blocks_repost(): r = _good() record(r, path) from decisions.ledger import transition + transition(r.signature(), "done", path=path) allow, reason = should_post(r, path) @@ -113,8 +115,10 @@ def test_assigned_status_debounces_repost(): r = _good() record(r, path) from decisions.ledger import transition - transition(r.signature(), "assigned", owner="U0AKJK1J7GR", - due="2026-07-05", path=path) + + transition( + r.signature(), "assigned", owner="U0AKJK1J7GR", due="2026-07-05", path=path + ) # Immediate re-raise while assigned must be suppressed. allow, reason = should_post(r, path) @@ -141,12 +145,15 @@ def test_formatter_rejects_invalid(): # --- 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") + mode="w", suffix=".yaml", delete=False, encoding="utf-8" + ) cfg.write("message_contract:\n max_step_len: 20\n") cfg.close() try: @@ -165,9 +172,11 @@ def test_config_overrides_max_step_len(): 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") + mode="w", suffix=".yaml", delete=False, encoding="utf-8" + ) cfg.write( "recommendation_debounce:\n" " repost_policy:\n" @@ -181,6 +190,7 @@ def test_config_overrides_debounce_window(): 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 @@ -192,13 +202,16 @@ def test_config_overrides_debounce_window(): # --- 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 @@ -211,10 +224,12 @@ def test_transition_rejects_unknown_status(): # --- 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: @@ -238,10 +253,11 @@ def writer(): 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)" - ) + 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"