"""Public Byte Repository Intelligence service operations. These functions are JSON-shaped transport boundaries for the NDJSON host and MCP. They keep historical claim outcome separate from current applicability: changed evidence opens an obligation and requests rechecking; it never rewrites the old observation into a falsehood. """ from __future__ import annotations import hashlib import json import os import re import sqlite3 from dataclasses import asdict from datetime import datetime, timezone from pathlib import Path from typing import Mapping, Sequence from .knowledge_store import ( Claim, ClaimStatus, DependencyCheck, DependencyInput, EvidenceInput, ExactBlobDependency, KnowledgeEvent, KnowledgeStore, NegativeQueryDependency, PrecomputedQueryDependency, QueryEvaluation, QuerySetDependency, RecheckResult, Snapshot, _enforce_store_budget, _execute_sql_script, _insert_event_row, _store_write_transaction, _validated_store_database, _valid_typescript_boundary, query_input_fingerprint, query_typescript_boundary, ) from .probe_runner import get_probe, load_probe_policy, plan_probe, run_probe SERVICE_NAME = "Byte Repository Intelligence" SERVICE_VERSION = "1" OBLIGATION_SCHEMA_VERSION = 1 DEFAULT_CLAIM_LIMIT = 20 MAX_CLAIM_LIMIT = 200 _SOURCE_EXTENSIONS = ( "c", "cc", "cpp", "cs", "go", "h", "hpp", "java", "js", "jsx", "kt", "php", "py", "rb", "rs", "swift", "ts", "tsx", ) SOURCE_DOMAINS = tuple( pattern for extension in _SOURCE_EXTENSIONS for pattern in (f"*.{extension}", f"**/*.{extension}") ) # JSON participates through resolveJsonModule; declaration sidecars already end # in a TypeScript extension. Package, lock, and tsconfig inputs are fingerprinted # separately by query_input_fingerprint. _TYPESCRIPT_EXTENSIONS = ("cts", "js", "json", "jsx", "mts", "ts", "tsx") TYPESCRIPT_DOMAINS = tuple( pattern for extension in _TYPESCRIPT_EXTENSIONS for pattern in (f"*.{extension}", f"**/*.{extension}") ) def snapshot_capture( root: str | os.PathLike[str], source: str = "working_tree" ) -> dict[str, object]: """Capture an immutable source/config/lock manifest for one workspace.""" if source != "working_tree": raise ValueError("only working_tree snapshot capture is currently supported") with KnowledgeStore(root) as store: snapshot = store.capture_snapshot() event = store.append_event( "snapshot.captured", {"source": source, "files": len(snapshot.files), "service_version": SERVICE_VERSION}, snapshot_id=snapshot.id, ) return {**_snapshot_dict(snapshot), "source": source, "event_cursor": event.cursor} def claim_create( root: str | os.PathLike[str], snapshot_id: str, statement: str, *, status: str = "unresolved", scope: Mapping[str, object] | None = None, evidence: Sequence[Mapping[str, object]] = (), dependencies: Sequence[Mapping[str, object]] = (), ) -> dict[str, object]: """Create a typed claim from JSON-shaped evidence and dependency inputs.""" evidence_inputs = tuple(_evidence_input(item) for item in evidence) dependency_inputs = tuple(_dependency_input(item) for item in dependencies) with KnowledgeStore(root) as store: claim = store.create_claim( snapshot_id, statement, status=ClaimStatus(status), scope=dict(scope or {}), evidence=evidence_inputs, dependencies=dependency_inputs, ) return _claim_dict(claim) def typescript_boundary( root: str | os.PathLike[str], snapshot_id: str, identifiers: Sequence[str], max_results: int = 250, ) -> dict[str, object]: """Return compiler-resolved consumers from one immutable snapshot, never live files.""" with KnowledgeStore(root) as store: return query_typescript_boundary(store, snapshot_id, identifiers, max_results=max_results) def knowledge_record_answer( root: str | os.PathLike[str], snapshot_id: str, question: str, answer: str, evidence: Sequence[Mapping[str, object]], *, boundary: Mapping[str, object] | None = None, ) -> dict[str, object]: """Record an agent answer as an unresolved interpretation with a recheck boundary.""" spans = [_span(item) for item in evidence] if not spans and not boundary: raise ValueError("a repository answer requires source evidence or a bounded query") paths = sorted({item["path"] for item in spans}) evidence_inputs: list[EvidenceInput] = [ EvidenceInput( "model_interpretation", { "question": str(question), "answer": str(answer), "epistemic_status": "unverified_interpretation", }, ), EvidenceInput("source_spans", {"spans": spans}), ] dependencies: list[DependencyInput] = [ExactBlobDependency(path) for path in paths] boundary_dependency_start = len(dependencies) compiler_boundary: dict[str, object] | None = None scope: dict[str, object] = { "question": str(question), "source_paths": paths, "unresolved_boundaries": [ "The model's interpretation has not been independently established by an oracle." ], } if boundary: normalized_boundary = _json_object(boundary) compiler_boundary = normalized_boundary evidence_inputs.append(EvidenceInput("compiler_query_set", normalized_boundary)) query = normalized_boundary.get("query") identifiers = query.get("identifiers") if isinstance(query, dict) else None consumers = normalized_boundary.get("consumers") complete = normalized_boundary.get("complete") is True if isinstance(identifiers, list): for identifier in identifiers: if not isinstance(identifier, str) or not identifier: continue dependency = { "query_kind": "text_contains", "parameters": {"text": identifier, "case_sensitive": True}, "domain": TYPESCRIPT_DOMAINS, } if complete and isinstance(consumers, list) and not consumers: dependencies.append(NegativeQueryDependency(**dependency)) else: dependencies.append(QuerySetDependency(**dependency)) domain = normalized_boundary.get("domain") config_files = domain.get("configFiles") if isinstance(domain, dict) else None if isinstance(config_files, list): for config_file in config_files: if isinstance(config_file, str) and config_file: dependencies.append(ExactBlobDependency(config_file)) scope["compiler_boundary"] = { "adapter": normalized_boundary.get("adapter"), "query_fingerprint": normalized_boundary.get("queryFingerprint"), "result_fingerprint": normalized_boundary.get("resultFingerprint"), "complete_within_declared_domain": complete, "unknowns": normalized_boundary.get("unknowns", []), } boundary_unknowns = normalized_boundary.get("unknowns", []) if isinstance(boundary_unknowns, list): scope["unresolved_boundaries"] = [ *scope["unresolved_boundaries"], *(str(item) for item in boundary_unknowns), ] with KnowledgeStore(root) as store: snapshot = store.get_snapshot(snapshot_id) for span in spans: content = store.read_snapshot_file(snapshot.id, str(span["path"])) if content is None: raise ValueError(f"source evidence is absent from the snapshot: {span['path']}") try: line_count = len(content.decode("utf-8").splitlines()) except UnicodeDecodeError as error: raise ValueError(f"source evidence is not UTF-8 text: {span['path']}") from error if int(span["end"]) > line_count: raise ValueError( f"source evidence range exceeds {span['path']}: {span['end']} > {line_count}" ) if compiler_boundary is not None: boundary_authenticated = _authenticate_compiler_boundary( store, snapshot, compiler_boundary ) compiler_dependency, boundary_valid = _compiler_dependency( snapshot, compiler_boundary, authenticated=boundary_authenticated, ) if not boundary_valid: del dependencies[boundary_dependency_start:] compiler_scope = scope.get("compiler_boundary") if isinstance(compiler_scope, dict): compiler_scope["complete_within_declared_domain"] = False unresolved = scope.get("unresolved_boundaries") if isinstance(unresolved, list): unresolved.append( "The submitted compiler boundary failed Byte protocol validation or did not match a fresh analysis of the immutable snapshot." ) dependencies.append(compiler_dependency) known_paths = {item.path for item in dependencies if isinstance(item, ExactBlobDependency)} for config_path, _digest in (*snapshot.manifest.config_hashes, *snapshot.manifest.lock_hashes): if config_path not in known_paths: dependencies.append(ExactBlobDependency(config_path)) claim = store.create_claim( snapshot_id, str(answer), status=ClaimStatus.UNRESOLVED, scope=scope, evidence=evidence_inputs, dependencies=dependencies, ) return _claim_dict(claim) def claim_check( root: str | os.PathLike[str], claim_ids: Sequence[str], target_snapshot_id: str | None = None, ) -> dict[str, object]: """Recheck named claims against one immutable target snapshot.""" with KnowledgeStore(root) as store: snapshot = store.get_snapshot(target_snapshot_id) if target_snapshot_id else store.capture_snapshot() checks = [store.recheck_claim(str(claim_id), snapshot.id) for claim_id in claim_ids] return { "snapshot_id": snapshot.id, "checks": [_recheck_dict(item) for item in checks], "event_cursor": _event_cursor(Path(root).expanduser().resolve()), } def knowledge_query( root: str | os.PathLike[str], question: str, *, snapshot_id: str | None = None, scope: Mapping[str, object] | None = None, budget: Mapping[str, object] | None = None, ) -> dict[str, object]: """Find related claims and recheck their applicability; never executes code.""" budget_data = dict(budget or {}) limit = min(MAX_CLAIM_LIMIT, max(1, int(budget_data.get("max_claims", DEFAULT_CLAIM_LIMIT)))) with KnowledgeStore(root) as store: snapshot = store.get_snapshot(snapshot_id) if snapshot_id else store.capture_snapshot() matches = store.find_claims(str(question), limit=limit) checks = {item.id: store.recheck_claim(item.id, snapshot.id) for item in matches} refreshed = [store.get_claim(item.id) for item in matches] claims = [] for claim in refreshed: claim_data = _claim_dict(claim) claim_data["recheck"] = _recheck_dict(checks[claim.id]) claims.append(claim_data) return { "service": SERVICE_NAME, "snapshot_id": snapshot.id, "question": str(question), "scope": dict(scope or {}), "claims": claims, "execution_performed": False, "event_cursor": _event_cursor(Path(root).expanduser().resolve()), } def change_analyze( root: str | os.PathLike[str], base_snapshot_id: str, target_snapshot_id: str, intent_ref: str = "", ) -> dict[str, object]: """Recheck base claims and persist narrowly scoped change obligations.""" opened: list[dict[str, object]] = [] claims_checked = 0 with KnowledgeStore(root) as store: store.get_snapshot(base_snapshot_id) store.get_snapshot(target_snapshot_id) for claim in store.iter_snapshot_claims(base_snapshot_id): check = store.recheck_claim(claim.id, target_snapshot_id) claims_checked += 1 affected = [ dependency for dependency in check.dependencies if dependency.state in {"changed", "unknown"} ] if affected: dependency = DependencyCheck( ",".join(sorted(item.dependency_id for item in affected)), affected[0].type, "changed" if any(item.state == "changed" for item in affected) else "unknown", "; ".join(dict.fromkeys(item.reason for item in affected)), ) obligation, _was_created = _open_obligation( Path(root).expanduser().resolve(), claim, dependency, base_snapshot_id, target_snapshot_id, str(intent_ref), ) if obligation["status"] == "open": opened.append(obligation) return { "base_snapshot_id": base_snapshot_id, "target_snapshot_id": target_snapshot_id, "intent_ref": str(intent_ref), "claims_checked": claims_checked, "obligations": opened, "event_cursor": _event_cursor(Path(root).expanduser().resolve()), } def obligation_list( root: str | os.PathLike[str], status: str = "open", limit: int = 100 ) -> tuple[dict[str, object], bool]: """List durable change obligations without converting them into edit requirements.""" connection = _obligation_connection(Path(root).expanduser().resolve()) try: cap = min(1_000, max(0, int(limit))) rows = connection.execute( "SELECT * FROM obligations WHERE status = ? ORDER BY created_at, id LIMIT ?", (str(status), cap), ).fetchall() return {"items": [_obligation_from_row(row) for row in rows]} finally: connection.close() def obligation_resolve( root: str | os.PathLike[str], obligation_id: str, resolution: str, rationale: str ) -> dict[str, object]: """Resolve or accept an obligation with an explicit rationale.""" if resolution not in {"accepted", "fixed", "not_applicable", "verified"}: raise ValueError("unsupported obligation resolution") if not str(rationale).strip(): raise ValueError("obligation resolution requires a rationale") root_path = Path(root).expanduser().resolve() connection = _obligation_connection(root_path) try: resolved_at = _now() with _store_write_transaction(connection): changed = connection.execute( """ UPDATE obligations SET status = 'resolved', resolution = ?, rationale = ?, resolved_at = ? WHERE id = ? AND status = 'open' """, (resolution, str(rationale), resolved_at, str(obligation_id)), ).rowcount if not changed: raise KeyError(f"unknown or already resolved obligation: {obligation_id}") row = connection.execute( "SELECT * FROM obligations WHERE id = ?", (obligation_id,) ).fetchone() item = _obligation_from_row(row) _insert_event_row( connection, "obligation.resolved", { "obligation_id": obligation_id, "resolution": resolution, "rationale": str(rationale), }, claim_id=str(item["claim_id"]), snapshot_id=str(item["target_snapshot_id"]), created_at=resolved_at, ) finally: connection.close() return item def knowledge_events( root: str | os.PathLike[str], after_cursor: int = 0, limit: int = 100 ) -> dict[str, object]: """Read the monotonic invalidation/obligation/probe event stream.""" with KnowledgeStore(root) as store: events = store.read_events(after_cursor=after_cursor, limit=limit) return { "items": [_event_dict(item) for item in events], "event_cursor": events[-1].cursor if events else max(0, int(after_cursor)), } def probe_plan( root: str | os.PathLike[str], snapshot_id: str, hypothesis: str, command: Sequence[str], oracle_ref: Mapping[str, object], *, policy_id: str, claim_id: str | None = None, working_directory: str = ".", fixture: Mapping[str, object] | None = None, intervention: Mapping[str, object] | None = None, ) -> dict[str, object]: """Persist a reproducible probe; planning never executes it.""" plan = plan_probe( root, snapshot_id, hypothesis, command, oracle_ref, policy_id=policy_id, claim_id=claim_id, working_directory=working_directory, fixture=fixture, intervention=intervention, ) return asdict(plan) def probe_run( root: str | os.PathLike[str], probe_id: str, policy_id: str | None = None, ) -> dict[str, object]: """Execute a probe under a user-configured policy, never a model-supplied policy body.""" plan = get_probe(root, probe_id) if policy_id is not None and str(policy_id) != plan.policy_id: raise ValueError(f"probe requires policy {plan.policy_id!r}, not {policy_id!r}") resolved = load_probe_policy(plan.policy_id) return asdict(run_probe(root, probe_id, resolved)) def probe_get(root: str | os.PathLike[str], probe_id: str) -> dict[str, object]: return asdict(get_probe(root, probe_id)) def _snapshot_dict(snapshot: Snapshot) -> dict[str, object]: return { "snapshot_id": snapshot.id, "root": snapshot.root, "created_at": snapshot.created_at, "files": len(snapshot.files), "analyzer_version": snapshot.manifest.analyzer_version, "config_hashes": dict(snapshot.manifest.config_hashes), "lock_hashes": dict(snapshot.manifest.lock_hashes), } def _claim_dict(claim: Claim) -> dict[str, object]: return { "id": claim.id, "statement": claim.statement, "status": claim.status.value, "applicability": claim.applicability.value, "created_snapshot_id": claim.created_snapshot_id, "current_snapshot_id": claim.current_snapshot_id, "scope": dict(claim.scope), "created_at": claim.created_at, "updated_at": claim.updated_at, "evidence": [ { "id": item.id, "snapshot_id": item.snapshot_id, "kind": item.kind, "payload": dict(item.payload), "digest": item.digest, } for item in claim.evidence ], "dependencies": [ { "id": item.id, "type": item.type.value, "path": item.path, "query_kind": item.query_kind, "parameters": dict(item.parameters), "domain": list(item.domain), "domain_fingerprint": item.domain_fingerprint, "result_fingerprint": item.result_fingerprint, "results": list(item.results), "complete": item.complete, } for item in claim.dependencies ], } def _recheck_dict(check: RecheckResult) -> dict[str, object]: return { "check_id": check.check_id, "claim_id": check.claim_id, "target_snapshot_id": check.target_snapshot_id, "status": check.status.value, "applicability": check.applicability.value, "dependencies": [ { "dependency_id": item.dependency_id, "type": item.type.value, "state": item.state, "reason": item.reason, } for item in check.dependencies ], } def _event_dict(event: KnowledgeEvent) -> dict[str, object]: return { "cursor": event.cursor, "kind": event.kind, "claim_id": event.claim_id, "snapshot_id": event.snapshot_id, "payload": dict(event.payload), "created_at": event.created_at, } def _evidence_input(item: Mapping[str, object]) -> EvidenceInput: kind = item.get("kind") payload = item.get("payload", {}) if not isinstance(kind, str) or not kind: raise ValueError("evidence.kind is required") if not isinstance(payload, Mapping): raise ValueError("evidence.payload must be an object") return EvidenceInput(kind, dict(payload)) def _dependency_input(item: Mapping[str, object]) -> DependencyInput: dependency_type = item.get("type") if dependency_type == "exact_blob": return ExactBlobDependency(str(item.get("path", ""))) query_kind = item.get("query_kind") parameters = item.get("parameters", {}) domain = item.get("domain", ("**/*",)) if not isinstance(query_kind, str) or not query_kind: raise ValueError("query dependency requires query_kind") if not isinstance(parameters, Mapping) or not isinstance(domain, (list, tuple)): raise ValueError("query dependency parameters/domain are malformed") arguments = { "query_kind": query_kind, "parameters": dict(parameters), "domain": tuple(str(pattern) for pattern in domain), } if dependency_type == "query_set": return QuerySetDependency(**arguments) if dependency_type == "negative_query": return NegativeQueryDependency(**arguments) raise ValueError(f"unsupported dependency type: {dependency_type!r}") def _span(item: Mapping[str, object]) -> dict[str, object]: path = item.get("path") start = item.get("start") end = item.get("end") if not isinstance(path, str) or not path or not isinstance(start, int) or not isinstance(end, int): raise ValueError("source span requires path/start/end") if start < 1 or end < start: raise ValueError("source span line range is invalid") return {"path": path, "start": start, "end": end} def _json_object(value: Mapping[str, object]) -> dict[str, object]: encoded = json.dumps(dict(value), ensure_ascii=False, sort_keys=True) decoded = json.loads(encoded) if not isinstance(decoded, dict): raise ValueError("expected a JSON object") return decoded def _authenticate_compiler_boundary( store: KnowledgeStore, snapshot: Snapshot, boundary: Mapping[str, object], ) -> bool: """Bind a caller-supplied result to the packaged worker's own snapshot analysis.""" query = boundary.get("query") identifiers = query.get("identifiers") if isinstance(query, dict) else None consumers = boundary.get("consumers") if ( not isinstance(identifiers, list) or not identifiers or not isinstance(consumers, list) or not _valid_typescript_boundary(boundary, identifiers, snapshot, 1_000) ): return False cap = len(consumers) if boundary.get("truncated") is True else 1_000 try: observed = query_typescript_boundary( store, snapshot.id, identifiers, max_results=max(1, min(1_000, cap)), ) except (OSError, RuntimeError, ValueError): return False if observed.get("available") is not True or observed.get("snapshot_id") != snapshot.id: return False transport_fields = {"available", "snapshot_id"} authenticated = { key: value for key, value in observed.items() if key not in transport_fields } try: options = { "ensure_ascii": False, "sort_keys": True, "separators": (",", ":"), "allow_nan": False, } return json.dumps(authenticated, **options) == json.dumps(dict(boundary), **options) except (TypeError, ValueError): return False def _compiler_dependency( snapshot: Snapshot, boundary: Mapping[str, object], *, authenticated: bool, ) -> tuple[PrecomputedQueryDependency, bool]: query = boundary.get("query") identifiers = query.get("identifiers") if isinstance(query, dict) else None domain = boundary.get("domain") consumers = boundary.get("consumers") membership = domain.get("membershipFingerprint") if isinstance(domain, dict) else None resolution = domain.get("resolutionFingerprint") if isinstance(domain, dict) else None result_fingerprint = boundary.get("resultFingerprint") adapter_version = boundary.get("adapterVersion") identifiers_valid = ( isinstance(identifiers, list) and bool(identifiers) and all( isinstance(item, str) and re.fullmatch(r"[A-Za-z_$][A-Za-z0-9_$]*", item) is not None for item in identifiers ) and identifiers == sorted(set(identifiers)) ) valid = authenticated and identifiers_valid and _valid_typescript_boundary( boundary, list(identifiers), snapshot, 1_000 ) encoded_results: list[str] = [] if valid: for consumer in consumers: if not isinstance(consumer, dict): valid = False break required = [consumer.get(key) for key in ("path", "line", "column", "kind", "name")] if any(value is None for value in required): valid = False break encoded_results.append("\0".join(str(value) for value in required)) complete = valid and boundary.get("complete") is True and boundary.get("truncated") is not True domain_fingerprint = _fingerprint([str(membership), str(resolution)]) if valid else "" result_value = str(result_fingerprint) if valid else "" input_fingerprint = query_input_fingerprint(snapshot, TYPESCRIPT_DOMAINS) parameters: dict[str, object] = { "identifiers": list(identifiers) if isinstance(identifiers, list) else [], "adapter_version": str(adapter_version) if isinstance(adapter_version, str) else "", "baseline_input_fingerprint": input_fingerprint, "baseline_domain_fingerprint": domain_fingerprint, "baseline_result_fingerprint": result_value, "baseline_results": sorted(encoded_results), "baseline_complete": complete, } return ( PrecomputedQueryDependency( "byte.typescript.consumers", parameters, QueryEvaluation(domain_fingerprint, result_value, tuple(sorted(encoded_results)), complete), TYPESCRIPT_DOMAINS, negative=complete and not encoded_results, ), valid, ) def _fingerprint(items: Sequence[str]) -> str: payload = json.dumps(list(items), ensure_ascii=True, sort_keys=True, separators=(",", ":")) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def _open_obligation( root: Path, claim: Claim, dependency: DependencyCheck, base_snapshot_id: str, target_snapshot_id: str, intent_ref: str, ) -> tuple[dict[str, object], bool]: obligation_id = hashlib.sha256( "\0".join((claim.id, dependency.dependency_id, target_snapshot_id)).encode("utf-8") ).hexdigest() kind = "recheck_changed_dependency" if dependency.state == "changed" else "resolve_unknown_boundary" now = _now() connection = _obligation_connection(root) try: with _store_write_transaction(connection): cursor = connection.execute( """ INSERT OR IGNORE INTO obligations( id, claim_id, dependency_id, base_snapshot_id, target_snapshot_id, intent_ref, kind, reason, status, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'open', ?) """, ( obligation_id, claim.id, dependency.dependency_id, base_snapshot_id, target_snapshot_id, intent_ref, kind, dependency.reason, now, ), ) row = connection.execute( "SELECT * FROM obligations WHERE id = ?", (obligation_id,) ).fetchone() was_created = cursor.rowcount == 1 obligation = _obligation_from_row(row) if was_created: _insert_event_row( connection, "obligation.opened", obligation, claim_id=claim.id, snapshot_id=target_snapshot_id, created_at=now, ) finally: connection.close() return obligation, was_created def _obligation_connection(root: Path) -> sqlite3.Connection: try: database = _validated_store_database( root, create_store_directory=False, require_database=True ) except FileNotFoundError: raise FileNotFoundError("capture a repository snapshot before using obligations") connection = sqlite3.connect(database, isolation_level=None) try: _validated_store_database(root, create_store_directory=False, require_database=True) connection.row_factory = sqlite3.Row connection.execute("PRAGMA foreign_keys = ON") connection.execute("PRAGMA busy_timeout = 5000") _enforce_store_budget(connection, database) connection.execute("PRAGMA journal_mode = WAL") _enforce_store_budget(connection, database) _validated_store_database(root, create_store_directory=False, require_database=True) with _store_write_transaction(connection, database): _execute_sql_script( connection, """ CREATE TABLE IF NOT EXISTS obligation_schema_info(version INTEGER NOT NULL); INSERT INTO obligation_schema_info(version) SELECT 1 WHERE NOT EXISTS (SELECT 1 FROM obligation_schema_info); CREATE TABLE IF NOT EXISTS obligations( id TEXT PRIMARY KEY, claim_id TEXT NOT NULL, dependency_id TEXT NOT NULL, base_snapshot_id TEXT NOT NULL, target_snapshot_id TEXT NOT NULL, intent_ref TEXT NOT NULL, kind TEXT NOT NULL, reason TEXT NOT NULL, status TEXT NOT NULL, resolution TEXT, rationale TEXT, created_at TEXT NOT NULL, resolved_at TEXT ); CREATE INDEX IF NOT EXISTS obligations_status_idx ON obligations(status, created_at); """, ) version = connection.execute("SELECT version FROM obligation_schema_info").fetchone() if version is None or int(version["version"]) != OBLIGATION_SCHEMA_VERSION: found = "missing" if version is None else version["version"] raise RuntimeError(f"unsupported obligation schema {found}; expected {OBLIGATION_SCHEMA_VERSION}") except BaseException: connection.close() raise return connection def _event_cursor(root: Path) -> int: try: database = _validated_store_database( root, create_store_directory=False, require_database=True ) except FileNotFoundError: return 0 connection = sqlite3.connect(database) try: _validated_store_database(root, create_store_directory=False, require_database=True) row = connection.execute("SELECT COALESCE(MAX(cursor), 0) FROM events").fetchone() return int(row[0]) if row else 0 finally: connection.close() def _obligation_from_row(row: sqlite3.Row | None) -> dict[str, object]: if row is None: raise RuntimeError("obligation insert did not produce a row") return { "id": row["id"], "claim_id": row["claim_id"], "dependency_id": row["dependency_id"], "base_snapshot_id": row["base_snapshot_id"], "target_snapshot_id": row["target_snapshot_id"], "intent_ref": row["intent_ref"], "kind": row["kind"], "reason": row["reason"], "status": row["status"], "resolution": row["resolution"], "rationale": row["rationale"], "created_at": row["created_at"], "resolved_at": row["resolved_at"], } def _now() -> str: return datetime.now(timezone.utc).isoformat()