From 036b78e31ea5d036452b3a52e0e086b01e0c763f Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 9 Jul 2026 22:27:51 -0400 Subject: [PATCH 1/4] feat: control-plane DB substrate for atomic assign/lease (Closes #613) Add SQLite single-writer MVP control plane per the allocator ADR: sessions, work_items (issue/pr only), atomic assign+lease, mutation gates, terminal-lock index, and provider-neutral incident_links. DB coordinates concurrency; Gitea remains assignable work; raw Sentry/GlitchTip incidents are never work items. #600 and #612 build on this substrate. --- control_plane_db.py | 944 ++++++++++++++++++ .../control-plane-db-substrate.md | 48 + tests/test_control_plane_db.py | 377 +++++++ 3 files changed, 1369 insertions(+) create mode 100644 control_plane_db.py create mode 100644 docs/architecture/control-plane-db-substrate.md create mode 100644 tests/test_control_plane_db.py diff --git a/control_plane_db.py b/control_plane_db.py new file mode 100644 index 0000000..adaa542 --- /dev/null +++ b/control_plane_db.py @@ -0,0 +1,944 @@ +"""Control-plane DB substrate for multi-session coordination (#613). + +Implements the durable coordination store described by +``docs/architecture/mcp-allocator-control-plane-observability-adr.md``: + +* DB coordinates live concurrency (sessions, atomic assignment+lease, + heartbeats, terminal-lock index, events, ``incident_links``). +* Gitea remains the durable work record and the only assignable work unit + (``issue`` / ``pr`` — never raw Sentry/GlitchTip incidents). +* SQLite is the single-writer MVP backend; the API is backend-shaped so a + shared Postgres or single allocator daemon can replace the connection + layer later without changing call sites. + +This module is the **substrate** for #600 (allocator policy/tool) and #612 +(incident bridge). It does **not** implement ``gitea_allocate_next_work`` +routing policy or provider adapters. +""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import threading +import time +import uuid +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any, Iterator, Sequence + +SCHEMA_VERSION = 1 + +# Assignable work kinds only — raw monitoring incidents are never work items. +WORK_KINDS = frozenset({"issue", "pr"}) + +DEFAULT_LEASE_TTL_SECONDS = 4 * 3600 + +# Environment: path for SQLite MVP (single-writer). +DB_PATH_ENV = "GITEA_CONTROL_PLANE_DB" +DEFAULT_DB_PATH = os.path.expanduser("~/.cache/gitea-tools/control-plane/control_plane.sqlite3") + +_SCHEMA_SQL = """ +PRAGMA foreign_keys = ON; + +CREATE TABLE IF NOT EXISTS schema_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS sessions ( + session_id TEXT PRIMARY KEY, + role TEXT NOT NULL, + profile TEXT, + namespace TEXT, + pid INTEGER, + started_at TEXT NOT NULL, + last_heartbeat_at TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active' +); + +CREATE TABLE IF NOT EXISTS work_items ( + work_item_id INTEGER PRIMARY KEY AUTOINCREMENT, + remote TEXT NOT NULL, + org TEXT NOT NULL, + repo TEXT NOT NULL, + kind TEXT NOT NULL CHECK (kind IN ('issue', 'pr')), + number INTEGER NOT NULL, + state TEXT NOT NULL DEFAULT 'open', + priority INTEGER NOT NULL DEFAULT 0, + current_head_sha TEXT, + updated_at TEXT NOT NULL, + UNIQUE (remote, org, repo, kind, number) +); + +CREATE TABLE IF NOT EXISTS leases ( + lease_id TEXT PRIMARY KEY, + work_item_id INTEGER NOT NULL REFERENCES work_items(work_item_id), + session_id TEXT NOT NULL REFERENCES sessions(session_id), + role TEXT NOT NULL, + phase TEXT NOT NULL DEFAULT 'claimed', + expires_at TEXT NOT NULL, + heartbeat_at TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active' +); + +CREATE TABLE IF NOT EXISTS assignments ( + assignment_id TEXT PRIMARY KEY, + work_item_id INTEGER NOT NULL REFERENCES work_items(work_item_id), + session_id TEXT NOT NULL REFERENCES sessions(session_id), + lease_id TEXT NOT NULL REFERENCES leases(lease_id), + allowed_actions TEXT NOT NULL, + forbidden_actions TEXT NOT NULL, + expected_head_sha TEXT, + role TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active', + created_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS terminal_locks ( + terminal_lock_id INTEGER PRIMARY KEY AUTOINCREMENT, + remote TEXT NOT NULL, + org TEXT NOT NULL, + repo TEXT NOT NULL, + terminal_pr INTEGER NOT NULL, + review_id TEXT, + decision TEXT, + status TEXT NOT NULL DEFAULT 'active', + cleanup_state TEXT, + created_at TEXT NOT NULL, + UNIQUE (remote, org, repo, terminal_pr) +); + +CREATE TABLE IF NOT EXISTS events ( + event_id INTEGER PRIMARY KEY AUTOINCREMENT, + work_item_id INTEGER REFERENCES work_items(work_item_id), + event_type TEXT NOT NULL, + message TEXT NOT NULL, + created_at TEXT NOT NULL +); + +-- Provider-neutral incident ↔ Gitea link model (ADR §8–9). Not assignable work. +CREATE TABLE IF NOT EXISTS incident_links ( + link_id INTEGER PRIMARY KEY AUTOINCREMENT, + provider TEXT NOT NULL, + provider_base_url TEXT, + provider_org TEXT, + provider_project TEXT, + provider_issue_id TEXT NOT NULL, + provider_short_id TEXT, + provider_permalink TEXT, + fingerprint TEXT, + gitea_org TEXT NOT NULL, + gitea_repo TEXT NOT NULL, + gitea_issue_number INTEGER NOT NULL, + linked_pr_numbers TEXT, + first_seen TEXT, + last_seen TEXT, + event_count INTEGER, + status TEXT NOT NULL DEFAULT 'open', + release_resolved_at TEXT, + last_sync_at TEXT, + UNIQUE (provider, provider_base_url, provider_org, provider_project, provider_issue_id) +); + +CREATE INDEX IF NOT EXISTS idx_leases_work_status ON leases(work_item_id, status); +CREATE INDEX IF NOT EXISTS idx_assignments_session ON assignments(session_id, status); +CREATE INDEX IF NOT EXISTS idx_incident_gitea ON incident_links(gitea_org, gitea_repo, gitea_issue_number); +""" + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _ts(dt: datetime | None = None) -> str: + value = dt or _utc_now() + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def _parse_ts(value: str | None) -> datetime | None: + if not value: + return None + text = value.strip() + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + return datetime.fromisoformat(text) + except ValueError: + return None + + +def default_db_path() -> str: + raw = (os.environ.get(DB_PATH_ENV) or DEFAULT_DB_PATH).strip() + return raw or DEFAULT_DB_PATH + + +class ControlPlaneError(RuntimeError): + """Base error for control-plane substrate failures.""" + + +class InvalidWorkKindError(ControlPlaneError): + """Raised when a non-assignable work kind is requested.""" + + +class LeaseRequiredError(ControlPlaneError): + """Raised when a mutation is attempted without a valid assignment/lease.""" + + +class ForeignLeaseError(ControlPlaneError): + """Raised when another session holds the active lease.""" + + +@dataclass(frozen=True) +class AssignmentResult: + """Result of an atomic assign+lease transaction.""" + + outcome: str # assigned | wait | no_safe_work + assignment_id: str | None = None + lease_id: str | None = None + session_id: str | None = None + role: str | None = None + work_kind: str | None = None + work_number: int | None = None + remote: str | None = None + org: str | None = None + repo: str | None = None + expected_head_sha: str | None = None + allowed_actions: tuple[str, ...] = () + forbidden_actions: tuple[str, ...] = () + expires_at: str | None = None + owner_session_id: str | None = None + reason: str = "" + + def as_dict(self) -> dict[str, Any]: + return { + "outcome": self.outcome, + "assignment_id": self.assignment_id, + "lease_id": self.lease_id, + "session_id": self.session_id, + "role": self.role, + "work_kind": self.work_kind, + "work_number": self.work_number, + "remote": self.remote, + "org": self.org, + "repo": self.repo, + "expected_head_sha": self.expected_head_sha, + "allowed_actions": list(self.allowed_actions), + "forbidden_actions": list(self.forbidden_actions), + "expires_at": self.expires_at, + "owner_session_id": self.owner_session_id, + "reason": self.reason, + } + + +class ControlPlaneDB: + """SQLite single-writer MVP control-plane store. + + Use one process as writer for multi-session safety claims, or migrate to + Postgres / a single allocator daemon for multi-host production (ADR §6). + """ + + def __init__(self, db_path: str | None = None) -> None: + self.db_path = (db_path or default_db_path()).strip() + parent = os.path.dirname(self.db_path) + if parent: + os.makedirs(parent, mode=0o700, exist_ok=True) + self._lock = threading.RLock() + self._init_schema() + + def _connect(self) -> sqlite3.Connection: + # Default isolation (DEFERRED) so explicit BEGIN IMMEDIATE works. + conn = sqlite3.connect(self.db_path, timeout=30) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys = ON") + # Serialize writers even across threads in one process. + conn.execute("PRAGMA journal_mode = WAL") + return conn + + @contextmanager + def _tx(self, immediate: bool = True) -> Iterator[sqlite3.Connection]: + with self._lock: + conn = self._connect() + try: + if immediate: + conn.execute("BEGIN IMMEDIATE") + else: + conn.execute("BEGIN") + yield conn + conn.commit() + except Exception: + try: + conn.rollback() + except sqlite3.Error: + pass + raise + finally: + conn.close() + + def _init_schema(self) -> None: + # executescript auto-commits; run schema outside an open txn, then meta. + with self._lock: + conn = self._connect() + try: + conn.executescript(_SCHEMA_SQL) + conn.execute( + "INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)", + ("schema_version", str(SCHEMA_VERSION)), + ) + conn.execute( + "INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)", + ( + "architecture", + "DB coordinates; Gitea records; Sentry/GlitchTip observe; " + "bridge is only path from observations to Gitea work", + ), + ) + conn.commit() + finally: + conn.close() + + # ── sessions ────────────────────────────────────────────────────────── + + def upsert_session( + self, + *, + session_id: str, + role: str, + profile: str | None = None, + namespace: str | None = None, + pid: int | None = None, + status: str = "active", + ) -> dict[str, Any]: + now = _ts() + with self._tx() as conn: + existing = conn.execute( + "SELECT session_id FROM sessions WHERE session_id = ?", + (session_id,), + ).fetchone() + if existing: + conn.execute( + """ + UPDATE sessions + SET role = ?, profile = ?, namespace = ?, pid = ?, + last_heartbeat_at = ?, status = ? + WHERE session_id = ? + """, + (role, profile, namespace, pid, now, status, session_id), + ) + else: + conn.execute( + """ + INSERT INTO sessions( + session_id, role, profile, namespace, pid, + started_at, last_heartbeat_at, status + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + (session_id, role, profile, namespace, pid, now, now, status), + ) + row = conn.execute( + "SELECT * FROM sessions WHERE session_id = ?", + (session_id,), + ).fetchone() + return dict(row) + + def heartbeat_session(self, session_id: str) -> None: + with self._tx() as conn: + conn.execute( + "UPDATE sessions SET last_heartbeat_at = ? WHERE session_id = ?", + (_ts(), session_id), + ) + + # ── work items ──────────────────────────────────────────────────────── + + def upsert_work_item( + self, + *, + remote: str, + org: str, + repo: str, + kind: str, + number: int, + state: str = "open", + priority: int = 0, + current_head_sha: str | None = None, + ) -> int: + kind_norm = (kind or "").strip().lower() + if kind_norm not in WORK_KINDS: + raise InvalidWorkKindError( + f"work kind '{kind}' is not assignable; only {sorted(WORK_KINDS)} " + f"are allowed (raw monitoring incidents are never work items)" + ) + now = _ts() + with self._tx() as conn: + conn.execute( + """ + INSERT INTO work_items( + remote, org, repo, kind, number, state, priority, + current_head_sha, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(remote, org, repo, kind, number) DO UPDATE SET + state = excluded.state, + priority = excluded.priority, + current_head_sha = excluded.current_head_sha, + updated_at = excluded.updated_at + """, + ( + remote, + org, + repo, + kind_norm, + int(number), + state, + int(priority), + current_head_sha, + now, + ), + ) + row = conn.execute( + """ + SELECT work_item_id FROM work_items + WHERE remote = ? AND org = ? AND repo = ? AND kind = ? AND number = ? + """, + (remote, org, repo, kind_norm, int(number)), + ).fetchone() + return int(row["work_item_id"]) + + # ── atomic assign + lease ───────────────────────────────────────────── + + def assign_and_lease( + self, + *, + session_id: str, + role: str, + remote: str, + org: str, + repo: str, + kind: str, + number: int, + expected_head_sha: str | None = None, + allowed_actions: Sequence[str] | None = None, + forbidden_actions: Sequence[str] | None = None, + lease_ttl_seconds: int = DEFAULT_LEASE_TTL_SECONDS, + phase: str = "claimed", + now: datetime | None = None, + ) -> AssignmentResult: + """Atomically create assignment + lease for one Gitea work item. + + Concurrent sessions cannot both receive the same open work item. + Returns ``wait`` with owner_session_id when a live foreign lease exists. + """ + kind_norm = (kind or "").strip().lower() + if kind_norm not in WORK_KINDS: + raise InvalidWorkKindError( + f"cannot assign kind '{kind}'; only {sorted(WORK_KINDS)}" + ) + + allowed = tuple(allowed_actions or ("implement", "comment")) + forbidden = tuple( + forbidden_actions + or ("approve", "merge", "self_select_without_assignment") + ) + moment = now or _utc_now() + now_s = _ts(moment) + expires = _ts(moment + timedelta(seconds=int(lease_ttl_seconds))) + + with self._tx(immediate=True) as conn: + # Ensure session exists + sess = conn.execute( + "SELECT session_id FROM sessions WHERE session_id = ?", + (session_id,), + ).fetchone() + if not sess: + conn.execute( + """ + INSERT INTO sessions( + session_id, role, profile, namespace, pid, + started_at, last_heartbeat_at, status + ) VALUES (?, ?, NULL, NULL, ?, ?, ?, 'active') + """, + (session_id, role, os.getpid(), now_s, now_s), + ) + + # Upsert work item + conn.execute( + """ + INSERT INTO work_items( + remote, org, repo, kind, number, state, priority, + current_head_sha, updated_at + ) VALUES (?, ?, ?, ?, ?, 'open', 0, ?, ?) + ON CONFLICT(remote, org, repo, kind, number) DO UPDATE SET + current_head_sha = COALESCE(excluded.current_head_sha, work_items.current_head_sha), + updated_at = excluded.updated_at + """, + (remote, org, repo, kind_norm, int(number), expected_head_sha, now_s), + ) + work = conn.execute( + """ + SELECT * FROM work_items + WHERE remote = ? AND org = ? AND repo = ? AND kind = ? AND number = ? + """, + (remote, org, repo, kind_norm, int(number)), + ).fetchone() + work_item_id = int(work["work_item_id"]) + state = (work["state"] or "open").lower() + if state in ("merged", "closed"): + return AssignmentResult( + outcome="no_safe_work", + reason=f"work item {kind_norm}#{number} is {state}; never assign", + ) + + # Expire stale leases on this work item first + self._expire_stale_leases_conn(conn, work_item_id=work_item_id, now_s=now_s) + + active = conn.execute( + """ + SELECT * FROM leases + WHERE work_item_id = ? AND status = 'active' + ORDER BY expires_at DESC + LIMIT 1 + """, + (work_item_id,), + ).fetchone() + if active: + owner = active["session_id"] + if owner == session_id: + # Owner-resume: refresh heartbeat and return existing assignment + conn.execute( + """ + UPDATE leases + SET heartbeat_at = ?, expires_at = ?, phase = ? + WHERE lease_id = ? + """, + (now_s, expires, phase, active["lease_id"]), + ) + asn = conn.execute( + """ + SELECT * FROM assignments + WHERE lease_id = ? AND status = 'active' + ORDER BY created_at DESC LIMIT 1 + """, + (active["lease_id"],), + ).fetchone() + if asn: + return AssignmentResult( + outcome="assigned", + assignment_id=asn["assignment_id"], + lease_id=active["lease_id"], + session_id=session_id, + role=asn["role"], + work_kind=kind_norm, + work_number=int(number), + remote=remote, + org=org, + repo=repo, + expected_head_sha=asn["expected_head_sha"], + allowed_actions=tuple(json.loads(asn["allowed_actions"])), + forbidden_actions=tuple(json.loads(asn["forbidden_actions"])), + expires_at=expires, + owner_session_id=session_id, + reason="owner-resume: refreshed existing lease", + ) + return AssignmentResult( + outcome="wait", + owner_session_id=owner, + reason=( + f"foreign active lease held by session {owner} on " + f"{kind_norm}#{number}" + ), + ) + + lease_id = f"lease-{uuid.uuid4().hex[:16]}" + assignment_id = f"asn-{uuid.uuid4().hex[:16]}" + conn.execute( + """ + INSERT INTO leases( + lease_id, work_item_id, session_id, role, phase, + expires_at, heartbeat_at, status + ) VALUES (?, ?, ?, ?, ?, ?, ?, 'active') + """, + (lease_id, work_item_id, session_id, role, phase, expires, now_s), + ) + conn.execute( + """ + INSERT INTO assignments( + assignment_id, work_item_id, session_id, lease_id, + allowed_actions, forbidden_actions, expected_head_sha, + role, status, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'active', ?) + """, + ( + assignment_id, + work_item_id, + session_id, + lease_id, + json.dumps(list(allowed)), + json.dumps(list(forbidden)), + expected_head_sha or work["current_head_sha"], + role, + now_s, + ), + ) + conn.execute( + """ + INSERT INTO events(work_item_id, event_type, message, created_at) + VALUES (?, 'assigned', ?, ?) + """, + ( + work_item_id, + f"session {session_id} assigned {kind_norm}#{number} lease={lease_id}", + now_s, + ), + ) + return AssignmentResult( + outcome="assigned", + assignment_id=assignment_id, + lease_id=lease_id, + session_id=session_id, + role=role, + work_kind=kind_norm, + work_number=int(number), + remote=remote, + org=org, + repo=repo, + expected_head_sha=expected_head_sha or work["current_head_sha"], + allowed_actions=allowed, + forbidden_actions=forbidden, + expires_at=expires, + owner_session_id=session_id, + reason="atomic assign+lease created", + ) + + def _expire_stale_leases_conn( + self, + conn: sqlite3.Connection, + *, + work_item_id: int | None = None, + now_s: str | None = None, + ) -> int: + now_s = now_s or _ts() + if work_item_id is None: + rows = conn.execute( + "SELECT lease_id FROM leases WHERE status = 'active' AND expires_at <= ?", + (now_s,), + ).fetchall() + else: + rows = conn.execute( + """ + SELECT lease_id FROM leases + WHERE status = 'active' AND expires_at <= ? AND work_item_id = ? + """, + (now_s, work_item_id), + ).fetchall() + for row in rows: + lid = row["lease_id"] + conn.execute( + "UPDATE leases SET status = 'expired' WHERE lease_id = ?", + (lid,), + ) + conn.execute( + "UPDATE assignments SET status = 'expired' WHERE lease_id = ?", + (lid,), + ) + return len(rows) + + def expire_stale_leases(self) -> int: + with self._tx() as conn: + return self._expire_stale_leases_conn(conn) + + def heartbeat_lease(self, lease_id: str, *, session_id: str) -> dict[str, Any]: + now_s = _ts() + with self._tx() as conn: + row = conn.execute( + "SELECT * FROM leases WHERE lease_id = ?", + (lease_id,), + ).fetchone() + if not row: + raise ControlPlaneError(f"unknown lease_id {lease_id}") + if row["session_id"] != session_id: + raise ForeignLeaseError( + f"lease {lease_id} owned by {row['session_id']}, not {session_id}" + ) + if row["status"] != "active": + raise ControlPlaneError(f"lease {lease_id} status is {row['status']}") + exp = _parse_ts(row["expires_at"]) + if exp and exp <= _utc_now(): + conn.execute( + "UPDATE leases SET status = 'expired' WHERE lease_id = ?", + (lease_id,), + ) + raise ControlPlaneError(f"lease {lease_id} already expired") + # Extend TTL on heartbeat + new_exp = _ts(_utc_now() + timedelta(seconds=DEFAULT_LEASE_TTL_SECONDS)) + conn.execute( + """ + UPDATE leases SET heartbeat_at = ?, expires_at = ? + WHERE lease_id = ? + """, + (now_s, new_exp, lease_id), + ) + conn.execute( + "UPDATE sessions SET last_heartbeat_at = ? WHERE session_id = ?", + (now_s, session_id), + ) + return { + "lease_id": lease_id, + "heartbeat_at": now_s, + "expires_at": new_exp, + "session_id": session_id, + } + + def release_lease(self, lease_id: str, *, session_id: str) -> None: + with self._tx() as conn: + row = conn.execute( + "SELECT * FROM leases WHERE lease_id = ?", + (lease_id,), + ).fetchone() + if not row: + return + if row["session_id"] != session_id: + raise ForeignLeaseError( + f"cannot release lease {lease_id} owned by {row['session_id']}" + ) + conn.execute( + "UPDATE leases SET status = 'released' WHERE lease_id = ?", + (lease_id,), + ) + conn.execute( + "UPDATE assignments SET status = 'released' WHERE lease_id = ?", + (lease_id,), + ) + conn.execute( + """ + INSERT INTO events(work_item_id, event_type, message, created_at) + VALUES (?, 'lease_released', ?, ?) + """, + (row["work_item_id"], f"session {session_id} released {lease_id}", _ts()), + ) + + def require_valid_assignment( + self, + *, + session_id: str, + remote: str, + org: str, + repo: str, + kind: str, + number: int, + action: str, + ) -> dict[str, Any]: + """Gate a mutation: require active assignment+lease for this session/work.""" + kind_norm = (kind or "").strip().lower() + if kind_norm not in WORK_KINDS: + raise InvalidWorkKindError(f"invalid kind '{kind}'") + with self._tx(immediate=False) as conn: + work = conn.execute( + """ + SELECT * FROM work_items + WHERE remote = ? AND org = ? AND repo = ? AND kind = ? AND number = ? + """, + (remote, org, repo, kind_norm, int(number)), + ).fetchone() + if not work: + raise LeaseRequiredError( + f"no work_item for {kind_norm}#{number}; assign first" + ) + self._expire_stale_leases_conn(conn, work_item_id=int(work["work_item_id"])) + asn = conn.execute( + """ + SELECT a.*, l.status AS lease_status, l.expires_at, l.lease_id + FROM assignments a + JOIN leases l ON l.lease_id = a.lease_id + WHERE a.work_item_id = ? + AND a.session_id = ? + AND a.status = 'active' + AND l.status = 'active' + ORDER BY a.created_at DESC + LIMIT 1 + """, + (int(work["work_item_id"]), session_id), + ).fetchone() + if not asn: + raise LeaseRequiredError( + f"session {session_id} has no active assignment/lease for " + f"{kind_norm}#{number}" + ) + exp = _parse_ts(asn["expires_at"]) + if exp and exp <= _utc_now(): + raise LeaseRequiredError(f"lease {asn['lease_id']} expired") + allowed = json.loads(asn["allowed_actions"]) + forbidden = json.loads(asn["forbidden_actions"]) + if action in forbidden: + raise LeaseRequiredError( + f"action '{action}' is forbidden on assignment {asn['assignment_id']}" + ) + if allowed and action not in allowed and action != "heartbeat": + raise LeaseRequiredError( + f"action '{action}' not in allowed_actions {allowed}" + ) + return dict(asn) + + # ── terminal locks ──────────────────────────────────────────────────── + + def set_terminal_lock( + self, + *, + remote: str, + org: str, + repo: str, + terminal_pr: int, + review_id: str | None = None, + decision: str | None = None, + status: str = "active", + cleanup_state: str | None = None, + ) -> dict[str, Any]: + now_s = _ts() + with self._tx() as conn: + conn.execute( + """ + INSERT INTO terminal_locks( + remote, org, repo, terminal_pr, review_id, decision, + status, cleanup_state, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(remote, org, repo, terminal_pr) DO UPDATE SET + review_id = excluded.review_id, + decision = excluded.decision, + status = excluded.status, + cleanup_state = excluded.cleanup_state + """, + ( + remote, + org, + repo, + int(terminal_pr), + review_id, + decision, + status, + cleanup_state, + now_s, + ), + ) + row = conn.execute( + """ + SELECT * FROM terminal_locks + WHERE remote = ? AND org = ? AND repo = ? AND terminal_pr = ? + """, + (remote, org, repo, int(terminal_pr)), + ).fetchone() + return dict(row) + + def get_active_terminal_lock( + self, *, remote: str, org: str, repo: str + ) -> dict[str, Any] | None: + with self._tx(immediate=False) as conn: + row = conn.execute( + """ + SELECT * FROM terminal_locks + WHERE remote = ? AND org = ? AND repo = ? AND status = 'active' + ORDER BY created_at DESC LIMIT 1 + """, + (remote, org, repo), + ).fetchone() + return dict(row) if row else None + + # ── incident_links (provider-neutral; not assignable) ───────────────── + + def upsert_incident_link( + self, + *, + provider: str, + provider_issue_id: str, + gitea_org: str, + gitea_repo: str, + gitea_issue_number: int, + provider_base_url: str | None = None, + provider_org: str | None = None, + provider_project: str | None = None, + provider_short_id: str | None = None, + provider_permalink: str | None = None, + fingerprint: str | None = None, + linked_pr_numbers: Sequence[int] | None = None, + first_seen: str | None = None, + last_seen: str | None = None, + event_count: int | None = None, + status: str = "open", + ) -> dict[str, Any]: + now_s = _ts() + with self._tx() as conn: + conn.execute( + """ + INSERT INTO incident_links( + provider, provider_base_url, provider_org, provider_project, + provider_issue_id, provider_short_id, provider_permalink, + fingerprint, gitea_org, gitea_repo, gitea_issue_number, + linked_pr_numbers, first_seen, last_seen, event_count, + status, last_sync_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(provider, provider_base_url, provider_org, provider_project, provider_issue_id) + DO UPDATE SET + gitea_issue_number = excluded.gitea_issue_number, + gitea_org = excluded.gitea_org, + gitea_repo = excluded.gitea_repo, + provider_permalink = COALESCE(excluded.provider_permalink, incident_links.provider_permalink), + fingerprint = COALESCE(excluded.fingerprint, incident_links.fingerprint), + linked_pr_numbers = COALESCE(excluded.linked_pr_numbers, incident_links.linked_pr_numbers), + last_seen = COALESCE(excluded.last_seen, incident_links.last_seen), + event_count = COALESCE(excluded.event_count, incident_links.event_count), + status = excluded.status, + last_sync_at = excluded.last_sync_at + """, + ( + provider, + provider_base_url, + provider_org, + provider_project, + provider_issue_id, + provider_short_id, + provider_permalink, + fingerprint, + gitea_org, + gitea_repo, + int(gitea_issue_number), + json.dumps(list(linked_pr_numbers or [])), + first_seen, + last_seen, + event_count, + status, + now_s, + ), + ) + row = conn.execute( + """ + SELECT * FROM incident_links + WHERE provider = ? + AND IFNULL(provider_base_url, '') = IFNULL(?, '') + AND IFNULL(provider_org, '') = IFNULL(?, '') + AND IFNULL(provider_project, '') = IFNULL(?, '') + AND provider_issue_id = ? + """, + ( + provider, + provider_base_url, + provider_org, + provider_project, + provider_issue_id, + ), + ).fetchone() + return dict(row) + + def get_incident_link_for_gitea_issue( + self, *, gitea_org: str, gitea_repo: str, gitea_issue_number: int + ) -> dict[str, Any] | None: + with self._tx(immediate=False) as conn: + row = conn.execute( + """ + SELECT * FROM incident_links + WHERE gitea_org = ? AND gitea_repo = ? AND gitea_issue_number = ? + ORDER BY link_id DESC LIMIT 1 + """, + (gitea_org, gitea_repo, int(gitea_issue_number)), + ).fetchone() + return dict(row) if row else None diff --git a/docs/architecture/control-plane-db-substrate.md b/docs/architecture/control-plane-db-substrate.md new file mode 100644 index 0000000..b531799 --- /dev/null +++ b/docs/architecture/control-plane-db-substrate.md @@ -0,0 +1,48 @@ +# Control-plane DB substrate (#613) + +**Status:** Implemented (SQLite single-writer MVP) +**ADR:** [`mcp-allocator-control-plane-observability-adr.md`](mcp-allocator-control-plane-observability-adr.md) +**Module:** `control_plane_db.py` + +## Architecture statement + +> **DB coordinates, Gitea records, Sentry/GlitchTip observe; the bridge is the only path that turns observations into Gitea work.** + +## What this ships + +| Capability | Notes | +|------------|--------| +| Schema | `sessions`, `work_items`, `leases`, `assignments`, `terminal_locks`, `events`, `incident_links` | +| Atomic assign+lease | `ControlPlaneDB.assign_and_lease` — one `BEGIN IMMEDIATE` transaction | +| Mutation gate | `require_valid_assignment` — workers need a live assignment/lease | +| Heartbeat / release / expire | Lease lifecycle helpers | +| Terminal-lock index | Routing signal for #600 (terminal path first) | +| `incident_links` | Provider-neutral link model for #612 — **not** assignable work | + +## Hard rules (enforced in code) + +1. Assignable `work_items.kind` ∈ {`issue`, `pr`} only — **never** raw Sentry/GlitchTip incidents. +2. Two concurrent sessions cannot both receive an active assignment on the same open work item (second gets `wait`). +3. Merged/closed work items return `no_safe_work`. +4. SQLite path is the **single-writer MVP** (`GITEA_CONTROL_PLANE_DB`, default under `~/.cache/gitea-tools/control-plane/`). Multi-session multi-host production requires **Postgres** or a **single allocator daemon** (ADR §6). + +## Dependency chain + +```text +#613 control-plane DB (this) → #600 allocator API → #612 incident bridge +``` + +- **#600** must call this substrate (not file locks / comment-only leases alone) for completion. +- **#612** must write `incident_links` here and create **Gitea issues**; the allocator assigns those issues, not raw incidents. + +## Non-goals (intentionally deferred) + +- Full `gitea_allocate_next_work` routing policy (REQUEST_CHANGES → author, etc.) — **#600** +- Sentry/GlitchTip provider adapters and auto-watchdog — **#612** +- Gitea comment/label mirror writers (may be added by allocator tooling later) + +## Tests + +```bash +python3 -m pytest tests/test_control_plane_db.py -q +``` diff --git a/tests/test_control_plane_db.py b/tests/test_control_plane_db.py new file mode 100644 index 0000000..8091272 --- /dev/null +++ b/tests/test_control_plane_db.py @@ -0,0 +1,377 @@ +"""Tests for control-plane DB substrate (#613).""" + +from __future__ import annotations + +import os +import tempfile +import threading +import unittest +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import timedelta + +from control_plane_db import ( + ControlPlaneDB, + InvalidWorkKindError, + LeaseRequiredError, + WORK_KINDS, + _ts, + _utc_now, +) + + +class ControlPlaneDBTest(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self._tmp.name, "cp.sqlite3") + self.db = ControlPlaneDB(self.db_path) + + def tearDown(self) -> None: + self._tmp.cleanup() + + def test_schema_and_architecture_meta(self) -> None: + import sqlite3 + + conn = sqlite3.connect(self.db_path) + try: + rows = dict(conn.execute("SELECT key, value FROM schema_meta").fetchall()) + finally: + conn.close() + self.assertEqual(rows["schema_version"], "1") + self.assertIn("DB coordinates", rows["architecture"]) + self.assertIn("bridge", rows["architecture"].lower()) + + def test_rejects_raw_incident_as_work_kind(self) -> None: + with self.assertRaises(InvalidWorkKindError): + self.db.upsert_work_item( + remote="prgs", + org="org", + repo="repo", + kind="sentry_incident", + number=1, + ) + with self.assertRaises(InvalidWorkKindError): + self.db.assign_and_lease( + session_id="s1", + role="author", + remote="prgs", + org="org", + repo="repo", + kind="glitchtip_incident", + number=9, + ) + self.assertEqual(WORK_KINDS, frozenset({"issue", "pr"})) + + def test_atomic_assign_and_lease_fields(self) -> None: + self.db.upsert_session(session_id="s-a", role="author", profile="prgs-author") + result = self.db.assign_and_lease( + session_id="s-a", + role="author", + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + kind="issue", + number=613, + expected_head_sha="abc123", + allowed_actions=("implement", "comment"), + forbidden_actions=("approve", "merge"), + ) + self.assertEqual(result.outcome, "assigned") + self.assertIsNotNone(result.assignment_id) + self.assertIsNotNone(result.lease_id) + self.assertEqual(result.role, "author") + self.assertEqual(result.work_kind, "issue") + self.assertEqual(result.work_number, 613) + self.assertEqual(result.expected_head_sha, "abc123") + self.assertIn("implement", result.allowed_actions) + self.assertIn("merge", result.forbidden_actions) + self.assertIsNotNone(result.expires_at) + + def test_second_session_waits_on_foreign_lease(self) -> None: + self.db.upsert_session(session_id="s1", role="author") + self.db.upsert_session(session_id="s2", role="author") + first = self.db.assign_and_lease( + session_id="s1", + role="author", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=100, + expected_head_sha="deadbeef", + ) + self.assertEqual(first.outcome, "assigned") + second = self.db.assign_and_lease( + session_id="s2", + role="author", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=100, + expected_head_sha="deadbeef", + ) + self.assertEqual(second.outcome, "wait") + self.assertEqual(second.owner_session_id, "s1") + + def test_owner_resume_refreshes_lease(self) -> None: + self.db.upsert_session(session_id="s1", role="reviewer") + a = self.db.assign_and_lease( + session_id="s1", + role="reviewer", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=50, + ) + b = self.db.assign_and_lease( + session_id="s1", + role="reviewer", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=50, + ) + self.assertEqual(b.outcome, "assigned") + self.assertEqual(b.lease_id, a.lease_id) + self.assertIn("owner-resume", b.reason) + + def test_require_valid_assignment_gates_mutations(self) -> None: + self.db.upsert_session(session_id="s1", role="author") + self.db.assign_and_lease( + session_id="s1", + role="author", + remote="prgs", + org="o", + repo="r", + kind="issue", + number=7, + allowed_actions=("implement",), + forbidden_actions=("merge",), + ) + proof = self.db.require_valid_assignment( + session_id="s1", + remote="prgs", + org="o", + repo="r", + kind="issue", + number=7, + action="implement", + ) + self.assertEqual(proof["session_id"], "s1") + with self.assertRaises(LeaseRequiredError): + self.db.require_valid_assignment( + session_id="s1", + remote="prgs", + org="o", + repo="r", + kind="issue", + number=7, + action="merge", + ) + with self.assertRaises(LeaseRequiredError): + self.db.require_valid_assignment( + session_id="s-other", + remote="prgs", + org="o", + repo="r", + kind="issue", + number=7, + action="implement", + ) + + def test_expired_lease_allows_reassign(self) -> None: + self.db.upsert_session(session_id="s1", role="author") + self.db.upsert_session(session_id="s2", role="author") + past = _utc_now() - timedelta(hours=1) + # Create lease already expired by using negative TTL edge via direct assign then expire + assigned = self.db.assign_and_lease( + session_id="s1", + role="author", + remote="prgs", + org="o", + repo="r", + kind="issue", + number=3, + lease_ttl_seconds=1, + ) + self.assertEqual(assigned.outcome, "assigned") + # Force expiry in DB + import sqlite3 + + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE leases SET expires_at = ? WHERE lease_id = ?", + (_ts(past), assigned.lease_id), + ) + conn.commit() + finally: + conn.close() + n = self.db.expire_stale_leases() + self.assertGreaterEqual(n, 1) + second = self.db.assign_and_lease( + session_id="s2", + role="author", + remote="prgs", + org="o", + repo="r", + kind="issue", + number=3, + ) + self.assertEqual(second.outcome, "assigned") + self.assertEqual(second.session_id, "s2") + + def test_merged_work_never_assigned(self) -> None: + self.db.upsert_session(session_id="s1", role="merger") + self.db.upsert_work_item( + remote="prgs", + org="o", + repo="r", + kind="pr", + number=99, + state="merged", + ) + result = self.db.assign_and_lease( + session_id="s1", + role="merger", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=99, + ) + self.assertEqual(result.outcome, "no_safe_work") + + def test_four_concurrent_sessions_unique_assignments(self) -> None: + """Four concurrent assigners on four different issues — all succeed uniquely. + + Also two concurrent assigners on the *same* issue: at most one assigned. + """ + for i in range(4): + self.db.upsert_session(session_id=f"sess-{i}", role="author") + + def claim_unique(i: int): + return self.db.assign_and_lease( + session_id=f"sess-{i}", + role="author", + remote="prgs", + org="o", + repo="r", + kind="issue", + number=1000 + i, + ) + + with ThreadPoolExecutor(max_workers=4) as pool: + results = [f.result() for f in as_completed([pool.submit(claim_unique, i) for i in range(4)])] + self.assertEqual({r.outcome for r in results}, {"assigned"}) + numbers = sorted(r.work_number for r in results) + self.assertEqual(numbers, [1000, 1001, 1002, 1003]) + + # Contention on one item + self.db.upsert_session(session_id="c1", role="author") + self.db.upsert_session(session_id="c2", role="author") + self.db.upsert_session(session_id="c3", role="author") + self.db.upsert_session(session_id="c4", role="author") + barrier = threading.Barrier(4) + outcomes: list[str] = [] + lock = threading.Lock() + + def contend(sid: str) -> None: + barrier.wait() + res = self.db.assign_and_lease( + session_id=sid, + role="author", + remote="prgs", + org="o", + repo="r", + kind="issue", + number=7777, + ) + with lock: + outcomes.append(res.outcome) + + threads = [threading.Thread(target=contend, args=(f"c{i}",)) for i in range(1, 5)] + for t in threads: + t.start() + for t in threads: + t.join() + self.assertEqual(outcomes.count("assigned"), 1) + self.assertEqual(outcomes.count("wait"), 3) + + def test_terminal_lock_index(self) -> None: + self.db.set_terminal_lock( + remote="prgs", + org="o", + repo="r", + terminal_pr=332, + review_id="rev-1", + decision="approve", + ) + row = self.db.get_active_terminal_lock(remote="prgs", org="o", repo="r") + self.assertIsNotNone(row) + assert row is not None + self.assertEqual(row["terminal_pr"], 332) + self.assertEqual(row["status"], "active") + + def test_incident_links_not_work_items(self) -> None: + link = self.db.upsert_incident_link( + provider="sentry", + provider_base_url="https://sentry.prgs.cc", + provider_org="prgs", + provider_project="gitea-tools-mcp", + provider_issue_id="12345", + gitea_org="Scaled-Tech-Consulting", + gitea_repo="Gitea-Tools", + gitea_issue_number=9001, + fingerprint="fp-1", + ) + self.assertEqual(link["gitea_issue_number"], 9001) + found = self.db.get_incident_link_for_gitea_issue( + gitea_org="Scaled-Tech-Consulting", + gitea_repo="Gitea-Tools", + gitea_issue_number=9001, + ) + self.assertIsNotNone(found) + # Linking does not create a work_item of incident kind + with self.assertRaises(InvalidWorkKindError): + self.db.upsert_work_item( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + kind="sentry", + number=12345, + ) + + def test_heartbeat_and_release(self) -> None: + self.db.upsert_session(session_id="s1", role="author") + a = self.db.assign_and_lease( + session_id="s1", + role="author", + remote="prgs", + org="o", + repo="r", + kind="issue", + number=1, + ) + hb = self.db.heartbeat_lease(a.lease_id, session_id="s1") + self.assertEqual(hb["lease_id"], a.lease_id) + self.db.release_lease(a.lease_id, session_id="s1") + # After release another session can claim + self.db.upsert_session(session_id="s2", role="author") + b = self.db.assign_and_lease( + session_id="s2", + role="author", + remote="prgs", + org="o", + repo="r", + kind="issue", + number=1, + ) + self.assertEqual(b.outcome, "assigned") + self.assertEqual(b.session_id, "s2") + + +if __name__ == "__main__": + unittest.main() From 783ea88a01da721138429ede8dc8a6c0c186206a Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 9 Jul 2026 22:41:25 -0400 Subject: [PATCH 2/4] fix: fail closed on stale/terminal assignments; canonicalize incident_links Address REQUEST_CHANGES on PR #619: - require_valid_assignment rejects terminal work states and head drift - incident_links scope keys normalize NULL/blank to '' for UNIQUE - remove trailing whitespace in control-plane-db-substrate.md --- control_plane_db.py | 98 +++++++++-- .../control-plane-db-substrate.md | 16 +- tests/test_control_plane_db.py | 155 +++++++++++++++++- 3 files changed, 244 insertions(+), 25 deletions(-) diff --git a/control_plane_db.py b/control_plane_db.py index adaa542..1e70a06 100644 --- a/control_plane_db.py +++ b/control_plane_db.py @@ -29,11 +29,14 @@ from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import Any, Iterator, Sequence -SCHEMA_VERSION = 1 +SCHEMA_VERSION = 2 # Assignable work kinds only — raw monitoring incidents are never work items. WORK_KINDS = frozenset({"issue", "pr"}) +# Work item states that permanently revoke assignment/lease authority. +TERMINAL_WORK_STATES = frozenset({"merged", "closed"}) + DEFAULT_LEASE_TTL_SECONDS = 4 * 3600 # Environment: path for SQLite MVP (single-writer). @@ -120,12 +123,13 @@ CREATE TABLE IF NOT EXISTS events ( ); -- Provider-neutral incident ↔ Gitea link model (ADR §8–9). Not assignable work. +-- Optional scope fields are stored as '' (never NULL) so UNIQUE is NULL-safe. CREATE TABLE IF NOT EXISTS incident_links ( link_id INTEGER PRIMARY KEY AUTOINCREMENT, provider TEXT NOT NULL, - provider_base_url TEXT, - provider_org TEXT, - provider_project TEXT, + provider_base_url TEXT NOT NULL DEFAULT '', + provider_org TEXT NOT NULL DEFAULT '', + provider_project TEXT NOT NULL DEFAULT '', provider_issue_id TEXT NOT NULL, provider_short_id TEXT, provider_permalink TEXT, @@ -172,6 +176,17 @@ def _parse_ts(value: str | None) -> datetime | None: return None +def _norm_scope(value: str | None) -> str: + """Normalize optional incident-link scope keys for NULL-safe uniqueness. + + Empty / whitespace / None all become '' so SQLite UNIQUE treats them as one + canonical key component (SQLite treats multiple NULLs as distinct). + """ + if value is None: + return "" + return str(value).strip() + + def default_db_path() -> str: raw = (os.environ.get(DB_PATH_ENV) or DEFAULT_DB_PATH).strip() return raw or DEFAULT_DB_PATH @@ -285,6 +300,7 @@ class ControlPlaneDB: conn = self._connect() try: conn.executescript(_SCHEMA_SQL) + self._migrate_incident_links_null_scope(conn) conn.execute( "INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)", ("schema_version", str(SCHEMA_VERSION)), @@ -301,6 +317,35 @@ class ControlPlaneDB: finally: conn.close() + def _migrate_incident_links_null_scope(self, conn: sqlite3.Connection) -> None: + """Coerce NULL scope columns to '' and collapse any duplicate keys (#619 RC).""" + cols = { + row[1] + for row in conn.execute("PRAGMA table_info(incident_links)").fetchall() + } + if not cols: + return + for col in ("provider_base_url", "provider_org", "provider_project"): + if col in cols: + conn.execute( + f"UPDATE incident_links SET {col} = '' WHERE {col} IS NULL" + ) + # Drop duplicates keeping lowest link_id (canonical row). + conn.execute( + """ + DELETE FROM incident_links + WHERE link_id NOT IN ( + SELECT MIN(link_id) + FROM incident_links + GROUP BY provider, + IFNULL(provider_base_url, ''), + IFNULL(provider_org, ''), + IFNULL(provider_project, ''), + provider_issue_id + ) + """ + ) + # ── sessions ────────────────────────────────────────────────────────── def upsert_session( @@ -730,7 +775,12 @@ class ControlPlaneDB: number: int, action: str, ) -> dict[str, Any]: - """Gate a mutation: require active assignment+lease for this session/work.""" + """Gate a mutation: require active assignment+lease for this session/work. + + Fail-closed when the work item is terminal (merged/closed) or when the + assignment's expected_head_sha no longer matches the work item head + (stale-head after assignment time). + """ kind_norm = (kind or "").strip().lower() if kind_norm not in WORK_KINDS: raise InvalidWorkKindError(f"invalid kind '{kind}'") @@ -746,6 +796,12 @@ class ControlPlaneDB: raise LeaseRequiredError( f"no work_item for {kind_norm}#{number}; assign first" ) + work_state = (work["state"] or "").strip().lower() + if work_state in TERMINAL_WORK_STATES: + raise LeaseRequiredError( + f"work item {kind_norm}#{number} is terminal state " + f"'{work_state}'; assignment no longer authorizes mutations" + ) self._expire_stale_leases_conn(conn, work_item_id=int(work["work_item_id"])) asn = conn.execute( """ @@ -769,6 +825,14 @@ class ControlPlaneDB: exp = _parse_ts(asn["expires_at"]) if exp and exp <= _utc_now(): raise LeaseRequiredError(f"lease {asn['lease_id']} expired") + # Stale-head: pinned expected_head_sha must still match live work item. + assigned_head = (asn["expected_head_sha"] or "").strip() + current_head = (work["current_head_sha"] or "").strip() + if assigned_head and assigned_head != current_head: + raise LeaseRequiredError( + f"assignment {asn['assignment_id']} stale head: expected " + f"{assigned_head!r} but work item head is {current_head!r}" + ) allowed = json.loads(asn["allowed_actions"]) forbidden = json.loads(asn["forbidden_actions"]) if action in forbidden: @@ -867,6 +931,10 @@ class ControlPlaneDB: status: str = "open", ) -> dict[str, Any]: now_s = _ts() + # Canonical key: never store NULL in UNIQUE scope columns. + base_url = _norm_scope(provider_base_url) + p_org = _norm_scope(provider_org) + p_project = _norm_scope(provider_project) with self._tx() as conn: conn.execute( """ @@ -892,9 +960,9 @@ class ControlPlaneDB: """, ( provider, - provider_base_url, - provider_org, - provider_project, + base_url, + p_org, + p_project, provider_issue_id, provider_short_id, provider_permalink, @@ -914,18 +982,12 @@ class ControlPlaneDB: """ SELECT * FROM incident_links WHERE provider = ? - AND IFNULL(provider_base_url, '') = IFNULL(?, '') - AND IFNULL(provider_org, '') = IFNULL(?, '') - AND IFNULL(provider_project, '') = IFNULL(?, '') + AND provider_base_url = ? + AND provider_org = ? + AND provider_project = ? AND provider_issue_id = ? """, - ( - provider, - provider_base_url, - provider_org, - provider_project, - provider_issue_id, - ), + (provider, base_url, p_org, p_project, provider_issue_id), ).fetchone() return dict(row) diff --git a/docs/architecture/control-plane-db-substrate.md b/docs/architecture/control-plane-db-substrate.md index b531799..5bb589a 100644 --- a/docs/architecture/control-plane-db-substrate.md +++ b/docs/architecture/control-plane-db-substrate.md @@ -1,7 +1,9 @@ # Control-plane DB substrate (#613) -**Status:** Implemented (SQLite single-writer MVP) -**ADR:** [`mcp-allocator-control-plane-observability-adr.md`](mcp-allocator-control-plane-observability-adr.md) +**Status:** Implemented (SQLite single-writer MVP) + +**ADR:** [`mcp-allocator-control-plane-observability-adr.md`](mcp-allocator-control-plane-observability-adr.md) + **Module:** `control_plane_db.py` ## Architecture statement @@ -14,17 +16,19 @@ |------------|--------| | Schema | `sessions`, `work_items`, `leases`, `assignments`, `terminal_locks`, `events`, `incident_links` | | Atomic assign+lease | `ControlPlaneDB.assign_and_lease` — one `BEGIN IMMEDIATE` transaction | -| Mutation gate | `require_valid_assignment` — workers need a live assignment/lease | +| Mutation gate | `require_valid_assignment` — live lease + allowed action + non-terminal work + non-stale head | | Heartbeat / release / expire | Lease lifecycle helpers | | Terminal-lock index | Routing signal for #600 (terminal path first) | -| `incident_links` | Provider-neutral link model for #612 — **not** assignable work | +| `incident_links` | Provider-neutral link model for #612 — **not** assignable work; scope keys NULL-safe | ## Hard rules (enforced in code) 1. Assignable `work_items.kind` ∈ {`issue`, `pr`} only — **never** raw Sentry/GlitchTip incidents. 2. Two concurrent sessions cannot both receive an active assignment on the same open work item (second gets `wait`). -3. Merged/closed work items return `no_safe_work`. -4. SQLite path is the **single-writer MVP** (`GITEA_CONTROL_PLANE_DB`, default under `~/.cache/gitea-tools/control-plane/`). Multi-session multi-host production requires **Postgres** or a **single allocator daemon** (ADR §6). +3. Merged/closed work items return `no_safe_work` at assign time, and `require_valid_assignment` fails closed if the work item becomes terminal later. +4. Assignments pin `expected_head_sha`; mutations fail closed if the work item head drifts. +5. SQLite path is the **single-writer MVP** (`GITEA_CONTROL_PLANE_DB`, default under `~/.cache/gitea-tools/control-plane/`). Multi-session multi-host production requires **Postgres** or a **single allocator daemon** (ADR §6). +6. `incident_links` optional scope fields are stored as empty strings (never NULL) so UNIQUE is canonical across minimal upserts. ## Dependency chain diff --git a/tests/test_control_plane_db.py b/tests/test_control_plane_db.py index 8091272..8535335 100644 --- a/tests/test_control_plane_db.py +++ b/tests/test_control_plane_db.py @@ -36,7 +36,7 @@ class ControlPlaneDBTest(unittest.TestCase): rows = dict(conn.execute("SELECT key, value FROM schema_meta").fetchall()) finally: conn.close() - self.assertEqual(rows["schema_version"], "1") + self.assertEqual(rows["schema_version"], "2") self.assertIn("DB coordinates", rows["architecture"]) self.assertIn("bridge", rows["architecture"].lower()) @@ -372,6 +372,159 @@ class ControlPlaneDBTest(unittest.TestCase): self.assertEqual(b.outcome, "assigned") self.assertEqual(b.session_id, "s2") + def test_require_valid_assignment_rejects_stale_head(self) -> None: + """Assignment must not authorize mutations after work-item head drifts.""" + self.db.upsert_session(session_id="s1", role="author") + self.db.assign_and_lease( + session_id="s1", + role="author", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=42, + expected_head_sha="head-v1", + allowed_actions=("implement",), + ) + # Head drifts after assignment + self.db.upsert_work_item( + remote="prgs", + org="o", + repo="r", + kind="pr", + number=42, + current_head_sha="head-v2", + ) + with self.assertRaises(LeaseRequiredError) as ctx: + self.db.require_valid_assignment( + session_id="s1", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=42, + action="implement", + ) + self.assertIn("stale head", str(ctx.exception).lower()) + + def test_require_valid_assignment_rejects_terminal_state(self) -> None: + """Assignment must not authorize mutations after work item is merged/closed.""" + self.db.upsert_session(session_id="s1", role="author") + self.db.assign_and_lease( + session_id="s1", + role="author", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=55, + expected_head_sha="abc", + allowed_actions=("implement",), + ) + self.db.upsert_work_item( + remote="prgs", + org="o", + repo="r", + kind="pr", + number=55, + state="merged", + current_head_sha="abc", + ) + with self.assertRaises(LeaseRequiredError) as ctx: + self.db.require_valid_assignment( + session_id="s1", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=55, + action="implement", + ) + self.assertIn("terminal", str(ctx.exception).lower()) + + self.db.upsert_work_item( + remote="prgs", + org="o", + repo="r", + kind="issue", + number=56, + state="open", + ) + self.db.assign_and_lease( + session_id="s1", + role="author", + remote="prgs", + org="o", + repo="r", + kind="issue", + number=56, + allowed_actions=("implement",), + ) + self.db.upsert_work_item( + remote="prgs", + org="o", + repo="r", + kind="issue", + number=56, + state="closed", + ) + with self.assertRaises(LeaseRequiredError): + self.db.require_valid_assignment( + session_id="s1", + remote="prgs", + org="o", + repo="r", + kind="issue", + number=56, + action="implement", + ) + + def test_incident_links_minimal_upsert_is_canonical(self) -> None: + """Repeated minimal upserts must update one row (NULL-safe uniqueness).""" + a = self.db.upsert_incident_link( + provider="sentry", + provider_issue_id="inc-1", + gitea_org="org", + gitea_repo="repo", + gitea_issue_number=1, + ) + b = self.db.upsert_incident_link( + provider="sentry", + provider_issue_id="inc-1", + gitea_org="org", + gitea_repo="repo", + gitea_issue_number=2, + # omit optional scope fields again + ) + c = self.db.upsert_incident_link( + provider="sentry", + provider_issue_id="inc-1", + gitea_org="org", + gitea_repo="repo", + gitea_issue_number=3, + provider_base_url=None, + provider_org="", + provider_project=" ", + ) + self.assertEqual(a["link_id"], b["link_id"]) + self.assertEqual(b["link_id"], c["link_id"]) + self.assertEqual(c["gitea_issue_number"], 3) + self.assertEqual(c["provider_base_url"], "") + self.assertEqual(c["provider_org"], "") + self.assertEqual(c["provider_project"], "") + + import sqlite3 + + conn = sqlite3.connect(self.db_path) + try: + n = conn.execute( + "SELECT COUNT(*) FROM incident_links WHERE provider_issue_id = ?", + ("inc-1",), + ).fetchone()[0] + finally: + conn.close() + self.assertEqual(n, 1) + if __name__ == "__main__": unittest.main() From 16cd871fbd214529248e33c6dfc2d53884f67867 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 9 Jul 2026 22:51:27 -0400 Subject: [PATCH 3/4] fix: safe incident_links migration order; require PR head pin Address remaining PR #619 blockers on head 783ea88: - Deduplicate legacy NULL-scope incident_links before normalizing to '' - Fail closed when duplicate rows disagree on Gitea target - Require non-empty expected_head_sha for PR assign and mutation --- control_plane_db.py | 87 +++++++++++--- tests/test_control_plane_db.py | 207 +++++++++++++++++++++++++++++++++ 2 files changed, 277 insertions(+), 17 deletions(-) diff --git a/control_plane_db.py b/control_plane_db.py index 1e70a06..cbc09fa 100644 --- a/control_plane_db.py +++ b/control_plane_db.py @@ -318,33 +318,75 @@ class ControlPlaneDB: conn.close() def _migrate_incident_links_null_scope(self, conn: sqlite3.Connection) -> None: - """Coerce NULL scope columns to '' and collapse any duplicate keys (#619 RC).""" + """Collapse legacy NULL-scope duplicates, then normalize to '' (#619 RC2). + + Order is mandatory: SQLite UNIQUE treats multiple NULLs as distinct, so + normalizing NULL→'' first can hit the UNIQUE constraint and abort + migration. Deduplicate under the *normalized* key first, fail closed if + Gitea targets conflict within a group, then coerce NULLs to ''. + """ cols = { row[1] for row in conn.execute("PRAGMA table_info(incident_links)").fetchall() } if not cols: return + + rows = conn.execute( + """ + SELECT link_id, provider, + IFNULL(provider_base_url, '') AS base_url, + IFNULL(provider_org, '') AS p_org, + IFNULL(provider_project, '') AS p_project, + provider_issue_id, + gitea_org, gitea_repo, gitea_issue_number + FROM incident_links + ORDER BY link_id ASC + """ + ).fetchall() + + groups: dict[tuple[str, str, str, str, str], list[sqlite3.Row]] = {} + for row in rows: + key = ( + str(row["provider"]), + str(row["base_url"]), + str(row["p_org"]), + str(row["p_project"]), + str(row["provider_issue_id"]), + ) + groups.setdefault(key, []).append(row) + + to_delete: list[int] = [] + for key, members in groups.items(): + if len(members) < 2: + continue + # Canonical identity for observation links: Gitea issue target. + targets = { + ( + str(m["gitea_org"] or ""), + str(m["gitea_repo"] or ""), + int(m["gitea_issue_number"]), + ) + for m in members + } + if len(targets) > 1: + raise ControlPlaneError( + "incident_links migration fail closed: conflicting Gitea " + f"targets for provider key {key!r}: {sorted(targets)}" + ) + keep_id = int(members[0]["link_id"]) # lowest link_id (ORDER BY) + for m in members[1:]: + to_delete.append(int(m["link_id"])) + + for link_id in to_delete: + conn.execute("DELETE FROM incident_links WHERE link_id = ?", (link_id,)) + + # Safe only after dedupe: normalize NULL scope fields to empty string. for col in ("provider_base_url", "provider_org", "provider_project"): if col in cols: conn.execute( f"UPDATE incident_links SET {col} = '' WHERE {col} IS NULL" ) - # Drop duplicates keeping lowest link_id (canonical row). - conn.execute( - """ - DELETE FROM incident_links - WHERE link_id NOT IN ( - SELECT MIN(link_id) - FROM incident_links - GROUP BY provider, - IFNULL(provider_base_url, ''), - IFNULL(provider_org, ''), - IFNULL(provider_project, ''), - provider_issue_id - ) - """ - ) # ── sessions ────────────────────────────────────────────────────────── @@ -481,6 +523,11 @@ class ControlPlaneDB: raise InvalidWorkKindError( f"cannot assign kind '{kind}'; only {sorted(WORK_KINDS)}" ) + head_pin = (expected_head_sha or "").strip() + if kind_norm == "pr" and not head_pin: + raise LeaseRequiredError( + f"pr#{int(number)} assignment requires non-empty expected_head_sha pin" + ) allowed = tuple(allowed_actions or ("implement", "comment")) forbidden = tuple( @@ -825,9 +872,15 @@ class ControlPlaneDB: exp = _parse_ts(asn["expires_at"]) if exp and exp <= _utc_now(): raise LeaseRequiredError(f"lease {asn['lease_id']} expired") - # Stale-head: pinned expected_head_sha must still match live work item. + # Head pin: PRs require a non-empty expected_head_sha (fail closed). assigned_head = (asn["expected_head_sha"] or "").strip() current_head = (work["current_head_sha"] or "").strip() + if kind_norm == "pr" and not assigned_head: + raise LeaseRequiredError( + f"assignment {asn['assignment_id']} for pr#{number} has no " + f"expected_head_sha pin; PR mutations require a head pin" + ) + # Stale-head: pinned expected_head_sha must still match live work item. if assigned_head and assigned_head != current_head: raise LeaseRequiredError( f"assignment {asn['assignment_id']} stale head: expected " diff --git a/tests/test_control_plane_db.py b/tests/test_control_plane_db.py index 8535335..da8f0ed 100644 --- a/tests/test_control_plane_db.py +++ b/tests/test_control_plane_db.py @@ -123,6 +123,7 @@ class ControlPlaneDBTest(unittest.TestCase): repo="r", kind="pr", number=50, + expected_head_sha="head-50", ) b = self.db.assign_and_lease( session_id="s1", @@ -132,6 +133,7 @@ class ControlPlaneDBTest(unittest.TestCase): repo="r", kind="pr", number=50, + expected_head_sha="head-50", ) self.assertEqual(b.outcome, "assigned") self.assertEqual(b.lease_id, a.lease_id) @@ -241,6 +243,7 @@ class ControlPlaneDBTest(unittest.TestCase): repo="r", kind="pr", number=99, + expected_head_sha="merged-head", ) self.assertEqual(result.outcome, "no_safe_work") @@ -479,6 +482,210 @@ class ControlPlaneDBTest(unittest.TestCase): action="implement", ) + def test_pr_assignment_requires_expected_head_sha(self) -> None: + """PR assign and mutation must fail closed without a head pin.""" + self.db.upsert_session(session_id="s1", role="author") + with self.assertRaises(LeaseRequiredError) as ctx: + self.db.assign_and_lease( + session_id="s1", + role="author", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=88, + expected_head_sha=None, + allowed_actions=("implement",), + ) + self.assertIn("expected_head_sha", str(ctx.exception)) + + # Legacy path: force an unpinned PR assignment into the DB, then + # populate head and prove mutation is still rejected. + import sqlite3 + + self.db.upsert_work_item( + remote="prgs", + org="o", + repo="r", + kind="pr", + number=89, + current_head_sha=None, + ) + conn = sqlite3.connect(self.db_path) + try: + wid = conn.execute( + "SELECT work_item_id FROM work_items WHERE kind='pr' AND number=89" + ).fetchone()[0] + conn.execute( + """ + INSERT INTO sessions(session_id, role, started_at, last_heartbeat_at, status) + VALUES ('legacy', 'author', '2020-01-01T00:00:00Z', '2020-01-01T00:00:00Z', 'active') + """ + ) + conn.execute( + """ + INSERT INTO leases( + lease_id, work_item_id, session_id, role, phase, + expires_at, heartbeat_at, status + ) VALUES ( + 'lease-legacy', ?, 'legacy', 'author', 'claimed', + '2099-01-01T00:00:00Z', '2020-01-01T00:00:00Z', 'active' + ) + """, + (wid,), + ) + conn.execute( + """ + INSERT INTO assignments( + assignment_id, work_item_id, session_id, lease_id, + allowed_actions, forbidden_actions, expected_head_sha, + role, status, created_at + ) VALUES ( + 'asn-legacy', ?, 'legacy', 'lease-legacy', + '["implement"]', '["merge"]', NULL, + 'author', 'active', '2020-01-01T00:00:00Z' + ) + """, + (wid,), + ) + conn.commit() + finally: + conn.close() + self.db.upsert_work_item( + remote="prgs", + org="o", + repo="r", + kind="pr", + number=89, + current_head_sha="populated-head", + ) + with self.assertRaises(LeaseRequiredError) as ctx2: + self.db.require_valid_assignment( + session_id="legacy", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=89, + action="implement", + ) + self.assertIn("expected_head_sha pin", str(ctx2.exception)) + + def test_migrate_duplicate_null_scope_incident_links(self) -> None: + """Legacy NULL-scope duplicates must migrate without UNIQUE crash.""" + import sqlite3 + + from control_plane_db import ControlPlaneDB, ControlPlaneError + + # Build a v1-like table with nullable scope columns and insert dups + # that collapse under normalization, then open ControlPlaneDB on it. + path = os.path.join(self._tmp.name, "legacy_dups.sqlite3") + conn = sqlite3.connect(path) + try: + conn.executescript( + """ + CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL); + CREATE TABLE incident_links ( + link_id INTEGER PRIMARY KEY AUTOINCREMENT, + provider TEXT NOT NULL, + provider_base_url TEXT, + provider_org TEXT, + provider_project TEXT, + provider_issue_id TEXT NOT NULL, + provider_short_id TEXT, + provider_permalink TEXT, + fingerprint TEXT, + gitea_org TEXT NOT NULL, + gitea_repo TEXT NOT NULL, + gitea_issue_number INTEGER NOT NULL, + linked_pr_numbers TEXT, + first_seen TEXT, + last_seen TEXT, + event_count INTEGER, + status TEXT NOT NULL DEFAULT 'open', + release_resolved_at TEXT, + last_sync_at TEXT, + UNIQUE ( + provider, provider_base_url, provider_org, + provider_project, provider_issue_id + ) + ); + INSERT INTO incident_links( + provider, provider_base_url, provider_org, provider_project, + provider_issue_id, gitea_org, gitea_repo, gitea_issue_number, status + ) VALUES + ('sentry', NULL, NULL, NULL, 'dup-1', 'org', 'repo', 10, 'open'), + ('sentry', NULL, NULL, NULL, 'dup-1', 'org', 'repo', 10, 'open'); + """ + ) + # SQLite allows two NULL-scope rows with same provider/issue under UNIQUE. + n = conn.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0] + self.assertEqual(n, 2) + conn.commit() + finally: + conn.close() + + db = ControlPlaneDB(path) + conn2 = sqlite3.connect(path) + try: + n2 = conn2.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0] + rows = conn2.execute( + "SELECT provider_base_url, provider_org, provider_project, gitea_issue_number " + "FROM incident_links" + ).fetchall() + finally: + conn2.close() + self.assertEqual(n2, 1) + self.assertEqual(rows[0][0], "") + self.assertEqual(rows[0][1], "") + self.assertEqual(rows[0][2], "") + self.assertEqual(rows[0][3], 10) + # Touch to silence unused import in type checkers if needed + self.assertTrue(issubclass(ControlPlaneError, Exception)) + del db + + def test_migrate_conflicting_duplicate_incident_links_fails_closed(self) -> None: + """Conflicting Gitea targets for the same provider key must fail closed.""" + import sqlite3 + from control_plane_db import ControlPlaneDB, ControlPlaneError + + path = os.path.join(self._tmp.name, "legacy_conflict.sqlite3") + conn = sqlite3.connect(path) + try: + conn.executescript( + """ + CREATE TABLE incident_links ( + link_id INTEGER PRIMARY KEY AUTOINCREMENT, + provider TEXT NOT NULL, + provider_base_url TEXT, + provider_org TEXT, + provider_project TEXT, + provider_issue_id TEXT NOT NULL, + gitea_org TEXT NOT NULL, + gitea_repo TEXT NOT NULL, + gitea_issue_number INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'open', + UNIQUE ( + provider, provider_base_url, provider_org, + provider_project, provider_issue_id + ) + ); + INSERT INTO incident_links( + provider, provider_base_url, provider_org, provider_project, + provider_issue_id, gitea_org, gitea_repo, gitea_issue_number + ) VALUES + ('sentry', NULL, NULL, NULL, 'dup-c', 'org', 'repo', 1), + ('sentry', NULL, NULL, NULL, 'dup-c', 'org', 'repo', 2); + """ + ) + conn.commit() + finally: + conn.close() + + with self.assertRaises(ControlPlaneError) as ctx: + ControlPlaneDB(path) + self.assertIn("conflicting", str(ctx.exception).lower()) + def test_incident_links_minimal_upsert_is_canonical(self) -> None: """Repeated minimal upserts must update one row (NULL-safe uniqueness).""" a = self.db.upsert_incident_link( From 2429d9d2e826e519eb8eb4ade45987c00838aefb Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 9 Jul 2026 23:04:17 -0400 Subject: [PATCH 4/4] fix: fail closed on conflicting incident_links observation metadata Legacy NULL-scope dedupe only compared Gitea targets, so duplicate rows with the same provider key and issue but different fingerprint/status/ event_count/etc. collapsed to the lowest link_id and silently dropped observation data. Migration now compares all meaningful observation fields and refuses to discard conflicts (#619 RC3 / #613). --- control_plane_db.py | 118 ++++++++++++++++-- .../control-plane-db-substrate.md | 1 + tests/test_control_plane_db.py | 87 +++++++++++++ 3 files changed, 193 insertions(+), 13 deletions(-) diff --git a/control_plane_db.py b/control_plane_db.py index cbc09fa..de723a3 100644 --- a/control_plane_db.py +++ b/control_plane_db.py @@ -317,13 +317,68 @@ class ControlPlaneDB: finally: conn.close() + # Observation fields that must agree before collapsing legacy duplicates. + # Deleting a row would drop any of these; silent loss is forbidden (#619 RC3). + _INCIDENT_LINK_OBS_COMPARE_FIELDS: tuple[str, ...] = ( + "provider_short_id", + "provider_permalink", + "fingerprint", + "gitea_org", + "gitea_repo", + "gitea_issue_number", + "linked_pr_numbers", + "first_seen", + "last_seen", + "event_count", + "status", + "release_resolved_at", + "last_sync_at", + ) + + @staticmethod + def _norm_incident_obs_value(field: str, value: Any) -> Any: + """Normalize optional observation values for equality during migration. + + NULL and empty string are treated as equivalent for optional text + fields (legacy rows often omit them). Integer fields keep NULL + distinct from zero so event_count=0 vs NULL is not collapsed away. + """ + if field in ("gitea_issue_number", "event_count"): + if value is None or value == "": + return None + return int(value) + if value is None: + return "" + return str(value) + + def _incident_link_observation_identity( + self, row: sqlite3.Row, cols: set[str] + ) -> tuple[Any, ...]: + """Stable comparable identity of all meaningful observation metadata.""" + parts: list[Any] = [] + for field in self._INCIDENT_LINK_OBS_COMPARE_FIELDS: + if field not in cols: + continue + # Row keys match column names; missing keys treated as absent. + try: + raw = row[field] + except (IndexError, KeyError): + raw = None + parts.append((field, self._norm_incident_obs_value(field, raw))) + return tuple(parts) + def _migrate_incident_links_null_scope(self, conn: sqlite3.Connection) -> None: - """Collapse legacy NULL-scope duplicates, then normalize to '' (#619 RC2). + """Collapse legacy NULL-scope duplicates, then normalize to '' (#619 RC). Order is mandatory: SQLite UNIQUE treats multiple NULLs as distinct, so normalizing NULL→'' first can hit the UNIQUE constraint and abort migration. Deduplicate under the *normalized* key first, fail closed if - Gitea targets conflict within a group, then coerce NULLs to ''. + Gitea targets **or** any other meaningful observation metadata conflict + within a group (no silent data loss), then coerce NULLs to ''. + + Policy: **fail closed** — never invent a merge of fingerprint/status/ + event_count/etc. Only identical (after optional-text NULL≈'') rows may + collapse to the lowest ``link_id``. """ cols = { row[1] @@ -332,17 +387,41 @@ class ControlPlaneDB: if not cols: return + # Build SELECT from present columns so partial legacy schemas still migrate. + # Scope fields are normalized in the SELECT aliases used for grouping. + required = ("link_id", "provider", "provider_issue_id", "gitea_org", "gitea_repo", "gitea_issue_number") + if not all(c in cols for c in required): + return + + select_parts = [ + "link_id", + "provider", + "IFNULL(provider_base_url, '') AS base_url" + if "provider_base_url" in cols + else "'' AS base_url", + "IFNULL(provider_org, '') AS p_org" + if "provider_org" in cols + else "'' AS p_org", + "IFNULL(provider_project, '') AS p_project" + if "provider_project" in cols + else "'' AS p_project", + "provider_issue_id", + "gitea_org", + "gitea_repo", + "gitea_issue_number", + ] + for field in self._INCIDENT_LINK_OBS_COMPARE_FIELDS: + if field in ( + "gitea_org", + "gitea_repo", + "gitea_issue_number", + ): + continue # already selected + if field in cols: + select_parts.append(field) + rows = conn.execute( - """ - SELECT link_id, provider, - IFNULL(provider_base_url, '') AS base_url, - IFNULL(provider_org, '') AS p_org, - IFNULL(provider_project, '') AS p_project, - provider_issue_id, - gitea_org, gitea_repo, gitea_issue_number - FROM incident_links - ORDER BY link_id ASC - """ + f"SELECT {', '.join(select_parts)} FROM incident_links ORDER BY link_id ASC" ).fetchall() groups: dict[tuple[str, str, str, str, str], list[sqlite3.Row]] = {} @@ -374,7 +453,20 @@ class ControlPlaneDB: "incident_links migration fail closed: conflicting Gitea " f"targets for provider key {key!r}: {sorted(targets)}" ) - keep_id = int(members[0]["link_id"]) # lowest link_id (ORDER BY) + # Same Gitea target is not enough: fingerprint/status/event_count/ + # permalinks/timestamps/etc. must also agree or we lose data. + obs_identities = { + self._incident_link_observation_identity(m, cols) for m in members + } + if len(obs_identities) > 1: + raise ControlPlaneError( + "incident_links migration fail closed: conflicting " + "observation metadata for provider key " + f"{key!r} (same Gitea target but differing fingerprint/" + "status/event_count/permalink/timestamps/or related fields); " + "refusing silent discard of duplicate rows" + ) + # lowest link_id kept (ORDER BY link_id ASC) for m in members[1:]: to_delete.append(int(m["link_id"])) diff --git a/docs/architecture/control-plane-db-substrate.md b/docs/architecture/control-plane-db-substrate.md index 5bb589a..3eefeb8 100644 --- a/docs/architecture/control-plane-db-substrate.md +++ b/docs/architecture/control-plane-db-substrate.md @@ -29,6 +29,7 @@ 4. Assignments pin `expected_head_sha`; mutations fail closed if the work item head drifts. 5. SQLite path is the **single-writer MVP** (`GITEA_CONTROL_PLANE_DB`, default under `~/.cache/gitea-tools/control-plane/`). Multi-session multi-host production requires **Postgres** or a **single allocator daemon** (ADR §6). 6. `incident_links` optional scope fields are stored as empty strings (never NULL) so UNIQUE is canonical across minimal upserts. +7. Legacy `incident_links` migration collapses NULL-scope duplicates **only** when Gitea targets **and** all meaningful observation metadata agree (fingerprint, status, event_count, permalink, timestamps, linked PRs, etc.). Conflicting metadata fails closed — no silent discard. ## Dependency chain diff --git a/tests/test_control_plane_db.py b/tests/test_control_plane_db.py index da8f0ed..c63cd39 100644 --- a/tests/test_control_plane_db.py +++ b/tests/test_control_plane_db.py @@ -686,6 +686,93 @@ class ControlPlaneDBTest(unittest.TestCase): ControlPlaneDB(path) self.assertIn("conflicting", str(ctx.exception).lower()) + def test_migrate_conflicting_observation_metadata_fails_closed(self) -> None: + """Same provider key + same Gitea target but differing obs metadata must fail closed. + + Regression for silent data loss: migration used to keep lowest link_id and + delete peers after comparing only Gitea targets (#619 RC3). + """ + import sqlite3 + from control_plane_db import ControlPlaneDB, ControlPlaneError + + path = os.path.join(self._tmp.name, "legacy_meta_conflict.sqlite3") + conn = sqlite3.connect(path) + try: + conn.executescript( + """ + CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL); + CREATE TABLE incident_links ( + link_id INTEGER PRIMARY KEY AUTOINCREMENT, + provider TEXT NOT NULL, + provider_base_url TEXT, + provider_org TEXT, + provider_project TEXT, + provider_issue_id TEXT NOT NULL, + provider_short_id TEXT, + provider_permalink TEXT, + fingerprint TEXT, + gitea_org TEXT NOT NULL, + gitea_repo TEXT NOT NULL, + gitea_issue_number INTEGER NOT NULL, + linked_pr_numbers TEXT, + first_seen TEXT, + last_seen TEXT, + event_count INTEGER, + status TEXT NOT NULL DEFAULT 'open', + release_resolved_at TEXT, + last_sync_at TEXT, + UNIQUE ( + provider, provider_base_url, provider_org, + provider_project, provider_issue_id + ) + ); + INSERT INTO incident_links( + provider, provider_base_url, provider_org, provider_project, + provider_issue_id, provider_permalink, fingerprint, + gitea_org, gitea_repo, gitea_issue_number, + event_count, status, first_seen, last_seen + ) VALUES + ( + 'sentry', NULL, NULL, NULL, 'dup-meta', + 'https://sentry.example/issues/1', 'fingerprint-A', + 'org', 'repo', 42, + 1, 'open', '2026-01-01T00:00:00Z', '2026-01-01T01:00:00Z' + ), + ( + 'sentry', NULL, NULL, NULL, 'dup-meta', + 'https://sentry.example/issues/1', 'fingerprint-B', + 'org', 'repo', 42, + 99, 'resolved', '2026-01-01T00:00:00Z', '2026-01-02T00:00:00Z' + ); + """ + ) + n = conn.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0] + self.assertEqual(n, 2) + conn.commit() + finally: + conn.close() + + with self.assertRaises(ControlPlaneError) as ctx: + ControlPlaneDB(path) + msg = str(ctx.exception).lower() + self.assertIn("conflicting", msg) + self.assertIn("observation metadata", msg) + + # Rows must still be present — migration must not delete before failing. + conn2 = sqlite3.connect(path) + try: + remaining = conn2.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0] + fps = { + r[0] + for r in conn2.execute( + "SELECT fingerprint FROM incident_links ORDER BY link_id" + ).fetchall() + } + finally: + conn2.close() + self.assertEqual(remaining, 2) + self.assertEqual(fps, {"fingerprint-A", "fingerprint-B"}) + def test_incident_links_minimal_upsert_is_canonical(self) -> None: """Repeated minimal upserts must update one row (NULL-safe uniqueness).""" a = self.db.upsert_incident_link(