sysadminandClaude Opus 4.8
7e18dccbd9
Add a versioned, redacted, reconcile-on-boot session checkpoint store to control_plane_db so a restart can recover session identity, stage, lease ownership, and next valid action instead of forcing human reconstruction (umbrella #655; #628 autonomous-handoff goal). - Bump SCHEMA_VERSION 4->5. New `session_checkpoints` table + two indexes, added via `CREATE TABLE IF NOT EXISTS` so table creation is itself the additive, idempotent v4->v5 migration (dependency_edges precedent). - Writer `write_session_checkpoint` upserts the current recoverable state per (remote, org, repo, session_id, work_kind, work_number); stage transitions audit to `events`. Readers `get_session_checkpoint` / `list_session_checkpoints`. - Every free-text and JSON field is passed through `gitea_audit.redact` before storage — no tokens/credential URLs can land in a checkpoint (AC4). - `reconcile_session_checkpoint` is pure and never restores: it diagnoses a stored checkpoint against live head/lease state and flags staleness (AC3). - Drain gate: `require_complete=True` fails closed (writes nothing) when a checkpoint is missing a drain-required field, so drain cannot complete on an unrecoverable record. - `lease_id`/`assignment_id` are soft references (no enforced FK) so a checkpoint survives deletion of the lease it names; `work_number=0` is the NULL-safe session-level sentinel. Tests: 13 new cases (schema/version, multi-role fixtures, upsert+audit, JSON round-trip, secret redaction, reconcile stale-head/dead-lease/ reassigned-lease/clean/unknown, drain fail-closed, sentinel key). Existing schema_version assertion updated 4->5. Full tests/test_control_plane_db.py suite: 33/33 pass. Links #652 #653 #655. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
1050 lines
38 KiB
Python
1050 lines
38 KiB
Python
"""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,
|
|
ControlPlaneError,
|
|
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"], "5")
|
|
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,
|
|
expected_head_sha="head-50",
|
|
)
|
|
b = self.db.assign_and_lease(
|
|
session_id="s1",
|
|
role="reviewer",
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="pr",
|
|
number=50,
|
|
expected_head_sha="head-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,
|
|
expected_head_sha="merged-head",
|
|
)
|
|
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")
|
|
|
|
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_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_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(
|
|
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)
|
|
|
|
|
|
class SessionCheckpointTest(unittest.TestCase):
|
|
"""Durable MCP session checkpoint schema, redaction, and reconcile (#660)."""
|
|
|
|
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 _write(self, **overrides):
|
|
base = dict(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
session_id="prgs-author-1-abc",
|
|
role="author",
|
|
work_kind="issue",
|
|
work_number=660,
|
|
branch="feat/issue-660-session-checkpoint-schema",
|
|
head_sha="deadbeef",
|
|
lease_id="lease-1",
|
|
workflow_stage="implementing",
|
|
last_completed_action="wrote schema",
|
|
next_valid_action="write tests",
|
|
recovery_instructions="re-lock #660 then continue tests",
|
|
)
|
|
base.update(overrides)
|
|
return self.db.write_session_checkpoint(**base)
|
|
|
|
# AC1 — schema documented and versioned.
|
|
def test_table_exists_and_row_carries_schema_version(self) -> None:
|
|
import sqlite3
|
|
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
names = {
|
|
r[0]
|
|
for r in conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'"
|
|
).fetchall()
|
|
}
|
|
finally:
|
|
conn.close()
|
|
self.assertIn("session_checkpoints", names)
|
|
record = self._write()
|
|
self.assertEqual(record["checkpoint_schema_version"], 5)
|
|
|
|
# AC2 — checkpoints written for multi-role session fixtures.
|
|
def test_multi_role_fixtures_each_get_a_row(self) -> None:
|
|
roles = [
|
|
("prgs-author-1", "author", "issue", 660),
|
|
("prgs-reviewer-2", "reviewer", "pr", 795),
|
|
("prgs-merger-3", "merger", "pr", 862),
|
|
("prgs-controller-4", "controller", "issue", 653),
|
|
]
|
|
for session_id, role, kind, number in roles:
|
|
self._write(
|
|
session_id=session_id,
|
|
role=role,
|
|
work_kind=kind,
|
|
work_number=number,
|
|
lease_id=f"lease-{session_id}",
|
|
)
|
|
rows = self.db.list_session_checkpoints(remote="prgs")
|
|
self.assertEqual(len(rows), 4)
|
|
self.assertEqual(
|
|
{r["role"] for r in rows},
|
|
{"author", "reviewer", "merger", "controller"},
|
|
)
|
|
|
|
def test_upsert_is_current_state_and_audits_stage_change(self) -> None:
|
|
first = self._write(workflow_stage="implementing")
|
|
second = self._write(workflow_stage="testing")
|
|
self.assertEqual(first["checkpoint_id"], second["checkpoint_id"])
|
|
rows = self.db.list_session_checkpoints(
|
|
remote="prgs", session_id="prgs-author-1-abc"
|
|
)
|
|
self.assertEqual(len(rows), 1)
|
|
self.assertEqual(rows[0]["workflow_stage"], "testing")
|
|
|
|
import sqlite3
|
|
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
n = conn.execute(
|
|
"SELECT COUNT(*) FROM events "
|
|
"WHERE event_type = 'session_checkpoint_stage_change'"
|
|
).fetchone()[0]
|
|
finally:
|
|
conn.close()
|
|
self.assertEqual(n, 1)
|
|
|
|
def test_get_and_roundtrip_json_fields(self) -> None:
|
|
self._write(
|
|
capabilities=["gitea.repo.commit", "gitea.pr.create"],
|
|
evidence={"tests": "4 passing"},
|
|
pending_mutation={"op": "commit_files", "files": ["control_plane_db.py"]},
|
|
)
|
|
got = self.db.get_session_checkpoint(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
session_id="prgs-author-1-abc",
|
|
work_kind="issue",
|
|
work_number=660,
|
|
)
|
|
self.assertIsNotNone(got)
|
|
self.assertEqual(got["capabilities"], ["gitea.repo.commit", "gitea.pr.create"])
|
|
self.assertEqual(got["evidence"], {"tests": "4 passing"})
|
|
self.assertEqual(got["pending_mutation"]["op"], "commit_files")
|
|
|
|
# AC4 — no secrets in stored records.
|
|
def test_secrets_are_redacted_before_storage(self) -> None:
|
|
self._write(
|
|
recovery_instructions=(
|
|
"resume with Authorization: Bearer sk-supersecrettoken then retry"
|
|
),
|
|
evidence={"authorization": "Bearer sk-anothersecret"},
|
|
pending_mutation={"url": "https://user:[email protected]/repo.git"},
|
|
)
|
|
import sqlite3
|
|
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT recovery_instructions, evidence, pending_mutation "
|
|
"FROM session_checkpoints"
|
|
).fetchone()
|
|
finally:
|
|
conn.close()
|
|
blob = " ".join(str(v) for v in row)
|
|
self.assertNotIn("sk-supersecrettoken", blob)
|
|
self.assertNotIn("sk-anothersecret", blob)
|
|
self.assertNotIn("password", blob)
|
|
self.assertIn("REDACTED", blob)
|
|
|
|
# AC3 — reconcile detects stale head / lease mismatch.
|
|
def test_reconcile_flags_stale_head(self) -> None:
|
|
record = self._write(head_sha="aaaa1111")
|
|
result = self.db.reconcile_session_checkpoint(
|
|
record, live_head_sha="bbbb2222", live_lease_active=True,
|
|
live_lease_id="lease-1",
|
|
)
|
|
self.assertTrue(result["stale"])
|
|
self.assertTrue(result["head_mismatch"])
|
|
self.assertFalse(result["lease_mismatch"])
|
|
self.assertEqual(result["reconcile_action"], "reconcile_required")
|
|
|
|
def test_reconcile_flags_dead_lease(self) -> None:
|
|
record = self._write(lease_id="lease-1", head_sha="aaaa1111")
|
|
result = self.db.reconcile_session_checkpoint(
|
|
record, live_head_sha="aaaa1111", live_lease_active=False,
|
|
)
|
|
self.assertTrue(result["stale"])
|
|
self.assertFalse(result["head_mismatch"])
|
|
self.assertTrue(result["lease_mismatch"])
|
|
|
|
def test_reconcile_reassigned_lease_is_stale(self) -> None:
|
|
record = self._write(lease_id="lease-1")
|
|
result = self.db.reconcile_session_checkpoint(
|
|
record, live_lease_active=True, live_lease_id="lease-999",
|
|
)
|
|
self.assertTrue(result["lease_mismatch"])
|
|
|
|
def test_reconcile_clean_state_is_safe_to_resume(self) -> None:
|
|
record = self._write(head_sha="aaaa1111", lease_id="lease-1")
|
|
result = self.db.reconcile_session_checkpoint(
|
|
record, live_head_sha="aaaa1111", live_lease_active=True,
|
|
live_lease_id="lease-1",
|
|
)
|
|
self.assertFalse(result["stale"])
|
|
self.assertEqual(result["reconcile_action"], "safe_to_resume")
|
|
|
|
def test_unknown_live_state_never_flags_mismatch(self) -> None:
|
|
record = self._write(head_sha="aaaa1111", lease_id="lease-1")
|
|
result = self.db.reconcile_session_checkpoint(record)
|
|
self.assertFalse(result["stale"])
|
|
|
|
# Drain gate — fail closed when a checkpoint is incomplete.
|
|
def test_drain_requires_complete_checkpoint(self) -> None:
|
|
with self.assertRaises(ControlPlaneError):
|
|
self.db.write_session_checkpoint(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
session_id="prgs-author-1-abc",
|
|
role="author",
|
|
workflow_stage="draining",
|
|
# next_valid_action + recovery_instructions intentionally absent
|
|
require_complete=True,
|
|
)
|
|
# Nothing was written.
|
|
rows = self.db.list_session_checkpoints(remote="prgs")
|
|
self.assertEqual(rows, [])
|
|
|
|
def test_drain_write_succeeds_when_complete(self) -> None:
|
|
record = self._write(require_complete=True)
|
|
self.assertEqual(record["status"], "active")
|
|
self.assertEqual(self.db.checkpoint_completeness(record), [])
|
|
|
|
def test_missing_session_id_fails_closed(self) -> None:
|
|
with self.assertRaises(ControlPlaneError):
|
|
self.db.write_session_checkpoint(
|
|
remote="prgs", org="o", repo="r", session_id="",
|
|
)
|
|
|
|
def test_session_level_checkpoint_uses_sentinel_key(self) -> None:
|
|
# No work unit -> ('', 0) sentinel; a second session-level write upserts.
|
|
self.db.write_session_checkpoint(
|
|
remote="prgs", org="o", repo="r", session_id="s-sess",
|
|
workflow_stage="idle",
|
|
)
|
|
self.db.write_session_checkpoint(
|
|
remote="prgs", org="o", repo="r", session_id="s-sess",
|
|
workflow_stage="booting",
|
|
)
|
|
rows = self.db.list_session_checkpoints(remote="prgs", session_id="s-sess")
|
|
self.assertEqual(len(rows), 1)
|
|
self.assertEqual(rows[0]["work_kind"], "")
|
|
self.assertEqual(rows[0]["work_number"], 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|