from __future__ import annotations import json import sqlite3 from datetime import UTC, datetime from hashlib import sha256 from typing import Any from uuid import uuid4 from .models import ( DiscoveryConfidence, DiscoveryDiff, DiscoveryLifecycle, DiscoveryObservation, DiscoveryRun, DiscoveryRunStatus, LogicalRelation, RelationEvidence, RelationType, ServiceSource, SuggestionStatus, TopologyChange, TopologyDiff, TopologySnapshot, TopologySuggestion, ) from .store import Store def _now() -> datetime: return datetime.now(UTC) def _iso(value: datetime | None) -> str | None: return value.isoformat() if value else None def _json(value: object) -> str: return json.dumps(value, separators=(",", ":"), sort_keys=True) def _obj(value: object) -> dict[str, object]: if isinstance(value, dict): return value if isinstance(value, str): try: parsed = json.loads(value) return parsed if isinstance(parsed, dict) else {} except json.JSONDecodeError: return {} return {} def _list(value: object) -> list[str]: if isinstance(value, list): return [str(item) for item in value] if isinstance(value, str): try: parsed = json.loads(value) return [str(item) for item in parsed] if isinstance(parsed, list) else [] except json.JSONDecodeError: return [] return [] _CONFIDENCE_RANK = { DiscoveryConfidence.LOW: 1, DiscoveryConfidence.MEDIUM: 2, DiscoveryConfidence.HIGH: 3, DiscoveryConfidence.CONFIRMED: 4, } _CONFIDENCE_RANK_INV = {rank: confidence.value for confidence, rank in _CONFIDENCE_RANK.items()} def _higher_confidence(*values: DiscoveryConfidence) -> DiscoveryConfidence: return max(values, key=lambda value: _CONFIDENCE_RANK[value]) class DiscoveryStore: """Durable discovery observations and review state. The existing Store remains the owner of canonical services and probe state. This companion store deliberately keeps observations and suggestions separate so a discovery run cannot mutate health topology unless reconciliation or an explicit suggestion action asks it to do so. """ def __init__(self, store: Store) -> None: self.store = store self._init_schema() self.ensure_baseline_snapshot() def _connect(self) -> sqlite3.Connection: return self.store._connect() def _init_schema(self) -> None: with self._connect() as db: db.executescript( """ CREATE TABLE IF NOT EXISTS discovery_runs ( id TEXT PRIMARY KEY, status TEXT NOT NULL, started_at TEXT NOT NULL, completed_at TEXT, source_counts TEXT NOT NULL DEFAULT '{}', diff TEXT NOT NULL DEFAULT '{}', error TEXT ); CREATE INDEX IF NOT EXISTS idx_discovery_runs_status ON discovery_runs(status, started_at DESC); CREATE TABLE IF NOT EXISTS discovery_sources ( id TEXT PRIMARY KEY, source_type TEXT NOT NULL, source_id TEXT NOT NULL, source_name TEXT, first_seen_at TEXT NOT NULL, last_seen_at TEXT NOT NULL, lifecycle TEXT NOT NULL DEFAULT 'active', managed INTEGER NOT NULL DEFAULT 1, metadata TEXT NOT NULL DEFAULT '{}', UNIQUE(source_type, source_id) ); CREATE INDEX IF NOT EXISTS idx_discovery_sources_identity ON discovery_sources(source_type, source_id); CREATE INDEX IF NOT EXISTS idx_discovery_sources_last_seen ON discovery_sources(last_seen_at DESC); CREATE TABLE IF NOT EXISTS service_sources ( service_id TEXT NOT NULL, source_id TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'observed', PRIMARY KEY(service_id, source_id), FOREIGN KEY(service_id) REFERENCES services(id) ON DELETE CASCADE, FOREIGN KEY(source_id) REFERENCES discovery_sources(id) ON DELETE CASCADE ); CREATE INDEX IF NOT EXISTS idx_service_sources_source ON service_sources(source_id, service_id); CREATE TABLE IF NOT EXISTS discovery_observations ( id TEXT PRIMARY KEY, run_id TEXT NOT NULL, source_id TEXT NOT NULL, source_type TEXT NOT NULL, source_name TEXT NOT NULL, lifecycle TEXT NOT NULL, change_type TEXT NOT NULL, observed_at TEXT NOT NULL, metadata TEXT NOT NULL DEFAULT '{}', FOREIGN KEY(run_id) REFERENCES discovery_runs(id) ON DELETE CASCADE, FOREIGN KEY(source_id) REFERENCES discovery_sources(id) ON DELETE CASCADE ); CREATE INDEX IF NOT EXISTS idx_discovery_observations_run ON discovery_observations(run_id, observed_at DESC); CREATE INDEX IF NOT EXISTS idx_discovery_observations_source ON discovery_observations(source_id, observed_at DESC); CREATE TABLE IF NOT EXISTS relation_evidence ( id TEXT PRIMARY KEY, source_service_id TEXT NOT NULL, target_service_id TEXT, target_ref TEXT, relation_type TEXT NOT NULL DEFAULT 'depends_on', evidence_type TEXT NOT NULL, confidence TEXT NOT NULL, explanation TEXT NOT NULL, source_ref TEXT, locked INTEGER NOT NULL DEFAULT 0, active INTEGER NOT NULL DEFAULT 1, first_seen_at TEXT NOT NULL, last_seen_at TEXT NOT NULL, UNIQUE(source_service_id, target_service_id, target_ref, relation_type, evidence_type), FOREIGN KEY(source_service_id) REFERENCES services(id) ON DELETE CASCADE, FOREIGN KEY(target_service_id) REFERENCES services(id) ON DELETE SET NULL ); CREATE INDEX IF NOT EXISTS idx_relation_evidence_source_target ON relation_evidence(source_service_id, target_service_id); CREATE INDEX IF NOT EXISTS idx_relation_evidence_confidence ON relation_evidence(confidence, active); CREATE TABLE IF NOT EXISTS topology_suggestions ( id TEXT PRIMARY KEY, run_id TEXT NOT NULL, suggestion_type TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'open', fingerprint TEXT NOT NULL UNIQUE, source_ref TEXT, source_service_id TEXT, target_ref TEXT, target_service_id TEXT, proposed_name TEXT, proposed_kind TEXT, proposed_url TEXT, proposed_host TEXT, proposed_port INTEGER, proposed_category TEXT NOT NULL DEFAULT 'unknown', proposed_group TEXT, relation_type TEXT, confidence TEXT NOT NULL, evidence TEXT NOT NULL DEFAULT '[]', created_at TEXT NOT NULL, updated_at TEXT NOT NULL, FOREIGN KEY(run_id) REFERENCES discovery_runs(id) ON DELETE CASCADE ); CREATE INDEX IF NOT EXISTS idx_topology_suggestions_status ON topology_suggestions(status, updated_at DESC); CREATE INDEX IF NOT EXISTS idx_topology_suggestions_source ON topology_suggestions(source_ref, target_ref); CREATE TABLE IF NOT EXISTS topology_overrides ( target_type TEXT NOT NULL, target_id TEXT NOT NULL, field_name TEXT NOT NULL, value TEXT NOT NULL, locked INTEGER NOT NULL DEFAULT 1, updated_at TEXT NOT NULL, PRIMARY KEY(target_type, target_id, field_name) ); CREATE INDEX IF NOT EXISTS idx_topology_overrides_target ON topology_overrides(target_type, target_id); CREATE TABLE IF NOT EXISTS topology_change_events ( id TEXT PRIMARY KEY, event_type TEXT NOT NULL, service_id TEXT, source_service_id TEXT, target_service_id TEXT, detail TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_topology_change_events_time ON topology_change_events(created_at DESC); CREATE TABLE IF NOT EXISTS topology_snapshots ( id TEXT PRIMARY KEY, run_id TEXT, created_at TEXT NOT NULL, revision TEXT NOT NULL UNIQUE, service_count INTEGER NOT NULL, relationship_count INTEGER NOT NULL, summary TEXT NOT NULL DEFAULT '{}', FOREIGN KEY(run_id) REFERENCES discovery_runs(id) ON DELETE SET NULL ); CREATE INDEX IF NOT EXISTS idx_topology_snapshots_time ON topology_snapshots(created_at DESC); CREATE TABLE IF NOT EXISTS topology_snapshot_services ( snapshot_id TEXT NOT NULL, service_id TEXT NOT NULL, name TEXT NOT NULL, fingerprint TEXT NOT NULL, payload TEXT NOT NULL, PRIMARY KEY(snapshot_id, service_id), FOREIGN KEY(snapshot_id) REFERENCES topology_snapshots(id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS topology_snapshot_relations ( snapshot_id TEXT NOT NULL, relation_id TEXT NOT NULL, source_service_id TEXT NOT NULL, target_service_id TEXT, relation_type TEXT NOT NULL, fingerprint TEXT NOT NULL, payload TEXT NOT NULL, PRIMARY KEY(snapshot_id, relation_id), FOREIGN KEY(snapshot_id) REFERENCES topology_snapshots(id) ON DELETE CASCADE ); """ ) @staticmethod def _diff(value: object) -> DiscoveryDiff: data = _obj(value) return DiscoveryDiff(**{key: int(data.get(key, 0) or 0) for key in DiscoveryDiff.model_fields}) @classmethod def _run(cls, row: sqlite3.Row | None) -> DiscoveryRun | None: if row is None: return None return DiscoveryRun( id=row["id"], status=DiscoveryRunStatus(row["status"]), started_at=datetime.fromisoformat(row["started_at"]), completed_at=datetime.fromisoformat(row["completed_at"]) if row["completed_at"] else None, source_counts={str(k): int(v) for k, v in _obj(row["source_counts"]).items()}, diff=cls._diff(row["diff"]), error=row["error"], ) @classmethod def _source(cls, row: sqlite3.Row) -> ServiceSource: return ServiceSource( id=row["id"], source_type=row["source_type"], source_id=row["source_id"], source_name=row["source_name"], first_seen_at=datetime.fromisoformat(row["first_seen_at"]), last_seen_at=datetime.fromisoformat(row["last_seen_at"]), lifecycle=DiscoveryLifecycle(row["lifecycle"]), managed=bool(row["managed"]), metadata=_obj(row["metadata"]), ) @classmethod def _suggestion(cls, row: sqlite3.Row) -> TopologySuggestion: return TopologySuggestion( id=row["id"], run_id=row["run_id"], suggestion_type=row["suggestion_type"], status=SuggestionStatus(row["status"]), source_ref=row["source_ref"], source_service_id=row["source_service_id"], target_ref=row["target_ref"], target_service_id=row["target_service_id"], proposed_name=row["proposed_name"], proposed_kind=row["proposed_kind"], proposed_url=row["proposed_url"], proposed_host=row["proposed_host"], proposed_port=row["proposed_port"], proposed_category=row["proposed_category"], proposed_group=row["proposed_group"], relation_type=row["relation_type"], confidence=DiscoveryConfidence(row["confidence"]), evidence=_list(row["evidence"]), created_at=datetime.fromisoformat(row["created_at"]), updated_at=datetime.fromisoformat(row["updated_at"]), ) @classmethod def _evidence(cls, row: sqlite3.Row) -> RelationEvidence: return RelationEvidence( id=row["id"], source_service_id=row["source_service_id"], target_service_id=row["target_service_id"], target_ref=row["target_ref"], relation_type=RelationType(row["relation_type"]), evidence_type=row["evidence_type"], confidence=DiscoveryConfidence(row["confidence"]), explanation=row["explanation"], source_ref=row["source_ref"], locked=bool(row["locked"]), active=bool(row["active"]), first_seen_at=datetime.fromisoformat(row["first_seen_at"]), last_seen_at=datetime.fromisoformat(row["last_seen_at"]), ) def start_run(self) -> DiscoveryRun: run = DiscoveryRun(id=f"disc-{uuid4().hex[:12]}", status=DiscoveryRunStatus.DISCOVERING, started_at=_now()) with self._connect() as db: # A process restart or an unexpected reconcile exception can leave # a durable run in an active state. Close that stale marker before # creating the next run so status never reports a phantom active # discovery forever. db.execute( "UPDATE discovery_runs SET status=?, completed_at=?, error=COALESCE(error, ?) WHERE status IN (?, ?)", ( DiscoveryRunStatus.FAILED.value, run.started_at.isoformat(), "Superseded by a newer discovery run", DiscoveryRunStatus.DISCOVERING.value, DiscoveryRunStatus.RECONCILING.value, ), ) db.execute( "INSERT INTO discovery_runs (id, status, started_at) VALUES (?, ?, ?)", (run.id, run.status.value, run.started_at.isoformat()), ) return run def finish_run( self, run_id: str, *, status: DiscoveryRunStatus, source_counts: dict[str, int], diff: DiscoveryDiff, error: str | None = None, ) -> DiscoveryRun: completed = _now() with self._connect() as db: db.execute( "UPDATE discovery_runs SET status=?, completed_at=?, source_counts=?, diff=?, error=? WHERE id=?", (status.value, completed.isoformat(), _json(source_counts), _json(diff.model_dump()), error, run_id), ) return self.get_run(run_id) or DiscoveryRun( id=run_id, status=status, started_at=completed, completed_at=completed, source_counts=source_counts, diff=diff, error=error, ) def get_run(self, run_id: str) -> DiscoveryRun | None: with self._connect() as db: row = db.execute("SELECT * FROM discovery_runs WHERE id = ?", (run_id,)).fetchone() return self._run(row) def last_completed(self) -> DiscoveryRun | None: with self._connect() as db: row = db.execute( "SELECT * FROM discovery_runs WHERE status = 'completed' ORDER BY completed_at DESC LIMIT 1" ).fetchone() return self._run(row) def status(self) -> tuple[DiscoveryRun | None, DiscoveryRun | None]: with self._connect() as db: current_row = db.execute( "SELECT * FROM discovery_runs WHERE status IN ('discovering', 'reconciling') ORDER BY started_at DESC LIMIT 1" ).fetchone() return self._run(current_row), self.last_completed() @staticmethod def _snapshot_row(row: sqlite3.Row | None) -> TopologySnapshot | None: if row is None: return None return TopologySnapshot( id=row["id"], run_id=row["run_id"], created_at=datetime.fromisoformat(row["created_at"]), revision=row["revision"], service_count=int(row["service_count"]), relationship_count=int(row["relationship_count"]), summary={str(key): int(value) for key, value in _obj(row["summary"]).items()}, ) def ensure_baseline_snapshot(self) -> TopologySnapshot: """Create one migration baseline without manufacturing repeated no-op snapshots.""" with self._connect() as db: existing = db.execute("SELECT * FROM topology_snapshots ORDER BY created_at DESC LIMIT 1").fetchone() return self._snapshot_row(existing) if existing else self.capture_snapshot(run_id=None) def capture_snapshot(self, *, run_id: str | None) -> TopologySnapshot: """Persist a compact canonical topology capture when its revision changes.""" services = sorted(self.store.list_services(), key=lambda item: item.id) service_rows: list[tuple[str, str, str, dict[str, object]]] = [] relation_rows: list[tuple[str, str, str | None, str, str, dict[str, object]]] = [] for service in services: payload: dict[str, object] = { "id": service.id, "name": service.name, "kind": service.kind.value, "source_ref": service.source_ref, "application": service.application, "domain": service.domain, "component_role": service.component_role.value, "category": service.category.value, "criticality": service.criticality.value, "visibility": service.visibility.value, "lifecycle": service.lifecycle.value, "monitoring_state": service.monitoring_state.value, "depends_on": sorted(service.depends_on), } service_rows.append((service.id, service.name, sha256(_json(payload).encode()).hexdigest(), payload)) for target_id in sorted(service.depends_on): relation_payload: dict[str, object] = { "source_service_id": service.id, "target_service_id": target_id, "relation_type": RelationType.DEPENDS_ON.value, } relation_id = sha256(_json(relation_payload).encode()).hexdigest()[:24] relation_rows.append( ( relation_id, service.id, target_id, RelationType.DEPENDS_ON.value, sha256(_json(relation_payload).encode()).hexdigest(), relation_payload, ) ) revision_payload = {"services": [(item[0], item[2]) for item in service_rows], "relations": [(item[0], item[4]) for item in relation_rows]} revision = sha256(_json(revision_payload).encode()).hexdigest() with self._connect() as db: existing = db.execute("SELECT * FROM topology_snapshots WHERE revision=?", (revision,)).fetchone() if existing: return self._snapshot_row(existing) # type: ignore[return-value] created_at = _now() summary = { "visible_services": sum(1 for item in services if item.visibility.value == "visible"), "active_services": sum(1 for item in services if item.lifecycle.value == "active"), } snapshot = TopologySnapshot( id=f"topo-{uuid4().hex[:14]}", run_id=run_id, created_at=created_at, revision=revision, service_count=len(service_rows), relationship_count=len(relation_rows), summary=summary, ) db.execute( "INSERT INTO topology_snapshots VALUES (?, ?, ?, ?, ?, ?, ?)", (snapshot.id, snapshot.run_id, snapshot.created_at.isoformat(), snapshot.revision, snapshot.service_count, snapshot.relationship_count, _json(snapshot.summary)), ) db.executemany( "INSERT INTO topology_snapshot_services VALUES (?, ?, ?, ?, ?)", [(snapshot.id, service_id, name, fingerprint, _json(payload)) for service_id, name, fingerprint, payload in service_rows], ) db.executemany( "INSERT INTO topology_snapshot_relations VALUES (?, ?, ?, ?, ?, ?, ?)", [(snapshot.id, relation_id, source_id, target_id, relation_type, fingerprint, _json(payload)) for relation_id, source_id, target_id, relation_type, fingerprint, payload in relation_rows], ) self.record_change("topology_snapshot", f"Topology snapshot captured ({snapshot.service_count} services)") return snapshot def list_snapshots(self, *, limit: int = 50) -> list[TopologySnapshot]: with self._connect() as db: rows = db.execute("SELECT * FROM topology_snapshots ORDER BY created_at DESC LIMIT ?", (limit,)).fetchall() return [self._snapshot_row(row) for row in rows if self._snapshot_row(row) is not None] # type: ignore[list-item] def topology_diff(self, *, after_snapshot_id: str, before_snapshot_id: str | None = None) -> TopologyDiff: with self._connect() as db: after = db.execute("SELECT * FROM topology_snapshots WHERE id=?", (after_snapshot_id,)).fetchone() if after is None: raise ValueError("Topology snapshot not found") before = ( db.execute("SELECT * FROM topology_snapshots WHERE id=?", (before_snapshot_id,)).fetchone() if before_snapshot_id else db.execute("SELECT * FROM topology_snapshots WHERE created_at < ? ORDER BY created_at DESC LIMIT 1", (after["created_at"],)).fetchone() ) after_services = { row["service_id"]: row for row in db.execute("SELECT * FROM topology_snapshot_services WHERE snapshot_id=?", (after_snapshot_id,)).fetchall() } before_services = ( {row["service_id"]: row for row in db.execute("SELECT * FROM topology_snapshot_services WHERE snapshot_id=?", (before["id"],)).fetchall()} if before else {} ) after_relations = { row["relation_id"]: row for row in db.execute("SELECT * FROM topology_snapshot_relations WHERE snapshot_id=?", (after_snapshot_id,)).fetchall() } before_relations = ( {row["relation_id"]: row for row in db.execute("SELECT * FROM topology_snapshot_relations WHERE snapshot_id=?", (before["id"],)).fetchall()} if before else {} ) added: list[TopologyChange] = [] removed: list[TopologyChange] = [] changed: list[TopologyChange] = [] for item_id, row in after_services.items(): payload = _obj(row["payload"]) if item_id not in before_services: added.append(TopologyChange(change_type="added", entity_type="service", entity_id=item_id, name=row["name"], after=payload)) elif row["fingerprint"] != before_services[item_id]["fingerprint"]: changed.append(TopologyChange(change_type="changed", entity_type="service", entity_id=item_id, name=row["name"], before=_obj(before_services[item_id]["payload"]), after=payload)) for item_id, row in before_services.items(): if item_id not in after_services: removed.append(TopologyChange(change_type="removed", entity_type="service", entity_id=item_id, name=row["name"], before=_obj(row["payload"]))) for item_id, row in after_relations.items(): name = f"{row['source_service_id']} → {row['target_service_id'] or 'unknown'}" if item_id not in before_relations: added.append(TopologyChange(change_type="added", entity_type="relationship", entity_id=item_id, name=name, after=_obj(row["payload"]))) elif row["fingerprint"] != before_relations[item_id]["fingerprint"]: changed.append(TopologyChange(change_type="changed", entity_type="relationship", entity_id=item_id, name=name, before=_obj(before_relations[item_id]["payload"]), after=_obj(row["payload"]))) for item_id, row in before_relations.items(): if item_id not in after_relations: removed.append(TopologyChange(change_type="removed", entity_type="relationship", entity_id=item_id, name=f"{row['source_service_id']} → {row['target_service_id'] or 'unknown'}", before=_obj(row["payload"]))) return TopologyDiff( before_snapshot_id=before["id"] if before else None, after_snapshot_id=after_snapshot_id, added=added, removed=removed, changed=changed, summary={"added": len(added), "removed": len(removed), "changed": len(changed)}, ) def save_source( self, *, source_type: str, source_id: str, source_name: str | None, lifecycle: DiscoveryLifecycle, metadata: dict[str, object], managed: bool = True, observed_at: datetime | None = None, ) -> ServiceSource: at = observed_at or _now() with self._connect() as db: existing = db.execute( "SELECT * FROM discovery_sources WHERE source_type=? AND source_id=?", (source_type, source_id), ).fetchone() source_pk = existing["id"] if existing else f"src-{uuid4().hex[:14]}" first_seen = existing["first_seen_at"] if existing else at.isoformat() db.execute( """ INSERT INTO discovery_sources (id, source_type, source_id, source_name, first_seen_at, last_seen_at, lifecycle, managed, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(source_type, source_id) DO UPDATE SET source_name=excluded.source_name, last_seen_at=excluded.last_seen_at, lifecycle=excluded.lifecycle, managed=excluded.managed, metadata=excluded.metadata """, (source_pk, source_type, source_id, source_name, first_seen, at.isoformat(), lifecycle.value, int(managed), _json(metadata)), ) row = db.execute("SELECT * FROM discovery_sources WHERE id=?", (source_pk,)).fetchone() return self._source(row) def attach_source(self, service_id: str, source_id: str, role: str = "observed") -> None: with self._connect() as db: db.execute( "INSERT OR IGNORE INTO service_sources (service_id, source_id, role) VALUES (?, ?, ?)", (service_id, source_id, role), ) def get_source(self, source_type: str, source_id: str) -> ServiceSource | None: with self._connect() as db: row = db.execute( "SELECT * FROM discovery_sources WHERE source_type=? AND source_id=?", (source_type, source_id), ).fetchone() return self._source(row) if row else None def save_observation( self, *, run_id: str, source: ServiceSource, source_name: str, lifecycle: DiscoveryLifecycle, change_type: str, metadata: dict[str, object], observed_at: datetime | None = None, ) -> DiscoveryObservation: at = observed_at or _now() observation = DiscoveryObservation( id=f"obs-{uuid4().hex[:14]}", run_id=run_id, source_id=source.id, source_type=source.source_type, source_name=source_name, lifecycle=lifecycle, change_type=change_type, observed_at=at, metadata=metadata, ) with self._connect() as db: db.execute( "INSERT INTO discovery_observations VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", (observation.id, run_id, source.id, source.source_type, source_name, lifecycle.value, change_type, at.isoformat(), _json(metadata)), ) return observation def save_suggestion(self, suggestion: TopologySuggestion, fingerprint: str) -> TopologySuggestion: with self._connect() as db: existing = db.execute( "SELECT evidence, confidence FROM topology_suggestions WHERE fingerprint=?", (fingerprint,) ).fetchone() merged_evidence = list(dict.fromkeys([ *(_list(existing["evidence"]) if existing else []), *suggestion.evidence, ])) confidence = _higher_confidence( suggestion.confidence, DiscoveryConfidence(existing["confidence"]) if existing else suggestion.confidence, ) db.execute( """ INSERT INTO topology_suggestions (id, run_id, suggestion_type, status, fingerprint, source_ref, source_service_id, target_ref, target_service_id, proposed_name, proposed_kind, proposed_url, proposed_host, proposed_port, proposed_category, proposed_group, relation_type, confidence, evidence, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(fingerprint) DO UPDATE SET run_id=excluded.run_id, source_ref=excluded.source_ref, source_service_id=excluded.source_service_id, target_ref=excluded.target_ref, target_service_id=excluded.target_service_id, proposed_name=excluded.proposed_name, proposed_kind=excluded.proposed_kind, proposed_url=excluded.proposed_url, proposed_host=excluded.proposed_host, proposed_port=excluded.proposed_port, proposed_category=excluded.proposed_category, proposed_group=excluded.proposed_group, relation_type=excluded.relation_type, confidence=excluded.confidence, evidence=excluded.evidence, updated_at=excluded.updated_at """, ( suggestion.id, suggestion.run_id, suggestion.suggestion_type, suggestion.status.value, fingerprint, suggestion.source_ref, suggestion.source_service_id, suggestion.target_ref, suggestion.target_service_id, suggestion.proposed_name, suggestion.proposed_kind.value if suggestion.proposed_kind else None, suggestion.proposed_url, suggestion.proposed_host, suggestion.proposed_port, suggestion.proposed_category.value, suggestion.proposed_group, suggestion.relation_type.value if suggestion.relation_type else None, confidence.value, _json(merged_evidence), suggestion.created_at.isoformat(), suggestion.updated_at.isoformat(), ), ) row = db.execute("SELECT * FROM topology_suggestions WHERE fingerprint=?", (fingerprint,)).fetchone() return self._suggestion(row) def list_suggestions( self, status: SuggestionStatus | None = None, limit: int = 500, run_id: str | None = None, ) -> list[TopologySuggestion]: query = "SELECT * FROM topology_suggestions" params: list[Any] = [] filters: list[str] = [] if status: filters.append("status = ?") params.append(status.value) if run_id: filters.append("run_id = ?") params.append(run_id) if filters: query += " WHERE " + " AND ".join(filters) query += " ORDER BY updated_at DESC LIMIT ?" params.append(limit) with self._connect() as db: rows = db.execute(query, params).fetchall() return [self._suggestion(row) for row in rows] def get_suggestion(self, suggestion_id: str) -> TopologySuggestion | None: with self._connect() as db: row = db.execute("SELECT * FROM topology_suggestions WHERE id=?", (suggestion_id,)).fetchone() return self._suggestion(row) if row else None def set_suggestion_status(self, suggestion_id: str, status: SuggestionStatus) -> TopologySuggestion | None: now = _now() with self._connect() as db: db.execute("UPDATE topology_suggestions SET status=?, updated_at=? WHERE id=?", (status.value, now.isoformat(), suggestion_id)) row = db.execute("SELECT * FROM topology_suggestions WHERE id=?", (suggestion_id,)).fetchone() return self._suggestion(row) if row else None def save_evidence(self, evidence: RelationEvidence) -> RelationEvidence: with self._connect() as db: existing = db.execute( """ SELECT * FROM relation_evidence WHERE source_service_id=? AND target_service_id IS ? AND target_ref IS ? AND relation_type=? AND evidence_type=? """, (evidence.source_service_id, evidence.target_service_id, evidence.target_ref, evidence.relation_type.value, evidence.evidence_type), ).fetchone() if existing: previous = self._evidence(existing) evidence = evidence.model_copy(update={ "confidence": _higher_confidence(previous.confidence, evidence.confidence), "explanation": "; ".join(dict.fromkeys(part for part in (previous.explanation, evidence.explanation) if part)), "source_ref": evidence.source_ref or previous.source_ref, "locked": previous.locked or evidence.locked, "active": evidence.active, "first_seen_at": min(previous.first_seen_at, evidence.first_seen_at), }) db.execute( """ INSERT INTO relation_evidence (id, source_service_id, target_service_id, target_ref, relation_type, evidence_type, confidence, explanation, source_ref, locked, active, first_seen_at, last_seen_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(source_service_id, target_service_id, target_ref, relation_type, evidence_type) DO UPDATE SET target_service_id=excluded.target_service_id, confidence=excluded.confidence, explanation=excluded.explanation, source_ref=excluded.source_ref, locked=MAX(relation_evidence.locked, excluded.locked), active=excluded.active, last_seen_at=excluded.last_seen_at """, (evidence.id, evidence.source_service_id, evidence.target_service_id, evidence.target_ref, evidence.relation_type.value, evidence.evidence_type, evidence.confidence.value, evidence.explanation, evidence.source_ref, int(evidence.locked), int(evidence.active), evidence.first_seen_at.isoformat(), evidence.last_seen_at.isoformat()), ) row = db.execute( "SELECT * FROM relation_evidence WHERE id=?", (evidence.id,), ).fetchone() if row is None: row = db.execute( "SELECT * FROM relation_evidence WHERE source_service_id=? AND target_service_id IS ? AND target_ref IS ? AND relation_type=? AND evidence_type=?", (evidence.source_service_id, evidence.target_service_id, evidence.target_ref, evidence.relation_type.value, evidence.evidence_type), ).fetchone() return self._evidence(row) def list_logical_relations(self, service_id: str | None = None, active_only: bool = True) -> list[LogicalRelation]: """Return one canonical graph edge while retaining every evidence row. `relation_evidence` is an append/merge audit record. Rendering every row as an edge made a relation appear duplicated whenever independent discovery sources agreed. The aggregation is deliberately query-time so existing v1.3 evidence is preserved without a destructive rebuild. """ grouped: dict[tuple[str, str | None, str | None, RelationType], list[RelationEvidence]] = {} for evidence in self.list_evidence(service_id=service_id, active_only=active_only): key = (evidence.source_service_id, evidence.target_service_id, evidence.target_ref, evidence.relation_type) grouped.setdefault(key, []).append(evidence) result: list[LogicalRelation] = [] for (source_id, target_id, target_ref, relation_type), evidence_rows in grouped.items(): confidence = _higher_confidence(*(item.confidence for item in evidence_rows)) independent_sources = {f"{item.evidence_type}:{item.source_ref or item.explanation}" for item in evidence_rows} if len(independent_sources) > 1 and confidence != DiscoveryConfidence.CONFIRMED: confidence = DiscoveryConfidence(_CONFIDENCE_RANK_INV[_CONFIDENCE_RANK[confidence] + 1]) first_seen = min(item.first_seen_at for item in evidence_rows) last_seen = max(item.last_seen_at for item in evidence_rows) digest = sha256("|".join((source_id, target_id or "", target_ref or "", relation_type.value)).encode("utf-8")).hexdigest()[:16] result.append(LogicalRelation( id=f"relation-{digest}", source_service_id=source_id, target_service_id=target_id, target_ref=target_ref, relation_type=relation_type, confidence=confidence, active=any(item.active for item in evidence_rows), locked=any(item.locked for item in evidence_rows), first_seen_at=first_seen, last_seen_at=last_seen, evidence=sorted(evidence_rows, key=lambda item: item.last_seen_at, reverse=True), )) return sorted(result, key=lambda item: item.last_seen_at, reverse=True) def list_evidence(self, service_id: str | None = None, active_only: bool = True) -> list[RelationEvidence]: query = "SELECT * FROM relation_evidence WHERE 1=1" params: list[Any] = [] if service_id: query += " AND (source_service_id=? OR target_service_id=?)" params.extend([service_id, service_id]) if active_only: query += " AND active=1" query += " ORDER BY last_seen_at DESC" with self._connect() as db: rows = db.execute(query, params).fetchall() return [self._evidence(row) for row in rows] def get_evidence(self, evidence_id: str) -> RelationEvidence | None: with self._connect() as db: row = db.execute("SELECT * FROM relation_evidence WHERE id=?", (evidence_id,)).fetchone() return self._evidence(row) if row else None def get_evidence_for_relation( self, source_service_id: str, target_service_id: str | None, target_ref: str | None, relation_type: RelationType, evidence_type: str = "discovery", ) -> RelationEvidence | None: with self._connect() as db: row = db.execute( """ SELECT * FROM relation_evidence WHERE source_service_id=? AND target_service_id IS ? AND target_ref IS ? AND relation_type=? AND evidence_type=? """, (source_service_id, target_service_id, target_ref, relation_type.value, evidence_type), ).fetchone() return self._evidence(row) if row else None def set_evidence_state( self, evidence_id: str, *, active: bool | None = None, locked: bool | None = None, ) -> RelationEvidence | None: current = self.get_evidence(evidence_id) if current is None: return None with self._connect() as db: db.execute( "UPDATE relation_evidence SET active=?, locked=?, last_seen_at=? WHERE id=?", (int(current.active if active is None else active), int(current.locked if locked is None else locked), _now().isoformat(), evidence_id), ) return self.get_evidence(evidence_id) def evidence_for_service(self, service_id: str) -> list[RelationEvidence]: return self.list_evidence(service_id, active_only=True) def sources_for_service(self, service_id: str) -> list[ServiceSource]: with self._connect() as db: rows = db.execute( "SELECT s.* FROM discovery_sources s JOIN service_sources ss ON ss.source_id=s.id WHERE ss.service_id=? ORDER BY s.source_type, s.source_name", (service_id,), ).fetchall() return [self._source(row) for row in rows] def observations_for_run(self, run_id: str, limit: int = 1000) -> list[DiscoveryObservation]: with self._connect() as db: rows = db.execute( "SELECT * FROM discovery_observations WHERE run_id=? ORDER BY source_name COLLATE NOCASE LIMIT ?", (run_id, limit), ).fetchall() return [ DiscoveryObservation( id=row["id"], run_id=row["run_id"], source_id=row["source_id"], source_type=row["source_type"], source_name=row["source_name"], lifecycle=DiscoveryLifecycle(row["lifecycle"]), change_type=row["change_type"], observed_at=datetime.fromisoformat(row["observed_at"]), metadata=_obj(row["metadata"]), ) for row in rows ] def candidates_for_run(self, run_id: str, limit: int = 1000) -> list[dict[str, object]]: candidates: list[dict[str, object]] = [] for observation in self.observations_for_run(run_id, limit): candidate = observation.metadata.get("candidate") if isinstance(candidate, dict): candidates.append(candidate) return candidates def set_run_status(self, run_id: str, status: DiscoveryRunStatus) -> None: with self._connect() as db: db.execute("UPDATE discovery_runs SET status=? WHERE id=?", (status.value, run_id)) def record_change( self, event_type: str, detail: str, *, service_id: str | None = None, source_service_id: str | None = None, target_service_id: str | None = None, ) -> None: with self._connect() as db: db.execute( "INSERT INTO topology_change_events (id, event_type, service_id, source_service_id, target_service_id, detail, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", (f"change-{uuid4().hex[:14]}", event_type, service_id, source_service_id, target_service_id, detail, _now().isoformat()), ) def save_override(self, target_type: str, target_id: str, field_name: str, value: object, locked: bool = True) -> None: with self._connect() as db: db.execute( "INSERT INTO topology_overrides (target_type, target_id, field_name, value, locked, updated_at) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(target_type, target_id, field_name) DO UPDATE SET value=excluded.value, locked=excluded.locked, updated_at=excluded.updated_at", (target_type, target_id, field_name, _json(value), int(locked), _now().isoformat()), ) def has_override(self, target_type: str, target_id: str, field_name: str) -> bool: with self._connect() as db: row = db.execute( "SELECT 1 FROM topology_overrides WHERE target_type=? AND target_id=? AND field_name=? AND locked=1", (target_type, target_id, field_name), ).fetchone() return row is not None def list_overrides(self, target_type: str | None = None, target_id: str | None = None) -> list[dict[str, object]]: query = "SELECT target_type, target_id, field_name, value, locked, updated_at FROM topology_overrides WHERE 1=1" params: list[Any] = [] if target_type: query += " AND target_type=?" params.append(target_type) if target_id: query += " AND target_id=?" params.append(target_id) with self._connect() as db: rows = db.execute(query, params).fetchall() return [ { "target_type": row["target_type"], "target_id": row["target_id"], "field": row["field_name"], "value": _obj(row["value"]) if row["value"].startswith("{") else row["value"], "locked": bool(row["locked"]), "updated_at": row["updated_at"], } for row in rows ]