"""SQLite data model — the shared store the whole system records into. openai4s exposes these tables read-only through `host.query`; the host writes them as turns/cells/artifacts/compactions happen. Schema and write paths: frames turn tree (self-referential), per-turn model/effort/token/cost action_groups canonical provider/action groups, ordered per branch action_events append-only proposed/result/lifecycle events within a group execution_attempts attempt-first code execution lifecycle records execution_log per-cell record (code + usage wall/cpu/rss + error) artifacts logical artifact (filename, content_type) artifact_versions versioned bytes (version_id, checksum) -> artifacts compaction_archives compacted history slices agents agent profile definitions custom_skills user-authored SKILL.md bodies skill_* immutable Skill blobs/versions and activation history capability_* scoped enablement events/state + bootstrap manifests memories memory blocks (scope/block-listed in host.query) managed_endpoints local model endpoints notes project notes lineage_edges object-level data lineage: input_version -> output_version host_call_log RPC audit (DERIVABLE_HOST_CALLS are NOT logged; the args of SECRET_ARG_HOST_CALLS are redacted before write) Agent SQL (`host.query`) runs under a real SQLite authorizer installed for the duration of each statement, not behind a substring filter on the statement text. Secret-bearing tables (`settings` holds the LLM API key + model profiles, `connectors` holds MCP server env/command) and the internal audit/memory tables are refused outright; the SQLite catalog and the `pragma_*` table-valued functions are refused by prefix. The artifact family -- `artifacts`, `artifact_versions`, `lineage_edges`, `env_snapshots` -- is reachable only through the session-scoped `my_*` views, because a bundled Skill has a real reason to read `artifact_versions.source` and no reason to read another project's. `frames` stays directly readable: cross-session frame enumeration is a real leak, but closing it is a separate decision (see `_VIEW_ONLY_TABLES`). An authorizer sees resolved table names, so identifier quoting, a schema qualifier, an alias, a CTE and a name arriving in a bound parameter are all the same thing to it; a text filter had to enumerate spellings and could not see the last of those at all. All timestamps are epoch-ms. Booleans are 0/1. One DB per data_dir. """ from __future__ import annotations import json import re import sqlite3 import sys import threading import time from pathlib import Path from typing import Any, Callable, Mapping from openai4s.capabilities import CapabilityStateService, SpecialistProfileService from openai4s.execution.dependencies import ( analyze_code, default_replay_policy, default_visibility, ) from openai4s.security.permissions import harden_db, harden_dir from openai4s.security.secret_broker import is_ref from openai4s.storage.actions import ActionLedgerRepository from openai4s.storage.activation import SessionActivationRepository from openai4s.storage.agents import AgentProfileRepository from openai4s.storage.annotations import AnnotationRepository from openai4s.storage.artifact_observations import ( create_artifact_observations_schema, ) from openai4s.storage.artifacts import ArtifactRepository from openai4s.storage.artifacts import file_identity as _file_identity from openai4s.storage.artifacts import same_file_path as _same_file_path from openai4s.storage.auto_mode import ( AutoModeRepository, create_auto_mode_budget_schema, create_auto_mode_schema, install_auto_mode_action_guards, ) from openai4s.storage.branch_projection import count_cursor, project_branch_records from openai4s.storage.capabilities import CapabilityStateRepository from openai4s.storage.checkpoint_state import CheckpointStateRepository from openai4s.storage.compute_jobs import ComputeJobRepository from openai4s.storage.connectors import ( ConnectorRepository, broker_connector_env, forget_connector_env, resolve_connector_env, ) from openai4s.storage.datapro_index import ( DataProIndexRepository, create_datapro_index_schema, ) from openai4s.storage.delegation import DelegationProjectionRepository from openai4s.storage.delegation_attempts import create_delegation_request_schema from openai4s.storage.delivery import ( CompletionDeliveryRepository, create_completion_delivery_schema, ) from openai4s.storage.frames import FrameRepository, visible_session_clause from openai4s.storage.governance import ( GovernanceRepository, create_governance_schema, ) from openai4s.storage.kernels import KernelGenerationRepository from openai4s.storage.leases import LeaseRepository, create_lease_schema from openai4s.storage.memories import MemoryRepository from openai4s.storage.metadata import ( DERIVABLE_HOST_CALLS, SECRET_ARG_HOST_CALLS, CompactionRepository, EndpointRepository, FolderRepository, HostCallRepository, NotesRepository, ) from openai4s.storage.migrations import ( SCHEMA_VERSION, MigrationError, _is_duplicate_column, applied_migrations, current_version, preflight_schema, require_supported_schema, run_migrations, ) from openai4s.storage.model_capability_receipts import ( ModelCapabilityReceiptRepository, create_model_capability_receipts_schema, ) from openai4s.storage.permissions import ( DEFAULT_PERMISSION_RULES as _DEFAULT_PERMISSION_RULES, ) from openai4s.storage.permissions import PermissionRuleRepository from openai4s.storage.permissions import perm_match as _perm_match from openai4s.storage.plans import PlanRepository from openai4s.storage.recovery import RecoveryJournalRepository from openai4s.storage.session_imports import SessionImportRepository from openai4s.storage.settings import SettingsRepository from openai4s.storage.shares import SharesRepository from openai4s.storage.skills import SkillVersionRepository from openai4s.storage.snapshots import ( SessionSnapshotRepository, WorkspaceCAS, revert_recovery_setting_key, ) from openai4s.storage.team import ( TeamRepository, create_session_owners_schema, create_team_schema, ) from openai4s.storage.user_keys import UserKeyRepository, create_user_key_schema from openai4s.storage.workloads import WorkloadRepository, create_workload_schema _SCHEMA = """ CREATE TABLE IF NOT EXISTS frames ( frame_id TEXT PRIMARY KEY, parent_id TEXT, project_id TEXT NOT NULL DEFAULT 'default', root_frame_id TEXT, kind TEXT, -- 'turn' | 'delegate' | 'compaction_fork' name TEXT, task_summary TEXT, -- auto one-line summary shown in the UI model TEXT, effort TEXT, status TEXT, -- 'processing'|'done'|'failed'|'awaiting_user_response' runtime_env TEXT, depth INTEGER NOT NULL DEFAULT 0, input_tokens INTEGER, output_tokens INTEGER, cost_usd REAL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS ix_frames_parent ON frames(parent_id); CREATE INDEX IF NOT EXISTS ix_frames_project ON frames(project_id); CREATE TABLE IF NOT EXISTS projects ( project_id TEXT PRIMARY KEY, name TEXT, description TEXT, context TEXT, -- agent context prepended to prompts is_example INTEGER NOT NULL DEFAULT 0, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS messages ( message_id TEXT PRIMARY KEY, root_frame_id TEXT NOT NULL, branch_id TEXT, frame_id TEXT, seq INTEGER NOT NULL, role TEXT NOT NULL, -- 'user' | 'assistant' content TEXT, -- plain text (may be markdown) metadata TEXT, -- JSON blob (optional) created_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS ix_msg_root ON messages(root_frame_id); CREATE TABLE IF NOT EXISTS execution_log ( producing_cell_id TEXT PRIMARY KEY, frame_id TEXT, root_frame_id TEXT, project_id TEXT NOT NULL DEFAULT 'default', cell_seq INTEGER, cell_index INTEGER, state_revision INTEGER, kernel_id TEXT, language TEXT, status TEXT, origin TEXT, code TEXT NOT NULL, code_hash TEXT NOT NULL, visibility TEXT NOT NULL DEFAULT 'scientific' CHECK (visibility IN ('scientific','scratch','recovery','system')), pin INTEGER NOT NULL DEFAULT 0 CHECK (pin IN (0,1)), replay_policy TEXT NOT NULL DEFAULT 'conditional' CHECK (replay_policy IN ('safe','conditional','never')), variable_reads TEXT NOT NULL DEFAULT '[]', variable_writes TEXT NOT NULL DEFAULT '[]', variable_deletes TEXT NOT NULL DEFAULT '[]', mutation_uncertain INTEGER NOT NULL DEFAULT 0 CHECK (mutation_uncertain IN (0,1)), stdout TEXT, stderr TEXT, error TEXT, figures TEXT, -- JSON list of artifact filenames files_read TEXT, -- JSON list of relative paths files_written TEXT, -- JSON list of relative paths interrupted INTEGER NOT NULL DEFAULT 0, wall_s REAL, cpu_s REAL, peak_rss_kb INTEGER, created_at INTEGER NOT NULL, -- v28: the kernel generation that ran a directly-recorded Cell. Web -- cells derive theirs from execution_attempts; delegated children have -- no attempt row, so the log row itself may carry the binding. Last so -- fresh and ALTER-upgraded databases agree on column order. generation_id TEXT ); CREATE INDEX IF NOT EXISTS ix_exec_frame ON execution_log(frame_id); CREATE INDEX IF NOT EXISTS ix_exec_root ON execution_log(root_frame_id); CREATE TABLE IF NOT EXISTS artifacts ( artifact_id TEXT PRIMARY KEY, project_id TEXT NOT NULL DEFAULT 'default', root_frame_id TEXT, filename TEXT NOT NULL, content_type TEXT, is_user_upload INTEGER NOT NULL DEFAULT 0, priority INTEGER NOT NULL DEFAULT 0, latest_version_id TEXT, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); -- ix_artifacts_project_created is created by migration 32 only. This script is -- committed before the migration's backup is taken, so an object added here -- lands in an older database ahead of its pre-upgrade copy and survives a -- rolled-back upgrade. A new database gets the index from the same step. CREATE TABLE IF NOT EXISTS artifact_versions ( version_id TEXT PRIMARY KEY, artifact_id TEXT NOT NULL, filename TEXT, content_type TEXT, size_bytes INTEGER, checksum TEXT, path TEXT NOT NULL, snapshot_path TEXT, producing_cell_id TEXT, frame_id TEXT, created_at INTEGER NOT NULL, -- Where the data came from, when a version was derived from retrieved -- data rather than computed from nothing. JSON: the retrieval provenance -- envelope (database, source, retrieved_at, request_url, query, -- normalization_version, per-response hashes). This is a property of the -- VERSION, not the artifact: rerunning the same analysis a month later -- produces the same file from a different retrieval. source TEXT ); CREATE INDEX IF NOT EXISTS ix_ver_artifact ON artifact_versions(artifact_id); -- De-duplicated environment snapshots (one row per distinct kernel env). An -- artifact_version references one via env_snapshot_id so a figure records the -- environment that PRODUCED it -- taken from that cell's kernel generation, -- not from the daemon (see ArtifactManager.capture_environment). CREATE TABLE IF NOT EXISTS env_snapshots ( snapshot_id TEXT PRIMARY KEY, created_at INTEGER NOT NULL, kind TEXT, -- the RUNTIME: "python" | "r" python_version TEXT, -- NULL unless the kernel was this interpreter implementation TEXT, platform TEXT, package_count INTEGER, packages_json TEXT, remote_json TEXT, -- JSON list of remote-GPU job provenance -- Which kernel this actually describes. Without these two, an R kernel and -- a Python one in a conda env are indistinguishable rows, and the identity -- of the environment is exactly what provenance is for. interpreter TEXT, environment_name TEXT, -- The generation that produced the artifact, and why a package list may be -- absent (an R kernel has no Python distributions; a foreign interpreter -- may refuse to be read). Absence with a reason beats a borrowed list. generation_id TEXT, -- How far the generation above can be trusted. `verified` means the row's -- own address includes it, so it cannot have been shared by a second -- kernel; `legacy_unverified` means it was written before that was true -- -- the named generation did produce this environment, but it may not be the -- only one that did. Kept and labelled rather than cleared: a missing -- attribution is not more honest than a qualified one, it is just emptier. generation_confidence TEXT, packages_unavailable TEXT, -- Whether this row was measured from a kernel generation or assumed from -- the daemon. The fallback path has always set this and it was dropped at -- the INSERT, so the one marker separating a measured environment from a -- guessed one never reached the record a reader actually sees. provenance TEXT ); CREATE TABLE IF NOT EXISTS compaction_archives ( archive_id TEXT PRIMARY KEY, frame_id TEXT, project_id TEXT NOT NULL DEFAULT 'default', branch_id TEXT, ledger_cursor TEXT, recovery_pointer TEXT, generation_id TEXT, metadata TEXT, summary TEXT, handoff TEXT, compacted TEXT, -- JSON of the raw slice n_messages INTEGER, context_before TEXT, context_after TEXT, artifact_refs TEXT, created_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS agents ( name TEXT PRIMARY KEY, -- UPPER_SNAKE (2-32) description TEXT, skill_names TEXT, -- JSON list or NULL (=unrestricted) connectors TEXT, -- JSON list unrestricted INTEGER NOT NULL DEFAULT 1, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS custom_skills ( name TEXT PRIMARY KEY, origin TEXT, skill_md TEXT, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); -- Shared enablement policy for Skills, Specialists, and future capability -- kinds. ``capability_events`` is append-only; ``capability_states`` is its -- efficient current-state projection. Session state overrides project state, -- which overrides global state. CREATE TABLE IF NOT EXISTS capability_states ( kind TEXT NOT NULL, name TEXT NOT NULL, normalized_name TEXT NOT NULL, scope TEXT NOT NULL, -- global | project | session scope_id TEXT NOT NULL DEFAULT '', enabled INTEGER NOT NULL, metadata TEXT, -- JSON, non-secret manifest hints created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, PRIMARY KEY(kind, normalized_name, scope, scope_id) ); CREATE TABLE IF NOT EXISTS capability_events ( event_id TEXT PRIMARY KEY, kind TEXT NOT NULL, name TEXT NOT NULL, normalized_name TEXT NOT NULL, scope TEXT NOT NULL, scope_id TEXT NOT NULL DEFAULT '', event TEXT NOT NULL, -- enabled | disabled | sidecar_loaded enabled INTEGER, metadata TEXT, created_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS ix_capability_events_lookup ON capability_events(kind, normalized_name, created_at); CREATE TABLE IF NOT EXISTS capability_manifests ( manifest_id TEXT PRIMARY KEY, session_id TEXT NOT NULL, project_id TEXT, kind TEXT NOT NULL, entries TEXT NOT NULL, -- JSON snapshot, loaded=false initially created_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS ix_capability_manifest_session ON capability_manifests(session_id, kind, created_at); CREATE TABLE IF NOT EXISTS memories ( memory_id TEXT PRIMARY KEY, project_id TEXT NOT NULL DEFAULT 'default', block TEXT, -- memory block name content TEXT, created_at INTEGER NOT NULL, updated_at INTEGER -- last edit; NULL means never edited ); CREATE TABLE IF NOT EXISTS managed_endpoints ( name TEXT PRIMARY KEY, url TEXT, skill TEXT, port INTEGER, status TEXT, -- 'registered'|'starting'|'live'|'stopped' credential TEXT, start_script TEXT, stop_script TEXT, live_route TEXT, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS notes ( note_id TEXT PRIMARY KEY, project_id TEXT NOT NULL DEFAULT 'default', title TEXT, body TEXT, created_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS lineage_edges ( edge_id TEXT PRIMARY KEY, input_version_id TEXT NOT NULL, output_version_id TEXT NOT NULL, producing_cell_id TEXT, frame_id TEXT, created_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS ix_edge_out ON lineage_edges(output_version_id); CREATE INDEX IF NOT EXISTS ix_edge_in ON lineage_edges(input_version_id); CREATE TABLE IF NOT EXISTS host_call_log ( call_id TEXT PRIMARY KEY, frame_id TEXT, action_group_id TEXT, action_id TEXT, permission_decision_id TEXT, method TEXT NOT NULL, args_preview TEXT, result_preview TEXT, result_digest TEXT, side_effect_class TEXT, resource_keys TEXT, ok INTEGER NOT NULL DEFAULT 1, created_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT, updated_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS folders ( folder_id TEXT PRIMARY KEY, project_id TEXT NOT NULL DEFAULT 'default', name TEXT NOT NULL, created_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS connectors ( connector_id TEXT PRIMARY KEY, -- slug name TEXT NOT NULL, description TEXT, command TEXT NOT NULL, -- JSON list argv OR a shell string args TEXT, -- JSON list env TEXT, -- JSON dict enabled INTEGER NOT NULL DEFAULT 1, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); -- Remote compute jobs. These outlive the process that submitted them: an ssh -- job keeps running under nohup and a byoc sandbox keeps billing whether or not -- this daemon is alive. Holding them only in memory (which is what -- ComputeManager did) meant a restart stranded every in-flight job — the remote -- work continued with nothing left that could find, harvest, or cancel it. CREATE TABLE IF NOT EXISTS compute_jobs ( job_id TEXT PRIMARY KEY, -- Stable across a resubmit of the same logical work. Reconciliation looks -- a job up by this before submitting, so a crash between "provider -- accepted" and "we recorded it" cannot become a double-charge. idempotency_key TEXT, provider TEXT NOT NULL, -- "ssh:" | "byoc:" status TEXT NOT NULL, -- see compute/states.py; enforced on write alias TEXT, -- ssh workdir TEXT, -- ssh pid TEXT, -- ssh -- The remote process GROUP, read back from the host at submit rather than -- assumed from `$!`. Cancellation signals this; a NULL means we never -- confirmed one and must not guess (see migration 6). pgid TEXT, -- ssh sandbox_id TEXT, -- byoc -- The provider's own acknowledgement of the submit. Evidence the job -- exists remotely, independent of anything we chose to believe. receipt TEXT, outputs TEXT, -- JSON: declared output globs -- Exact Artifact versions staged into the remote job. This is separate -- from `outputs`: it survives daemon restart and is copied onto every -- harvested version's provenance record. input_versions TEXT, -- JSON: ordered version ids -- Which session/workspace submitted this job. `_rehydrate` filters on it so -- a restart does not hand one session's live jobs — and their harvested -- outputs — to whichever session happens to build a manager first. NULL is -- the CLI / global context (see migration 9). owner_key TEXT, exit_code INTEGER, reason TEXT, -- free-text detail, provider-supplied -- Coded cause, from compute/states.py. `failed` because outputs could not -- be verified is a different fact from `failed` because the command -- exited non-zero, and the status alone cannot carry that. termination_reason TEXT, -- What the harvest actually produced: JSON [{path,size,sha256}] and one -- digest over the whole record. A job that declared `outputs` and -- produced none of them used to report success; these are what make that -- checkable, and the only way to see a transfer truncated at rc==0. artifact_manifest TEXT, integrity_sha256 TEXT, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, submitted_at INTEGER, terminal_at INTEGER ); CREATE INDEX IF NOT EXISTS ix_compute_jobs_status ON compute_jobs(status); -- The per-owner idempotency index is deliberately NOT here. It references -- `owner_key`, which migration 9 adds -- and this script runs on every open, -- *before* migrations, so on a pre-9 database `CREATE INDEX` would fail on a -- column that does not exist yet and take startup with it. Migration 11 creates -- it (and drops the old installation-wide one) for new and upgraded databases -- alike, which is the one place both paths pass through. -- Append-only, monotonically sequenced per job. A status column alone says -- where a job is; this says how it got there, which is what a restart needs to -- tell "we never submitted" from "we submitted and lost the response". CREATE TABLE IF NOT EXISTS compute_job_events ( job_id TEXT NOT NULL, seq INTEGER NOT NULL, kind TEXT NOT NULL, at INTEGER NOT NULL, payload TEXT, -- JSON PRIMARY KEY (job_id, seq) ); CREATE TABLE IF NOT EXISTS frame_steps ( step_id TEXT PRIMARY KEY, frame_id TEXT NOT NULL, seq INTEGER NOT NULL, kind TEXT NOT NULL, -- search|plan|env|skill|bash|edit|write|read|files|artifact|delegate|mcp|fetch|code title TEXT, summary TEXT, -- one-line result summary (shown as meta) input TEXT, -- JSON output TEXT, -- JSON status TEXT, -- running|done|warning|error created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_frame_steps_frame ON frame_steps(frame_id, seq); CREATE TABLE IF NOT EXISTS annotations ( annotation_id TEXT PRIMARY KEY, root_frame_id TEXT NOT NULL, artifact_id TEXT NOT NULL, artifact_name TEXT, rel_x REAL NOT NULL, -- 0..1 fraction of image width rel_y REAL NOT NULL, -- 0..1 fraction of image height -- The artifact VERSION the pin was taken against, and the sha256 of that -- version's bytes. A pin means "this point on the picture I am looking at"; -- without these the send path resolved the artifact's latest version, so -- re-plotting between the pin and the send silently sent a different image -- under the old coordinates. NULL on rows created before this existed. version_id TEXT, checksum TEXT, number INTEGER NOT NULL, -- pin ordinal within (frame,artifact) body TEXT NOT NULL, -- the comment -- Which in-flight request holds this pin. Admission is exactly-once: a -- request claims `open` rows atomically into `reserved` under its own id, -- and only what it claimed is quoted into the prompt. NULL whenever the -- row is not held. reservation_id TEXT, status TEXT NOT NULL DEFAULT 'open', -- open|reserved|sent|resolved|dismissed created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, -- Stage 9 workbench locators. Image pins stay on rel_x/rel_y; PDF/HTML -- comments name a quote or element. NULL on rows created before this. kind TEXT, locator TEXT ); -- One row per attempt to admit pinned comments into a message. -- -- A 202 can be lost: a dropped connection, a closed tab, a reload. The client -- then knows only that it sent something, and without a durable record tying -- the reservation to the request, the job and the frame there is nothing to -- reconcile against -- leaving only "resend" (double work) or "give up" -- (silent loss of the user's comments). CREATE TABLE IF NOT EXISTS annotation_admissions ( reservation_id TEXT PRIMARY KEY, root_frame_id TEXT NOT NULL, annotation_ids TEXT NOT NULL, -- JSON array, the exact claimed set request_id TEXT, job_id TEXT, message_id TEXT, state TEXT NOT NULL, -- reserved|sent|pending|released created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS ix_admission_frame ON annotation_admissions(root_frame_id); CREATE INDEX IF NOT EXISTS ix_admission_state ON annotation_admissions(state); CREATE INDEX IF NOT EXISTS ix_annot_frame ON annotations(root_frame_id); CREATE INDEX IF NOT EXISTS ix_annot_artifact ON annotations(artifact_id); CREATE TABLE IF NOT EXISTS plans ( plan_id TEXT PRIMARY KEY, frame_id TEXT NOT NULL, project_id TEXT NOT NULL DEFAULT 'default', title TEXT, rationale TEXT, confidence TEXT, -- 'high'|'medium'|'low' (or a 0..1 string) steps TEXT NOT NULL, -- JSON [{id,title,detail,deliverables:[...]}] status TEXT NOT NULL DEFAULT 'draft', -- see storage/plans.py PLAN_STATUSES step_status TEXT, -- JSON {step_id: {status, note, updated_at}} artifact_id TEXT, -- the plan_*.json artifact (so revises re-version it) created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_plans_frame ON plans(frame_id, created_at); -- opencode-style tool-call permission rules. Each rule maps a (tool, pattern) -- to allow|ask|deny at one of three scopes: 'global' (scope_id=''), -- 'project' (scope_id=project_id) or 'conversation' (scope_id=root_frame_id). CREATE TABLE IF NOT EXISTS permission_rules ( rule_id TEXT PRIMARY KEY, scope TEXT NOT NULL, -- global | project | conversation scope_id TEXT NOT NULL DEFAULT '', -- '' for global; project_id; root_frame_id tool TEXT NOT NULL, -- host method name, or '*' pattern TEXT NOT NULL DEFAULT '*', -- glob matched against the tool target decision TEXT NOT NULL, -- allow | ask | deny created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); CREATE UNIQUE INDEX IF NOT EXISTS ux_perm ON permission_rules(scope, scope_id, tool, pattern); CREATE INDEX IF NOT EXISTS ix_perm_scope ON permission_rules(scope, scope_id); -- Durable approval requests. Unlike permission_rules (standing policy), each -- row is one concrete action decision and remains auditable across reconnects -- and daemon restarts. Terminal requests are immutable. CREATE TABLE IF NOT EXISTS permission_requests ( decision_id TEXT PRIMARY KEY, root_frame_id TEXT, frame_id TEXT, project_id TEXT, action_group_id TEXT, action_id TEXT, tool_call_id TEXT, tool TEXT NOT NULL, target TEXT NOT NULL DEFAULT '', side_effect_class TEXT, resource_keys TEXT, payload TEXT, dangerous INTEGER NOT NULL DEFAULT 0, canonical_arguments_sha256 TEXT, action_digest TEXT, state TEXT NOT NULL DEFAULT 'pending', scope TEXT, pattern TEXT, message TEXT, resolution_context TEXT, continuation_required INTEGER NOT NULL DEFAULT 0, continuation_expires_at INTEGER, continuation_consumed_at INTEGER, created_at INTEGER NOT NULL, expires_at INTEGER, resolved_at INTEGER ); CREATE INDEX IF NOT EXISTS ix_permission_request_root ON permission_requests(root_frame_id, state, created_at); CREATE TABLE IF NOT EXISTS shares ( share_id TEXT PRIMARY KEY, root_frame_id TEXT NOT NULL, title TEXT, status TEXT NOT NULL DEFAULT 'publishing' CHECK (status IN ('publishing','ready','failed','revoked')), snapshot_id TEXT, pending_snapshot_id TEXT, bundle_sha256 TEXT, bundle_size INTEGER, projection_id TEXT, counts_json TEXT, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, revoked_at INTEGER, expires_at INTEGER ); CREATE UNIQUE INDEX IF NOT EXISTS ux_shares_active_frame ON shares(root_frame_id) WHERE status IN ('publishing','ready'); CREATE INDEX IF NOT EXISTS ix_shares_root ON shares(root_frame_id); """ # Tables host.query must refuse to read. These hold secrets or # internal audit/memory state that is not part of the agent-visible data model: # settings -> LLM API key + model profiles (which embed API keys) # connectors -> MCP server env vars / launch command (may embed tokens) # memories -> memory blocks (surfaced through host.remember, not SQL) # host_call_log -> RPC audit trail # permission_rules -> permission broker state # action_* / execution_attempts -> provider wire state and raw action audit QUERY_DENYLIST = frozenset( { "settings", "connectors", "memories", # Same class as `connectors`: `managed_endpoints.credential` is a # plaintext bearer token for a local model endpoint, and it was # readable with one `SELECT credential FROM managed_endpoints` from # any agent cell -- in single-user installs as well as team ones. A # credential the API never returns must not be reachable through the # query surface that goes around the API. "managed_endpoints", # Team-mode identity (INV-9 hygiene): password hashes, login-token # hashes, and the governance audit trail are never agent-readable. "users", "auth_sessions", "team_audit_log", "session_owners", # Team-mode governance (M2): invites hold credential digests, and # membership/metering/limits are the operator's data, not the agent's. "project_members", "invites", "usage_ledger", "quotas", # Orchestration state: submission tokens are credentials of a sort # (they are what INV-8 reconciliation matches on), and a workload's # spec carries another user's command line. "workloads", "allocations", # Cluster sessions (M3b/M4). `user_llm_keys` holds a broker reference # and every user's id -- the two fields `UserKeyRecord.public()` # deliberately withholds from a route, so leaving the table readable # would hand the agent exactly what the API refuses. `leases` and # `session_workloads` map sessions to workloads and to each other, # which is the same "who is working on what" that INV-13 protects # everywhere else. "user_llm_keys", "leases", "session_workloads", "host_call_log", "permission_rules", "permission_requests", "action_groups", "action_events", "execution_attempts", "kernel_generations", "capability_states", "capability_events", "capability_manifests", # Exact probe receipts. Not agent-working-data; they name a profile # revision and an endpoint digest, and they gate native completion. "model_capability_receipts", "skill_blobs", "skill_versions", "skill_version_files", "skill_installations", "skill_installation_events", "delegation_sessions", "delegation_children", "delegation_steering", "delegation_requests", "delegation_attempts", "session_branches", "session_branch_selection", "session_checkpoints", "checkpoint_state_snapshots", "snapshot_operations", "recovery_journal", # Auto Mode contains immutable Reviewer/Guardian inputs, exact action # digests, raw audit payloads, and recovery ownership. It is projected # only through the bounded server service; agent SQL cannot inspect or # use it as an alternate permission channel. "auto_mode_selections", "auto_mode_runs", "auto_mode_events", "auto_mode_budget_state", "auto_mode_budget_reservations", "auto_mode_budget_events", "review_runs", "review_findings", "repair_runs", "repair_execution_groups", "permission_review_assessments", # Publication ledger binds exact manifests to final assistant messages. # It is server recovery/audit state, not an agent data source. "completion_deliveries", "completion_delivery_artifacts", # SQLite's own catalogue. The denylist above protects the *contents* of # these tables and this one handed back their entire definition: # `SELECT sql FROM sqlite_master WHERE name='permission_rules'` returned # the full DDL of a denied table, and `SELECT name FROM sqlite_master` # enumerated every one of them by name. `schema()` has always excluded # the `sqlite_` prefix; `query()` never did, so the one surface actually # exposed to the model was the one that leaked. "sqlite_master", "sqlite_schema", "sqlite_temp_master", "sqlite_temp_schema", } ) # Single-quoted string literals and SQL comments are stripped before the denylist # substring test so a denied table name that appears only inside a *literal* # (e.g. SELECT 'see settings' AS note) is not falsely rejected — a real table # reference can never live inside a string literal. Double-quoted / bracketed / # backtick spans are left intact because SQL uses them to quote identifiers # (e.g. FROM "settings"), which must still trip the denylist. _SQL_LITERAL_RE = re.compile( r"'(?:[^']|'')*'" # single-quoted string (with '' escape) r"|--[^\n]*" # line comment r"|/\*.*?\*/", # block comment re.DOTALL, ) def _strip_sql_literals(sql: str) -> str: """Blank out single-quoted string literals and comments for denylist checks.""" return _SQL_LITERAL_RE.sub(" ", sql or "") #: Word-boundary matcher for the denylist pre-check. A bare substring test #: (`if bad in text`) denies any query that merely *contains* a denied word — #: harmless for a specific name like `settings`, but a real single-user #: regression once the team tables add generic words: `SELECT * FROM #: active_users` contains `users`. `\b` treats a denied table as a token, so #: `\busers\b` matches `FROM users`, `"users"`, `main.users`, and `users,` #: (every real reference) but NOT `active_users` (`_` is a word char). This #: only makes the cheap pre-check less aggressive; the SQLite authorizer, #: which denies by the *resolved* table name, stays the real enforcement, so #: no denied table becomes readable. _DENY_WORD_RE = re.compile( r"\b(?:" + "|".join(re.escape(t) for t in sorted(QUERY_DENYLIST)) + r")\b" ) def _now_ms() -> int: return int(time.time() * 1000) # How long a writer waits for a competing lock before raising "database is # locked". Python's sqlite3 already defaults this to 5s via connect(timeout=); # naming it makes the value a decision rather than a coincidence, and gives the # multi-process case (openai4s run / init alongside a live daemon) one place to # tune. _BUSY_TIMEOUT_S = 5.0 def _sql_quote(value: str) -> str: """Single-quote a literal for interpolation into DDL.""" return "'" + value.replace("'", "''") + "'" #: The views `host.query` may read the artifact family through. Reads of the base #: tables are permitted only when SQLite reports one of these as the view #: responsible for the access -- which is what the authorizer's fifth argument is #: for. A direct `SELECT * FROM artifacts` names no view and is refused. _SCOPED_VIEWS = frozenset( { "my_artifacts", "my_artifact_versions", "my_artifact_capture_observations", "my_lineage_edges", "my_frames", "my_env_snapshots", # The conversation/execution family, scoped the same way (team mode). "my_messages", "my_execution_log", } ) #: A CTE binding one of `_SCOPED_VIEWS` by name. The authorizer's fifth #: argument names "the view responsible for this read", and SQLite fills it in #: for a *common table expression* exactly as it does for a view -- measured: #: `WITH my_messages AS (SELECT * FROM main.messages) SELECT content FROM #: my_messages` hands the authorizer `(20, 'messages', 'content', 'main', #: 'my_messages')`, which is byte-for-byte what the real temp view produces. #: So the one string the escape hatch trusts was a string the caller could #: mint, and in team mode that string was the entire tenant boundary for #: `host.query`: one CTE returned every colleague's prompts, replies, cell #: code and stdout. #: #: SQLite accepts identifiers quoted with double quotes, brackets, backticks #: and (for compatibility) single quotes. The denylist scanner deliberately #: removes single-quoted *literals*, so a regex over that scanner can never #: distinguish `SELECT 'my_messages'` from `WITH 'my_messages' AS (...)`. #: This small lexer keeps quote kind and punctuation, then parses only the #: `WITH name [(columns)] AS [NOT] [MATERIALIZED] (...)` binding shape. It is #: intentionally not a SQL parser; SQLite's authorizer remains the enforcement #: for every other name and operation. def _sql_cte_tokens(sql: str) -> list[tuple[str, str]]: """Tokenize just enough SQL to recognize quoted CTE bindings.""" tokens: list[tuple[str, str]] = [] index = 0 length = len(sql or "") while index < length: char = sql[index] if char.isspace(): index += 1 continue if sql.startswith("--", index): newline = sql.find("\n", index + 2) index = length if newline < 0 else newline + 1 continue if sql.startswith("/*", index): end = sql.find("*/", index + 2) index = length if end < 0 else end + 2 continue if char in "'\"`[": closing = "]" if char == "[" else char index += 1 value: list[str] = [] while index < length: current = sql[index] if current == closing: # SQLite doubles quote delimiters inside quoted names. if index + 1 < length and sql[index + 1] == closing: value.append(closing) index += 2 continue index += 1 break value.append(current) index += 1 tokens.append(("quoted", "".join(value))) continue if char in "(),": tokens.append(("punct", char)) index += 1 continue # SQLite's ALPHABETIC class includes every non-ASCII code point, not # only characters Python calls alphanumeric. Combining marks are a # practical example: ``e\u0301`` is one valid bare identifier. Treating # the mark as an unknown token let an attacker put that harmless CTE # first and hide a scoped-view shadow later in the same WITH list. if char.isalnum() or char in "_$" or ord(char) >= 0x80: start = index while index < length and ( sql[index].isalnum() or sql[index] in "_$" or ord(sql[index]) >= 0x80 ): index += 1 tokens.append(("word", sql[start:index])) continue tokens.append(("other", char)) index += 1 return tokens def _after_sql_parentheses(tokens: list[tuple[str, str]], index: int) -> int | None: """Return the token after one balanced parenthesized span.""" if index >= len(tokens) or tokens[index] != ("punct", "("): return None depth = 0 while index < len(tokens): token = tokens[index] if token == ("punct", "("): depth += 1 elif token == ("punct", ")"): depth -= 1 if depth == 0: return index + 1 index += 1 return None def _is_sql_word(token: tuple[str, str], word: str) -> bool: return token[0] == "word" and token[1].casefold() == word def _cte_shadow_name(sql: str) -> str | None: """Return a scoped-view name rebound by any CTE in *sql*.""" tokens = _sql_cte_tokens(sql) for start, token in enumerate(tokens): if not _is_sql_word(token, "with"): continue index = start + 1 if index < len(tokens) and _is_sql_word(tokens[index], "recursive"): index += 1 while index < len(tokens): kind, spelling = tokens[index] if kind not in {"word", "quoted"}: break name = spelling.casefold() index += 1 if index < len(tokens) and tokens[index] == ("punct", "("): after_columns = _after_sql_parentheses(tokens, index) if after_columns is None: break index = after_columns if index >= len(tokens) or not _is_sql_word(tokens[index], "as"): break index += 1 if index < len(tokens) and _is_sql_word(tokens[index], "not"): index += 1 if index >= len(tokens) or not _is_sql_word( tokens[index], "materialized" ): break index += 1 elif index < len(tokens) and _is_sql_word(tokens[index], "materialized"): index += 1 if index >= len(tokens) or tokens[index] != ("punct", "("): break if name in _SCOPED_VIEWS: return name after_body = _after_sql_parentheses(tokens, index) if after_body is None: break index = after_body if index >= len(tokens) or tokens[index] != ("punct", ","): break index += 1 return None #: Base tables reachable only through `_SCOPED_VIEWS`. These were readable #: directly and were not on `QUERY_DENYLIST` at all, so one `SELECT` returned #: every project's artifacts with their filenames, checksums and absolute #: snapshot paths -- the exact information the scoped host helpers refuse one #: version id at a time. _VIEW_ONLY_TABLES = frozenset( { "artifacts", "artifact_versions", "artifact_capture_observations", "lineage_edges", # Interpreter, prefix and the complete installed-package manifest of # every kernel generation in the database. "env_snapshots", "datapro_index_batches", "datapro_index_entries", # `frames` is deliberately NOT here. Cross-session frame enumeration is a # real leak, but the plan does not list it and `tests/test_store.py` # documents direct `SELECT * FROM frames` as allowed -- so closing it is a # separate decision with its own migration for anything reading it, not a # side effect of this change. `my_frames` exists for callers that want the # scoped form. } ) #: The same rule, applied to the conversation and execution family, and only #: when team mode is on. #: #: In a single-user install "every session in this database" is the user's own #: work, and `tests/test_store.py` documents reading it. In team mode it is #: every colleague's prompts, the model's replies, and the code they ran -- #: which is INV-13 on the most sensitive surface the product has, reachable #: from a single agent `SELECT`. So the closure is conditional: the #: single-user path keeps the behaviour it has always had (INV-1), and a team #: daemon reads these through the scoped views or not at all. #: #: Longer than the conversation triple because the conversation is stored in #: more than one shape. `compaction_archives.compacted` is a JSON dump of the #: very message rows `messages` is closed for, so closing one and not the #: other closed nothing: `SELECT compacted FROM compaction_archives` returned #: a colleague's prompts verbatim. The rest are the same question asked of #: other per-session or per-project content -- review comments (which are #: also injected into turns), plans, notes, step records, share capabilities, #: and the command lines on compute jobs, a surface `team_policy` already #: makes admin-only *for reads* over HTTP. #: #: There is no `my_*` view for these yet, so in team mode they are unreadable #: rather than row-scoped. That is the deliberate direction: a table nobody #: has scoped is refused, not served whole. _TEAM_VIEW_ONLY_TABLES = frozenset( { "frames", "messages", "execution_log", "compaction_archives", "annotations", "annotation_admissions", "plans", "notes", "folders", "projects", "frame_steps", "shares", "compute_jobs", "compute_job_events", "custom_skills", "agents", } ) class _QueryAuthorizer: """SQLite's own answer to "may this statement touch that table?". An authorizer is called after parsing with *resolved* names, so quoting, a schema qualifier, an alias, a CTE wrapper and a table name arriving in a bound parameter are all the same thing to it. That is the whole reason for replacing the substring filter: the filter had to enumerate spellings and could not see a name that never appeared in the text. Deny rather than ignore. `SQLITE_IGNORE` on a read substitutes NULL and the query succeeds looking like an empty result, which is a worse answer than a refusal -- an agent would conclude the artifact does not exist. What was refused is recorded on the instance rather than recovered from SQLite's error text. The message differs by operation ("access to X.Y is prohibited" for a read, "not authorized" for others) and is not a documented interface, so matching on it would work until it did not. """ def __init__( self, view_only: frozenset[str] | None = None, *, published_views: frozenset[str] = frozenset(), ) -> None: self.denied: list[str] = [] # Which scoped views this statement may be read *through*. Empty when # the caller supplied no scope, so a query with no scope cannot reach # a view-only base table by claiming a view that was never created -- # the escape hatch used to accept the name whether or not anything # answered to it. self._published_views = frozenset(published_views) # Which base tables are reachable only through a scoped view. Passed # in rather than read from a module constant because the answer # depends on the deployment: a team daemon closes the conversation # family that a single-user one leaves open (INV-1). self._view_only = ( _VIEW_ONLY_TABLES if view_only is None else frozenset(view_only) ) def _deny(self, what: str) -> int: if what not in self.denied: self.denied.append(what) return sqlite3.SQLITE_DENY def __call__( self, action: int, arg1: str | None, arg2: str | None, dbname: str | None, source: str | None, ) -> int: # Anything that is not a read or a plain SELECT is refused outright. # This runs on the daemon's read-write connection, so it is the *only* # thing standing between agent SQL and an UPDATE -- and it refuses by # action code rather than by keyword, which is what makes a statement # like `SELECT 1; DROP TABLE artifacts` or a temp-table write, an ATTACH, # or a function-driven side effect unreachable rather than merely unspelled. if action not in (sqlite3.SQLITE_READ, sqlite3.SQLITE_SELECT): if action == sqlite3.SQLITE_FUNCTION: # Scalar/aggregate functions are fine; the table-valued pragma # functions arrive as reads of a `pragma_*` table below. return sqlite3.SQLITE_OK return self._deny(f"operation {action}") table = (arg1 or "").lower() if not table: return sqlite3.SQLITE_OK # The catalog, in all its spellings: sqlite_master, sqlite_schema, # sqlite_temp_master, sqlite_sequence, sqlite_stat1. Denying four names # by hand left the others open. if table.startswith("sqlite_"): return self._deny(table) # Table-valued pragma functions answer the same questions as the catalog # and slipped the ` pragma ` keyword check, which required spaces. if table.startswith("pragma_"): return self._deny(table) if table in QUERY_DENYLIST: return self._deny(table) if table in self._view_only: # Permitted only as the underlying read of a trusted scoped view # that this statement actually published. `_SCOPED_VIEWS` alone was # the test, and the name is caller-mintable through a CTE. if (source or "").lower() in self._published_views: return sqlite3.SQLITE_OK return self._deny(table) return sqlite3.SQLITE_OK #: `Connection.set_authorizer(None)` removes the authorizer from Python 3.11 #: onwards. On 3.10 -- this project's declared floor -- it does not. The C #: trampoline stays installed with no Python callable behind it, and SQLite #: reads a failed callback as `SQLITE_DENY`, so "take the guard off" silently #: became "deny everything". #: #: That is not a test-only difference. `Store.query` clears the guard twice: #: once to create the scoped views, which is a privileged setup step, and once #: in its `finally` to hand the connection back. On 3.10 the first clear made #: the view creation fail, and the second left the daemon's ONE connection #: deny-all for the rest of the process -- measured: after a single #: `host.query`, an ordinary `new_frame` raises `not authorized`. One agent SQL #: statement bricked the Store. #: #: A permissive callback is the portable way to say "no restrictions", and it #: costs nothing measurable: 11.4 us/query against 12.5 us with no authorizer #: at all, because SQLite only consults it while preparing a statement. _AUTHORIZER_ACCEPTS_NONE = sys.version_info >= (3, 11) def _allow_everything(*_event: object) -> int: return sqlite3.SQLITE_OK def _clear_authorizer(conn: sqlite3.Connection) -> None: """Take the guard off, on every interpreter this project supports.""" conn.set_authorizer(None if _AUTHORIZER_ACCEPTS_NONE else _allow_everything) class Store: """Thread-safe SQLite wrapper. One per data_dir; created lazily.""" def __init__(self, db_path: Path): self.db_path = Path(db_path) self._closed = False preflight_schema(self.db_path) # mode= on mkdir is masked by the umask and only applies on creation, # so harden explicitly and unconditionally afterwards. self.db_path.parent.mkdir(parents=True, exist_ok=True) self._lock = threading.RLock() self._conn = sqlite3.connect( str(self.db_path), check_same_thread=False, timeout=_BUSY_TIMEOUT_S, ) try: # Recheck the formal connection before hardening, DDL or seeds: # the database may have changed since the read-only preflight. require_supported_schema(self._conn) except BaseException: self._conn.close() self._closed = True raise harden_dir(self.db_path.parent) # SQLite creates the file at the process umask — 0644 on most systems. # This database holds plaintext credentials, so close it to the owner # as soon as it exists and before any schema is written into it. harden_db(self.db_path) self._conn.row_factory = sqlite3.Row self._apply_pragmas() self._conn.executescript(_SCHEMA) self._conn.commit() # Re-run: the schema write is the first thing that can materialise a # -wal/-shm sidecar, which would otherwise be born world-readable # carrying the same rows. harden_db(self.db_path) self._migrate() self._actions = ActionLedgerRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), admit_action_group=lambda group_id, operation: self._auto_mode.assert_repair_action_group_appendable( group_id, operation=operation ), admit_action_scope=lambda root_frame_id, branch_id, operation: self._auto_mode.assert_session_action_group_appendable( root_frame_id, branch_id, operation ), ) install_auto_mode_action_guards(self._conn) self._conn.commit() self._kernel_generations = KernelGenerationRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._checkpoint_states = CheckpointStateRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._session_snapshots = SessionSnapshotRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), checkpoint_state=self._checkpoint_states, ) self._auto_mode = AutoModeRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), get_branch=self._session_snapshots.get_branch, get_checkpoint=self._session_snapshots.get_checkpoint, checkpoint_is_restorable=lambda tree_id: WorkspaceCAS( self.db_path.parent / "workspace-cas" ).verify_tree(tree_id), get_action_group=lambda group_id: self._actions.get_group( group_id, include_events=True ), ) self._session_snapshots.set_revert_commit_hook( self._auto_mode.abandon_active_run_for_revert ) self._session_activation = SessionActivationRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), checkpoint_state=self._checkpoint_states, ) self._recovery_journal = RecoveryJournalRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._delegations = DelegationProjectionRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._plans = PlanRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._annotations = AnnotationRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._memories = MemoryRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._settings = SettingsRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._team = TeamRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._governance = GovernanceRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._workloads = WorkloadRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._leases = LeaseRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._user_keys = UserKeyRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._model_capability_receipts = ModelCapabilityReceiptRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._shares = SharesRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._permissions = PermissionRuleRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), get_setting=self.get_setting, set_setting=self.set_setting, admit_action_group=self._auto_mode.assert_repair_action_group_appendable, ) self._connectors = ConnectorRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._compute_jobs = ComputeJobRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._agents = AgentProfileRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._capability_repository = CapabilityStateRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._capabilities = CapabilityStateService(self._capability_repository) self._skill_versions = SkillVersionRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._specialists = SpecialistProfileService( self._agents, self._capabilities, ) self._frames = FrameRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), get_frame=lambda frame_id: self.get_frame(frame_id), resolve_frame_scope=lambda frame_id, **kwargs: self.resolve_frame_scope( frame_id, **kwargs ), get_project=lambda project_id: self.get_project(project_id), ) self._datapro_index = DataProIndexRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._artifacts = ArtifactRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), get_frame=lambda frame_id: self.get_frame(frame_id), resolve_frame_scope=lambda frame_id, **kwargs: self.resolve_frame_scope( frame_id, **kwargs ), resolve_artifact_write_scope=lambda **kwargs: self._artifact_write_scope( **kwargs ), execute=lambda sql, params=(): self._exec(sql, params), get_artifact=lambda artifact_id: self.get_artifact(artifact_id), get_env_snapshot=lambda snapshot_id: self.get_env_snapshot(snapshot_id), identify_file=lambda path: _file_identity(path), paths_match=lambda left, right: _same_file_path(left, right), delete_related=self._datapro_index.delete_for_artifact, ) self._completion_deliveries = CompletionDeliveryRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._session_imports = SessionImportRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._notes = NotesRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._folders = FolderRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._endpoints = EndpointRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._compactions = CompactionRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._host_calls = HostCallRepository( self._conn, self._lock, clock_ms=lambda: _now_ms(), ) self._load_capability_receipt_overlays() # --- migration (add columns missing from a pre-existing DB) ----------- _MIGRATIONS = { "messages": [("branch_id", "TEXT")], # Which in-flight request holds this pin. A reservation is what # makes admission exactly-once rather than at-most-once. "annotations": [("reservation_id", "TEXT")], "shares": [("expires_at", "INTEGER")], "frames": [ ("task_summary", "TEXT"), ("folder_id", "TEXT"), ("runtime_env", "TEXT"), ], "agents": [("system_prompt", "TEXT"), ("kind", "TEXT")], "artifact_versions": [("env_snapshot_id", "TEXT"), ("snapshot_path", "TEXT")], "env_snapshots": [("remote_json", "TEXT")], "execution_log": [ ("root_frame_id", "TEXT"), ("cell_index", "INTEGER"), ("state_revision", "INTEGER"), ("kernel_id", "TEXT"), ("language", "TEXT"), ("status", "TEXT"), ("code_hash", "TEXT"), ("visibility", "TEXT NOT NULL DEFAULT 'scientific'"), ("pin", "INTEGER NOT NULL DEFAULT 0"), ("replay_policy", "TEXT NOT NULL DEFAULT 'conditional'"), ("variable_reads", "TEXT NOT NULL DEFAULT '[]'"), ("variable_writes", "TEXT NOT NULL DEFAULT '[]'"), ("variable_deletes", "TEXT NOT NULL DEFAULT '[]'"), ("mutation_uncertain", "INTEGER NOT NULL DEFAULT 0"), ("figures", "TEXT"), ("files_read", "TEXT"), ("files_written", "TEXT"), ], "permission_requests": [ ("resolution_context", "TEXT"), ("continuation_required", "INTEGER NOT NULL DEFAULT 0"), ("continuation_expires_at", "INTEGER"), ("continuation_consumed_at", "INTEGER"), ("action_group_id", "TEXT"), ("action_id", "TEXT"), ("tool_call_id", "TEXT"), ("side_effect_class", "TEXT"), ("resource_keys", "TEXT"), ("dangerous", "INTEGER NOT NULL DEFAULT 0"), ("canonical_arguments_sha256", "TEXT"), ("action_digest", "TEXT"), ], "host_call_log": [ ("action_group_id", "TEXT"), ("action_id", "TEXT"), ("permission_decision_id", "TEXT"), ("result_preview", "TEXT"), ("result_digest", "TEXT"), ("side_effect_class", "TEXT"), ("resource_keys", "TEXT"), ], "compaction_archives": [ ("branch_id", "TEXT"), ("ledger_cursor", "TEXT"), ("recovery_pointer", "TEXT"), ("generation_id", "TEXT"), ("metadata", "TEXT"), ("handoff", "TEXT"), ("context_before", "TEXT"), ("context_after", "TEXT"), ("artifact_refs", "TEXT"), ], } def _migrate(self) -> None: """Bring the database to SCHEMA_VERSION, transactionally and once. The fast path is a ``PRAGMA user_version`` read: an already-current database does no probing at all, where previously every open re-derived the schema shape with a table_info scan per table. """ with self._lock: report = run_migrations( self._conn, self.db_path, { 1: ("legacy_baseline", self._apply_legacy_baseline), 2: ("compute_job_states", self._apply_compute_job_states), 3: ("compute_job_manifest", self._apply_compute_job_manifest), 4: ("artifact_env_identity", self._apply_artifact_env_identity), 5: ("artifact_source", self._apply_artifact_source), 6: ("compute_job_pgid", self._apply_compute_job_pgid), 7: ( "env_snapshot_generation", self._apply_env_snapshot_generation, ), 8: ( "env_snapshot_provenance", self._apply_env_snapshot_provenance, ), 9: ("compute_job_owner", self._apply_compute_job_owner), 10: ("frame_model_binding", self._apply_frame_model_binding), 11: ( "compute_job_idem_owner", self._apply_compute_job_idem_owner, ), 12: ( "annotation_version_binding", self._apply_annotation_version_binding, ), 13: ("memory_updated_at", self._apply_memory_updated_at), 14: ( "annotation_reservation", self._apply_annotation_reservation, ), 15: ( "annotation_admission_ledger", self._apply_annotation_admission_ledger, ), 16: ("datapro_content_index", self._apply_datapro_content_index), 17: ( "datapro_content_index_repair", self._apply_datapro_content_index_repair, ), 18: ("team_users", self._apply_team_users), 19: ("session_owners", self._apply_session_owners), 20: ("team_governance", self._apply_team_governance), 21: ( "orchestration_workloads", self._apply_orchestration_workloads, ), 22: ( "orchestration_leases", self._apply_orchestration_leases, ), 23: ("user_llm_keys", self._apply_user_llm_keys), 24: ( "artifact_observations_and_completion_delivery", self._apply_artifact_delivery, ), 25: ( "auto_mode_durable_state", self._apply_auto_mode_state, ), 26: ( "annotation_locators", self._apply_annotation_locators, ), 27: ( "compute_job_input_versions", self._apply_compute_job_input_versions, ), 28: ( "delegation_generation_and_task_status", self._apply_delegation_generation_and_task_status, ), 29: ( "auto_mode_budget_admission", self._apply_auto_mode_budget_admission, ), 30: ( "delegation_requests_and_attempts", self._apply_delegation_requests_and_attempts, ), 31: ( "model_capability_receipts", self._apply_model_capability_receipts, ), 32: ( "artifact_browse_index", self._apply_artifact_browse_index, ), }, ) if report["migrated"]: harden_db(self.db_path) def _apply_datapro_content_index(self, conn: sqlite3.Connection) -> None: """Version 16: lossless local indexing for DataPro responses.""" create_datapro_index_schema(conn) def _apply_team_users(self, conn: sqlite3.Connection) -> None: """Version 18: team-mode identity (users / auth_sessions / audit log). Additive only — no existing table changes shape, so a single-user install upgrades without behavioral change (INV-1). The DDL lives in storage/team.py and runs inside this numbered step, never at repository init (the v16/v17 lesson). """ create_team_schema(conn) def _apply_session_owners(self, conn: sqlite3.Connection) -> None: """Version 19: session ownership for team mode (M1-6, INV-13). Additive; existing sessions get no row, which the visibility rule reads as admin-only rather than everyone's. """ create_session_owners_schema(conn) def _apply_orchestration_workloads(self, conn: sqlite3.Connection) -> None: """Version 21: workloads and allocations (M3a-8). Carries the partial unique index that IS INV-3: a second live allocation for one workload is refused by the database, in the window between a check and an insert that no Python guard covers. """ create_workload_schema(conn) def _apply_orchestration_leases(self, conn: sqlite3.Connection) -> None: """Version 22: session leases and session↔workload bindings (M3b-4). Additive; a single-user install gets two empty tables and no behaviour change (INV-1). Nothing writes to them until a session actually asks for a cluster kernel. """ create_lease_schema(conn) def _apply_user_llm_keys(self, conn: sqlite3.Connection) -> None: """Version 23: per-user LLM credential references (M4-1, D7). Additive, and it holds references rather than keys — the secrets themselves stay in the SecretBroker, so this table copied off the machine names slots it cannot open. """ create_user_key_schema(conn) def _apply_artifact_delivery(self, conn: sqlite3.Connection) -> None: """Version 24: capture observations and recoverable final delivery. Both tables close one Artifact publication contract, so they advance together: an upgraded database can either record the producing Cell for reused bytes *and* bind final prose to exact versions, or it stays at v23 with neither half advertised as available. """ create_artifact_observations_schema(conn) create_completion_delivery_schema(conn) def _apply_auto_mode_state(self, conn: sqlite3.Connection) -> None: """Version 25: durable, idempotent Auto Run and audit facts. The focused repository constructor is deliberately passive. All seven tables and the checkpoint cursor advance in this one numbered, rollback-safe migration so an interrupted upgrade advertises neither half of the recovery contract. """ create_auto_mode_schema(conn) def _apply_auto_mode_budget_admission(self, conn: sqlite3.Connection) -> None: """Version 29: atomic Auto Mode budget reservations and circuit state. Additive, repeatable CREATE TABLE IF NOT EXISTS. Existing runs get no budget_state row and project as read-only ``legacy=true``; new runs create the row in the same start_run transaction. Guardian counters are not copied into these tables. """ create_auto_mode_budget_schema(conn) def _apply_annotation_locators(self, conn: sqlite3.Connection) -> None: """Version 26: PDF/HTML annotation locators next to image pins.""" from openai4s.storage.migrations import _is_duplicate_column for statement in ( "ALTER TABLE annotations ADD COLUMN kind TEXT", "ALTER TABLE annotations ADD COLUMN locator TEXT", ): try: conn.execute(statement) except sqlite3.OperationalError as error: if not _is_duplicate_column(error): raise def _apply_compute_job_input_versions(self, conn: sqlite3.Connection) -> None: """Version 27: durable lineage for inputs staged into remote jobs. Historical jobs keep NULL. Their input versions cannot be recovered from output manifests or command text, and inventing lineage would be worse than leaving it absent. """ from openai4s.storage.migrations import _is_duplicate_column try: conn.execute("ALTER TABLE compute_jobs ADD COLUMN input_versions TEXT") except sqlite3.OperationalError as error: if not _is_duplicate_column(error): raise def _apply_delegation_generation_and_task_status( self, conn: sqlite3.Connection ) -> None: """Version 28: durable identity for delegated child executions. ``execution_log.generation_id`` lets a directly-recorded Cell (a delegated child, which has no execution_attempts row) name the kernel generation that ran it; attempt-backed Web cells keep their attempt-derived binding. ``delegation_children.task_status`` records the child's derived completion contract alongside the lifecycle ``status``. Historical rows keep NULL — inventing either value would be provenance that is wrong rather than absent. """ from openai4s.storage.delegation import DELEGATION_SCHEMA from openai4s.storage.migrations import _is_duplicate_column, apply_ddl_script # The delegation tables belong to their repository, which is # constructed only after migrations finish — on a fresh database they # do not exist yet at this point. Idempotent DDL first, so the ALTER # below always has a table to alter (a freshly created table already # carries the column, which the guard treats as success). apply_ddl_script(conn, DELEGATION_SCHEMA) for statement in ( "ALTER TABLE execution_log ADD COLUMN generation_id TEXT", "ALTER TABLE delegation_children ADD COLUMN task_status TEXT", ): try: conn.execute(statement) except sqlite3.OperationalError as error: if not _is_duplicate_column(error): raise def _apply_delegation_requests_and_attempts(self, conn: sqlite3.Connection) -> None: """Version 30: durable request identity separate from attempt identity. Additive. Existing children keep no request row, so a replay without an explicit identity stays a new spawn. Rollback does not delete published Artifact versions. """ create_delegation_request_schema(conn) def _apply_model_capability_receipts(self, conn: sqlite3.Connection) -> None: """Version 31: exact model-capability receipts from an explicit probe. Additive. Timeout/auth/5xx never persist as false; only a current revision's positive evidence is adopted as an overlay. """ create_model_capability_receipts_schema(conn) def _apply_artifact_browse_index(self, conn: sqlite3.Connection) -> None: """Version 32: keyset browse index for the Artifact index route. Additive ``CREATE INDEX IF NOT EXISTS``. SQLite transactional DDL rolls the index away if this step fails; ``DROP INDEX IF EXISTS ix_artifacts_project_created`` is the reverse and leaves every Artifact row untouched. """ conn.execute( "CREATE INDEX IF NOT EXISTS ix_artifacts_project_created " "ON artifacts(project_id, created_at DESC, artifact_id DESC)" ) def _apply_team_governance(self, conn: sqlite3.Connection) -> None: """Version 20: membership, invites, usage ledger, quotas (M2). Additive; every table is dormant until team-mode governance uses it. """ create_governance_schema(conn) def _apply_datapro_content_index_repair(self, conn: sqlite3.Connection) -> None: """Version 17: repair an early v16 database stamped without its tables. During development the v16 DDL briefly ran outside the numbered migration and was later moved into its correct transaction. A local database opened between those revisions can therefore carry the v16 marker while missing one or both tables. Reapplying idempotent DDL under v17 makes that state recoverable without deleting user data. """ create_datapro_index_schema(conn) def _apply_annotation_admission_ledger(self, conn: sqlite3.Connection) -> None: """Version 15: a durable record of each admission attempt. The reservation column says a pin is held; it cannot say by which request, for which job, or whether the answer reached the client. After a lost response that is the only question worth asking, so it needs a row of its own rather than an inference from status. """ conn.execute( "CREATE TABLE IF NOT EXISTS annotation_admissions (" "reservation_id TEXT PRIMARY KEY," "root_frame_id TEXT NOT NULL," "annotation_ids TEXT NOT NULL," "request_id TEXT," "job_id TEXT," "message_id TEXT," "state TEXT NOT NULL," "created_at INTEGER NOT NULL," "updated_at INTEGER NOT NULL)" ) conn.execute( "CREATE INDEX IF NOT EXISTS ix_admission_frame " "ON annotation_admissions(root_frame_id)" ) conn.execute( "CREATE INDEX IF NOT EXISTS ix_admission_state " "ON annotation_admissions(state)" ) # The v14 rows this table cannot account for. # # Migration 14 added `reservation_id`, so a v14 install that crashed # mid-send has pins sitting at `reserved` with a holder. This migration # created the ledger *empty*, and every recovery path looks for a # ledger row -- so those pins upgraded into a state where nothing could # find them and nothing could free them: invisible in the composer, not # on any turn, permanently. The upgrade itself is the moment no request # is in flight, which is exactly when they are safe to release. # # No ledger row is fabricated for them. The request that held them left # no record of its id, its job or its outcome, and inventing one would # publish a correlation that never existed -- worse than the absence, # because a reconcile would believe it. They are simply given back. # Two statements, because they are two different claims. Only a pin # that is *held* goes back to `open`; a `sent`, `resolved` or # `dismissed` pin keeps its status, and a single `SET status='open'` # over `status='reserved' OR reservation_id IS NOT NULL` resurrected # every one of them that still carried a historical holder -- undoing # the user's review work and re-offering comments the model already # answered. conn.execute( "UPDATE annotations SET status='open', reservation_id=NULL " "WHERE status='reserved'" ) # The leftover holder on a pin that has already moved on is stale # bookkeeping: no live request answers for it, and `reservation_id` is # what every recovery and reconcile path matches on. Cleared without # touching the status. conn.execute( "UPDATE annotations SET reservation_id=NULL " "WHERE reservation_id IS NOT NULL" ) def _apply_annotation_reservation(self, conn: sqlite3.Connection) -> None: """Version 14: which in-flight request holds a pin. Added first to the ad-hoc add-column pass alone, which is why this exists. A *fresh* database gets the column from `CREATE TABLE`, so every test passed -- while an existing v13 install, meaning every install that already has data in it, would reach an `UPDATE annotations SET ... reservation_id=?` naming a column its table does not have. The blind spot is always the same one: fresh and upgraded are two different schemas, and only the fresh one is what tests build by default. """ try: conn.execute("ALTER TABLE annotations ADD COLUMN reservation_id TEXT") except sqlite3.OperationalError as exc: if not _is_duplicate_column(exc): raise # Unique per live reservation, so two requests cannot share an id and a # release cannot free somebody else's claim. Partial, because NULL is # the resting state of nearly every row. conn.execute( "CREATE UNIQUE INDEX IF NOT EXISTS ux_annot_reservation_row " "ON annotations(reservation_id, annotation_id) " "WHERE reservation_id IS NOT NULL" ) conn.execute( "CREATE INDEX IF NOT EXISTS ix_annot_reservation " "ON annotations(reservation_id) WHERE reservation_id IS NOT NULL" ) def _apply_memory_updated_at(self, conn: sqlite3.Connection) -> None: """Version 13: record when a memory was last edited. A memory could be written and deleted but never corrected, so fixing a typo in standing context meant deleting it and writing it again -- which loses its place in the newest-first order the pane and the injection both use, and which is a two-step round trip through a scope that may be at its cap. The column exists because retention needs it. `RETENTION_DAYS` withholds a memory that "has not been touched in a year", and with only `created_at` an edit was not a touch: correcting a stale instruction left it expiring on the original clock. Existing rows stay NULL, which reads as "never edited" and falls back to `created_at` -- no backfill, because when those rows were last edited is not recorded anywhere and writing today's date down would turn a guess into a fact. Runs inside the transaction owned by ``run_migrations``. """ try: conn.execute("ALTER TABLE memories ADD COLUMN updated_at INTEGER") except sqlite3.OperationalError as e: if not _is_duplicate_column(e): raise MigrationError( f"memories.updated_at could not be added: {e}" ) from e def _apply_annotation_version_binding(self, conn: sqlite3.Connection) -> None: """Version 12: bind an image annotation to the version it was pinned on. An annotation recorded only `artifact_id`, and the send path resolved that to the artifact's LATEST version -- so an agent re-plotting between the pin and the send handed the model a different picture while the pin coordinates still described the old one. Existing rows stay NULL rather than being backfilled with today's version: which version they were taken against is not recorded anywhere, and writing the current one down would turn a guess into a fact. The send path reads NULL as "unbound" and keeps the old behaviour for those rows only. Runs inside the transaction owned by ``run_migrations``. """ for column in ("version_id", "checksum"): try: conn.execute(f"ALTER TABLE annotations ADD COLUMN {column} TEXT") except sqlite3.OperationalError as e: if not _is_duplicate_column(e): raise MigrationError( f"annotations.{column} could not be added: {e}" ) from e def _apply_compute_job_idem_owner(self, conn: sqlite3.Connection) -> None: """Version 11: make the idempotency namespace per-owner. The old index was `UNIQUE(idempotency_key)` — installation-wide, while every other view of `compute_jobs` is per-owner. One session's key therefore blocked every other session's, and the duplicate refusal handed back the other session's `job_id` and status. Replacing an index is not additive, so it needs a real step rather than the idempotent catch-up pass. Order matters: build the new index first, so a database that already contains a cross-owner duplicate fails here — with the old index still in place — rather than losing the constraint and then failing. `COALESCE(owner_key,'')` because SQLite treats NULLs as distinct in a UNIQUE index, and NULL is exactly the CLI context this must keep protecting. Runs inside the transaction owned by ``run_migrations``. """ try: conn.execute( "CREATE UNIQUE INDEX IF NOT EXISTS ix_compute_jobs_idem_owner " "ON compute_jobs(COALESCE(owner_key,''), idempotency_key) " "WHERE idempotency_key IS NOT NULL" ) except sqlite3.IntegrityError as e: raise MigrationError( "compute_jobs holds rows that would violate a per-owner " f"idempotency index: {e}. Two jobs for one owner share a key; " "reconcile or remove one before upgrading." ) from e except sqlite3.OperationalError as e: raise MigrationError( f"the per-owner idempotency index could not be created: {e}" ) from e conn.execute("DROP INDEX IF EXISTS ix_compute_jobs_idem") def _apply_compute_job_states(self, conn: sqlite3.Connection) -> None: """Version 2: one enforced compute-job state vocabulary. Adds ``termination_reason`` and folds the two historical states that the new vocabulary does not have onto states that it does, preserving what each of them meant: * ``done`` -> ``succeeded`` (a rename, nothing else) * ``incomplete`` -> ``failed`` + ``outputs_unverified``. It meant the job exited 0 but its outputs could not be verified, which is not a success and must not keep a name that reads like one. * ``closed`` -> ``cancelled`` + ``handle_closed``. The user released the handle while the job was live. Idempotent: the ALTER is guarded, and every UPDATE selects only rows still carrying a legacy value, so a re-run after a partial apply converges rather than double-writing. Runs inside the transaction owned by ``run_migrations``; it must not commit. """ from openai4s.compute.states import LEGACY_STATUS_MAP have = { r["name"] for r in conn.execute("PRAGMA table_info(compute_jobs)").fetchall() } if "termination_reason" not in have: try: conn.execute( "ALTER TABLE compute_jobs ADD COLUMN termination_reason TEXT" ) except sqlite3.OperationalError as e: if not _is_duplicate_column(e): raise MigrationError( f"compute_jobs.termination_reason could not be added: {e}" ) from e for legacy, (status, reason) in LEGACY_STATUS_MAP.items(): conn.execute( "UPDATE compute_jobs SET status=?, termination_reason=? " "WHERE status=?", (status, reason, legacy), ) def _apply_artifact_source(self, conn: sqlite3.Connection) -> None: """Version 5: where a version's data came from. The Evidence scorecard asks every release-grade artifact to carry a source. There was no column at all, so the clause was structurally unmeetable rather than sparsely met. Historical rows keep NULL. A version written before retrieval was recorded has no recoverable source, and inventing one would be the same mistake as backfilling an environment: an unattributed record turned into a confidently wrong one. Runs inside the transaction owned by ``run_migrations``. """ have = { r["name"] for r in conn.execute("PRAGMA table_info(artifact_versions)").fetchall() } if "source" in have: return try: conn.execute("ALTER TABLE artifact_versions ADD COLUMN source TEXT") except sqlite3.OperationalError as e: if not _is_duplicate_column(e): raise MigrationError( f"artifact_versions.source could not be added: {e}" ) from e def _apply_artifact_env_identity(self, conn: sqlite3.Connection) -> None: """Version 4: say WHICH kernel an artifact's environment describes. Historical rows keep NULL. They were written by a snapshot that could only ever describe the daemon, so backfilling them with the daemon's identity would turn an unattributed record into a confidently wrong one -- the very failure this migration exists to stop. Runs inside the transaction owned by ``run_migrations``. """ have = { r["name"] for r in conn.execute("PRAGMA table_info(env_snapshots)").fetchall() } for column in ( "interpreter", "environment_name", "generation_id", "packages_unavailable", ): if column in have: continue try: conn.execute(f"ALTER TABLE env_snapshots ADD COLUMN {column} TEXT") except sqlite3.OperationalError as e: if not _is_duplicate_column(e): raise MigrationError( f"env_snapshots.{column} could not be added: {e}" ) from e def _apply_compute_job_pgid(self, conn: sqlite3.Connection) -> None: """Version 6: the remote process group, recorded rather than guessed. Cancellation signalled ``-$!``. In an interactive shell that is the pgid; in the non-interactive login shell an ``ssh host cmd`` actually gets — dash, ash, or bash without job control — ``set -m`` does not enable job control, so ``$!`` is the child's pid and its group is the login shell's. ``kill -- -`` then found no such group, exited 0 anyway on some hosts, and the caller was told the allocation was freed while the whole command tree kept running. Historical rows keep NULL: the group a finished submit landed in is not recoverable after the fact, and ``cancel`` treats a missing pgid as "cannot signal safely" rather than guessing one. Runs inside the transaction owned by ``run_migrations``. """ have = { r["name"] for r in conn.execute("PRAGMA table_info(compute_jobs)").fetchall() } if "pgid" in have: return try: conn.execute("ALTER TABLE compute_jobs ADD COLUMN pgid TEXT") except sqlite3.OperationalError as e: if not _is_duplicate_column(e): raise MigrationError( f"compute_jobs.pgid could not be added: {e}" ) from e def _apply_frame_model_binding(self, conn: sqlite3.Connection) -> None: """Version 10: record which model configuration a session actually used. A frame stored a model *string*. That answers "which model name" and not "which configuration", and the two differ in exactly the case that matters: two profiles can name the same model against different providers or endpoints, and editing a profile rewrote it in place, so a replayed session reported whatever the profile happened to say today. D2's rule is that a session binds `profile_id + revision` and never silently follows the latest. Historical rows keep NULL for both, which reads as *unbound* -- not as "used the default". That distinction is the point: an unbound session stays fully readable, and only sending a new message asks the user to rebind. Backfill happens at read time and only on a unique `(provider, endpoint, model)` match, because an ambiguous one is a guess and a guess here is the thing being removed. Runs inside the transaction owned by ``run_migrations``. """ have = {r["name"] for r in conn.execute("PRAGMA table_info(frames)").fetchall()} for column, decl in ( ("model_profile_id", "TEXT"), ("model_profile_revision", "INTEGER"), ): if column in have: continue try: conn.execute(f"ALTER TABLE frames ADD COLUMN {column} {decl}") except sqlite3.OperationalError as e: if not _is_duplicate_column(e): raise MigrationError( f"frames.{column} could not be added: {e}" ) from e def _apply_compute_job_owner(self, conn: sqlite3.Connection) -> None: """Version 9: record which session/workspace owns each compute job. Without it, ``_rehydrate`` loaded every installation-wide live row into whichever session built a manager first, so a restart could hand one session's job — and publish its harvested outputs — into a *different* session's workspace. The column lets recovery filter to the owning session. Historical rows keep NULL, which is the CLI / global context: only a NULL-owner (CLI) manager rehydrates them, never a Web session, so no session inherits another's pre-upgrade jobs. Runs inside the transaction owned by ``run_migrations``. """ have = { r["name"] for r in conn.execute("PRAGMA table_info(compute_jobs)").fetchall() } if "owner_key" in have: return try: conn.execute("ALTER TABLE compute_jobs ADD COLUMN owner_key TEXT") except sqlite3.OperationalError as e: if not _is_duplicate_column(e): raise MigrationError( f"compute_jobs.owner_key could not be added: {e}" ) from e def _apply_env_snapshot_generation(self, conn: sqlite3.Connection) -> None: """Version 7: qualify generation attributions instead of destroying them. ``env_snapshots`` rows are content-addressed, and until now the address did not include ``generation_id`` while ``upsert_env_snapshot`` never updated an existing row. A kernel restarted into an unchanged environment therefore resolved to the row already on disk — which kept naming the *first* generation. Every artifact produced by the second generation pointed at a snapshot recorded as the first, and no other column on the artifact carries a generation, so nothing could catch it. Which historical rows were actually shared is not recoverable: sharing leaves no trace. The first version of this migration answered that by clearing ``generation_id`` on every row written before the fix — which trades one wrong answer for a *missing* one and silently discards provenance that is right far more often than not. So the value is kept and **labelled**. ``generation_confidence`` says which reading applies: * ``verified`` — the row's address includes its generation, so it cannot have been shared; * ``legacy_unverified`` — written before the fix. The named generation produced this environment; it may not be the only one that did. A reader that needs certainty filters on the label. Nothing is lost, and nothing claims more than it can support. Idempotent: the label is derived from the row's own address, so a re-run recomputes the same answer. Runs inside the transaction owned by ``run_migrations``. """ from openai4s.storage.artifacts import env_snapshot_id have = { r["name"] for r in conn.execute("PRAGMA table_info(env_snapshots)").fetchall() } if "generation_confidence" not in have: try: conn.execute( "ALTER TABLE env_snapshots ADD COLUMN generation_confidence TEXT" ) except sqlite3.OperationalError as e: if not _is_duplicate_column(e): raise MigrationError( f"env_snapshots.generation_confidence could not be " f"added: {e}" ) from e rows = conn.execute( "SELECT snapshot_id,kind,python_version,implementation,platform," "interpreter,environment_name,generation_id,packages_json,remote_json " "FROM env_snapshots WHERE generation_id IS NOT NULL" ).fetchall() for row in rows: expected = env_snapshot_id( kind=row["kind"], python_version=row["python_version"], implementation=row["implementation"], platform=row["platform"], interpreter=row["interpreter"], environment_name=row["environment_name"], generation_id=row["generation_id"], # NULL is how an empty remote list is stored; the basis has # always used "[]" for it. packages_json=row["packages_json"] or "[]", remote_json=row["remote_json"] or "[]", ) conn.execute( "UPDATE env_snapshots SET generation_confidence=? " "WHERE snapshot_id=?", ( ( "verified" if expected == row["snapshot_id"] else "legacy_unverified" ), row["snapshot_id"], ), ) def _apply_env_snapshot_provenance(self, conn: sqlite3.Connection) -> None: """Version 8: room for the assumed-vs-measured marker. ``_snapshot_for`` has always stamped ``provenance: "assumed: no kernel generation on record"`` on the fallback path, and the INSERT never had a column for it — so the single field that tells a reader whether an environment was *observed* or *guessed from the daemon* was computed and thrown away on every write. Historical rows keep NULL. Whether a given old row was measured is not recoverable, and asserting either answer would be the same mistake the marker exists to prevent. Runs inside the transaction owned by ``run_migrations``. """ have = { r["name"] for r in conn.execute("PRAGMA table_info(env_snapshots)").fetchall() } if "provenance" in have: return try: conn.execute("ALTER TABLE env_snapshots ADD COLUMN provenance TEXT") except sqlite3.OperationalError as e: if not _is_duplicate_column(e): raise MigrationError( f"env_snapshots.provenance could not be added: {e}" ) from e def _apply_compute_job_manifest(self, conn: sqlite3.Connection) -> None: """Version 3: room to record what a job actually produced. Historical rows keep NULL — we cannot reconstruct a manifest for a harvest that happened before anything hashed it, and inventing one would be worse than admitting it is unknown. Only jobs harvested from here on carry the record. Runs inside the transaction owned by ``run_migrations``. """ have = { r["name"] for r in conn.execute("PRAGMA table_info(compute_jobs)").fetchall() } for column in ("artifact_manifest", "integrity_sha256"): if column in have: continue try: conn.execute(f"ALTER TABLE compute_jobs ADD COLUMN {column} TEXT") except sqlite3.OperationalError as e: if not _is_duplicate_column(e): raise MigrationError( f"compute_jobs.{column} could not be added: {e}" ) from e def _apply_legacy_baseline(self, conn: sqlite3.Connection) -> None: """Version 1: the historical catch-up pass, run once and then stamped. This is the whole of what ``_migrate`` used to do on every open. It is idempotent by construction — it adds only absent columns, and every backfill below is guarded by a predicate that selects only rows still needing it — which is exactly why a version can be retrofitted onto an existing database without reconstructing which ALTERs had already run. Converge once, stamp version 1, and stop re-deriving it forever after. Runs inside the transaction owned by run_migrations; it must not commit. """ for table, cols in self._MIGRATIONS.items(): have = { r["name"] for r in conn.execute(f"PRAGMA table_info({table})").fetchall() } for name, decl in cols: if name in have: continue try: conn.execute(f"ALTER TABLE {table} ADD COLUMN {name} {decl}") except sqlite3.OperationalError as e: # Only the one error a concurrent re-run legitimately # produces is benign. The old blanket swallow also hid # "database is locked" and "no such table", letting the # process continue against a schema missing a column it # believed it had — a migration failure surfacing much # later as an unexplained runtime error. if not _is_duplicate_column(e): raise MigrationError( f"ALTER TABLE {table} ADD COLUMN {name} {decl} " f"failed: {e}" ) from e # Historical child frames inherited the root id but silently kept # project_id='default'. Historical artifacts also used their actor # frame as root_frame_id. Repair both idempotently when the frame # tree still exists; unframed legacy uploads remain untouched. conn.execute( "UPDATE frames SET project_id=COALESCE((SELECT root.project_id " "FROM frames AS root WHERE root.frame_id=frames.root_frame_id)," "project_id) WHERE root_frame_id IS NOT NULL" ) conn.execute( "UPDATE artifacts SET project_id=COALESCE((SELECT root.project_id " "FROM frames AS actor JOIN frames AS root " "ON root.frame_id=actor.root_frame_id " "WHERE actor.frame_id=artifacts.root_frame_id),project_id) " "WHERE root_frame_id IN (SELECT frame_id FROM frames)" ) conn.execute( "UPDATE artifacts SET root_frame_id=COALESCE((SELECT " "actor.root_frame_id FROM frames AS actor " "WHERE actor.frame_id=artifacts.root_frame_id),root_frame_id) " "WHERE root_frame_id IN (SELECT frame_id FROM frames)" ) # Messages written before branch-aware history belonged to the # canonical root. Keep the rows immutable and backfill only their # newly-added routing projection. conn.execute( "UPDATE messages SET branch_id=root_frame_id " "WHERE branch_id IS NULL OR branch_id=''" ) conn.execute( "CREATE INDEX IF NOT EXISTS ix_msg_branch " "ON messages(root_frame_id,branch_id,seq)" ) # ``cell_index`` was already the session-monotonic allocation for # historical Web Cells. Backfill the explicitly named runtime # revision without pretending that rows which never had an index # carry recoverable state. conn.execute( "UPDATE execution_log SET state_revision=cell_index " "WHERE state_revision IS NULL AND cell_index IS NOT NULL" ) # Capture the same immutable dependency metadata for historical # Cells as for newly recorded ones. Rows are selected by the hash # sentinel, making the additive migration idempotent. legacy_cells = conn.execute( "SELECT producing_cell_id,code,language,origin,visibility," "replay_policy FROM execution_log WHERE code_hash IS NULL" ).fetchall() for cell in legacy_cells: visibility = cell["visibility"] or default_visibility(cell["origin"]) if visibility == "scientific" and str(cell["origin"] or "").lower() in { "system", "recovery", }: visibility = default_visibility(cell["origin"]) replay_policy = cell["replay_policy"] if not replay_policy or visibility in {"system", "recovery"}: replay_policy = default_replay_policy(visibility) metadata = analyze_code(cell["code"] or "", cell["language"] or "python") conn.execute( "UPDATE execution_log SET code_hash=?,visibility=?," "replay_policy=?,variable_reads=?,variable_writes=?," "variable_deletes=?,mutation_uncertain=? " "WHERE producing_cell_id=?", ( metadata.code_hash, visibility, replay_policy, json.dumps(metadata.reads, ensure_ascii=False), json.dumps(metadata.writes, ensure_ascii=False), json.dumps(metadata.deletes, ensure_ascii=False), 1 if metadata.uncertain else 0, cell["producing_cell_id"], ), ) def _apply_pragmas(self) -> None: """The connection's explicit PRAGMA policy. Stated rather than inherited, because a default that happens to be right is indistinguishable from one nobody chose — and the next person cannot tell which knobs were considered. Deliberately NOT set here: ``journal_mode``. It stays at the rollback-journal default. There is real multi-process access (``openai4s run`` and ``openai4s init`` open this database from their own process with no check that the daemon is not live), which is the usual argument for WAL — but measuring it showed a reader is not blocked by an in-flight writer under either mode, so there is no demonstrated problem for WAL to solve. Switching the on-disk format of a live user database on folklore is a bad trade; WAL also adds -wal/-shm sidecars and is unsafe on network filesystems. Revisit under a real concurrency and crash-recovery test, not before. ``synchronous``. Already FULL, which is the safe end. Lowering it trades crash durability for write speed on a database holding an audit ledger. Not a trade to make silently. """ with self._lock: # DataPro entries reference their batch with ON DELETE CASCADE. The # pragma is per-connection and OFF by default, so set it by policy # rather than leaving the first real foreign key as documentation. # Lifecycle repositories still delete both rows explicitly: that # keeps upgraded/externally-opened databases correct too. self._conn.execute("PRAGMA foreign_keys = ON") # --- secrets --------------------------------------------------------- @property def team(self) -> TeamRepository: """Team-mode identity: users, login sessions, audit log (M1-2).""" return self._team @property def governance(self) -> GovernanceRepository: """Team-mode governance: members, invites, usage, quotas (M2).""" return self._governance @property def workloads(self) -> WorkloadRepository: """Cluster workloads and allocations (M3a). Also the reconciler's WorkloadStore: the Protocol it needs is exactly this surface.""" return self._workloads @property def user_keys(self) -> UserKeyRepository: """Per-user LLM credential references (M4-1).""" return self._user_keys @property def model_capability_receipts(self) -> ModelCapabilityReceiptRepository: """Exact probe receipts bound to profile revision + endpoint.""" return self._model_capability_receipts @property def leases(self) -> LeaseRepository: """Session leases and session↔workload bindings (M3b-4).""" return self._leases @property def secrets(self): """The SecretBroker for this database, resolved once on first use. Lazy because resolution runs a real keychain round-trip self-test, and the overwhelming majority of Store construction (every test, every CLI subcommand that touches no credential) never needs a secret. """ with self._lock: broker = getattr(self, "_secret_broker", None) if broker is None: from openai4s.security.secret_broker import SecretBroker broker = SecretBroker(self) self._secret_broker = broker return broker def get_secret_setting(self, key: str, *, scope: str | None = None) -> str: """Read a credential setting, whether it is a reference or legacy plaintext. Both shapes have to work: an install that has not migrated, one that has, and one where migration failed for a single key must all keep running. Callers do not need to know which they are looking at. Nor does a caller need a row: a credential the operator injected into the daemon's environment resolves with no row at all, which is the only way a deployment that takes its credentials from the environment can ever have one. `scope` is optional because the known settings credentials are already mapped in `SETTINGS_SECRETS`; pass it for a key that is not. """ from openai4s.security.secret_migration import resolve_setting return resolve_setting(self, self.secrets, key, scope=scope) def set_secret_setting(self, key: str, value: str, *, scope: str) -> str: """Store a credential through the broker, recording only its reference. Returns the reference. An empty value clears both the reference and the stored secret — a cleared key must not linger in the keychain where the UI reports it as gone. One thing a clear cannot do is unset an operator-injected credential: the environment owns that value, `delete` on that backend is a no-op by design, and `get_secret_setting` keeps resolving it afterwards. That is the same boundary `put` states outright, so it is reported rather than hidden — the settings route answers with the `has_api_key` it re-reads after the write, which stays true. """ from openai4s.security.secret_broker import is_ref previous = self.get_setting(key) if not value: if is_ref(previous): try: self.secrets.delete(previous) except Exception: # noqa: BLE001 - clearing the row still matters pass self.set_setting(key, "") return "" ref = self.secrets.put(scope, key, value) # Verify before recording: a write that did not raise is not evidence # the value is retrievable, and a reference that resolves to nothing is # worse than the plaintext it replaced. if self.secrets.get(ref) != value: raise RuntimeError( f"refusing to record {key!r}: wrote to {ref} but could not read " f"it back" ) self.set_setting(key, ref) return ref # --- schema state ---------------------------------------------------- def schema_state(self) -> dict: """Report the database's schema version and how it got there. Exists so "is this database current, and what has been applied to it" is a question that can be answered without re-deriving the shape from table_info — which is what the code had to do before there was a version at all. """ with self._lock: version = current_version(self._conn) return { "version": version, "expected": SCHEMA_VERSION, "current": version == SCHEMA_VERSION, "applied": applied_migrations(self._conn), } # --- low-level ------------------------------------------------------- def _exec(self, sql: str, params: tuple = ()) -> None: with self._lock: self._conn.execute(sql, params) self._conn.commit() def close(self) -> None: with self._lock: if self._closed: return self._conn.close() self._closed = True _discard_store(self) # A managed DataPro HTTP connection retains a just-in-time header # provider closed over this Store and a bounded set of sent secrets for # reflection redaction. Once the Store generation ends, neither may # remain in the process-wide MCP cache. Resolve the scope from object # identity (never credential material), and crucially do not create an # MCP manager for the many Stores that never used a connector. from openai4s import datapro from openai4s.mcp_client import disconnect_if_initialized disconnect_if_initialized( datapro.CONNECTOR_ID, cache_scope=datapro.runtime_cache_scope(self), ) def _load_capability_receipt_overlays(self) -> None: """Re-adopt current-revision positive receipts after a restart.""" from openai4s.llm.capabilities import ( PROBE_VERSION, drop_receipt_overlays, install_receipt_overlay, ) drop_receipt_overlays() current = { str(profile.get("id") or ""): int(profile.get("revision") or 0) for profile in self.list_model_profiles() if not profile.get("deleted_at") } for receipt in self._model_capability_receipts.list_all(): profile_id = str(receipt.get("profile_id") or "") if current.get(profile_id) != int(receipt.get("revision") or 0): continue if int(receipt.get("probe_version") or 0) != PROBE_VERSION: continue install_receipt_overlay(receipt) # --- frames ---------------------------------------------------------- def new_frame( self, *, parent_id: str | None = None, project_id: str = "default", kind: str = "turn", name: str | None = None, model: str | None = None, depth: int = 0, status: str = "processing", ) -> str: return self._frames.new_frame( parent_id=parent_id, project_id=project_id, kind=kind, name=name, model=model, depth=depth, status=status, ) def resolve_frame_scope( self, frame_id: str | None, *, fallback_project: str = "default", ) -> dict: return self._frames.resolve_frame_scope( frame_id, fallback_project=fallback_project, ) def unpin_model(self, frame_id: str) -> None: self._frames.unpin_model(frame_id) def release_model_binding(self, profile_id: str) -> int: return self._frames.release_model_binding(profile_id) def update_frame(self, frame_id: str, **fields: Any) -> None: self._frames.update_frame(frame_id, **fields) def add_frame_tokens( self, frame_id: str, *, input_tokens: int = 0, output_tokens: int = 0, cost_usd: float = 0.0, ) -> None: self._frames.add_frame_tokens( frame_id, input_tokens=input_tokens, output_tokens=output_tokens, cost_usd=cost_usd, ) # --- projects -------------------------------------------------------- def create_project( self, *, name: str, description: str = "", context: str = "", project_id: str | None = None, is_example: bool = False, ) -> dict: return self._frames.create_project( name=name, description=description, context=context, project_id=project_id, is_example=is_example, ) def create_quarantined_import_session( self, *, project_id: str, quarantine_value: str, ) -> dict[str, Any]: return self._session_imports.create_quarantined_root( project_id=project_id, quarantine_value=quarantine_value, ) def get_project(self, project_id: str) -> dict | None: return self._frames.get_project(project_id) def update_project(self, project_id: str, **fields: Any) -> None: self._frames.update_project(project_id, **fields) def delete_project(self, project_id: str) -> dict: return self._frames.delete_project(project_id) def project_session_ids(self, project_id: str) -> list[str]: return self._frames.project_session_ids(project_id) def list_projects( self, *, q: str | None = None, limit: int | None = None, offset: int | None = None, before: tuple[int | None, str] | None = None, visible_to_user_id: str | None = None, ) -> list[dict]: return self._frames.list_projects( q=q, limit=limit, offset=offset, before=before, visible_to_user_id=visible_to_user_id, ) def count_projects( self, *, q: str | None = None, visible_to_user_id: str | None = None ) -> int: return self._frames.count_projects(q=q, visible_to_user_id=visible_to_user_id) # --- messages -------------------------------------------------------- def add_message( self, *, root_frame_id: str, branch_id: str | None = None, role: str, content: str, frame_id: str | None = None, metadata: dict | None = None, created_at: int | None = None, ) -> dict: return self._frames.add_message( root_frame_id=root_frame_id, branch_id=branch_id, role=role, content=content, frame_id=frame_id, metadata=metadata, created_at=created_at, ) def update_message_metadata(self, message_id: str, patch: dict) -> dict | None: return self._frames.update_message_metadata(message_id, patch) def promote_candidate_message( self, *, message_id: str, root_frame_id: str, branch_id: str, frame_id: str | None, expected_content: str, content: str, metadata: Mapping[str, Any], ) -> dict: return self._frames.promote_candidate_message( message_id=message_id, root_frame_id=root_frame_id, branch_id=branch_id, frame_id=frame_id, expected_content=expected_content, content=content, metadata=metadata, ) def commit_completion_delivery( self, *, idempotency_key: str, root_frame_id: str, branch_id: str | None, frame_id: str | None, content: str, manifest: Mapping[str, Any], message_metadata: Mapping[str, Any] | None = None, expected_manifest_sha256: str | None = None, created_at: int | None = None, snapshot_verifier: Callable[[Mapping[str, Any]], object] | None = None, ) -> dict[str, Any]: return self._completion_deliveries.commit_final_message( idempotency_key=idempotency_key, root_frame_id=root_frame_id, branch_id=branch_id, frame_id=frame_id, content=content, manifest=manifest, message_metadata=message_metadata, expected_manifest_sha256=expected_manifest_sha256, created_at=created_at, snapshot_verifier=snapshot_verifier, ) def promote_candidate_delivery( self, *, delivery_id: str, message_id: str, root_frame_id: str, branch_id: str | None, frame_id: str | None, expected_content: str, content: str, message_metadata: Mapping[str, Any], ) -> dict[str, Any]: return self._completion_deliveries.promote_candidate_delivery( delivery_id=delivery_id, message_id=message_id, root_frame_id=root_frame_id, branch_id=branch_id, frame_id=frame_id, expected_content=expected_content, content=content, message_metadata=message_metadata, ) def mark_completion_delivery_published( self, delivery_id: str, *, published_at: int | None = None, ) -> dict[str, Any]: return self._completion_deliveries.mark_published( delivery_id, published_at=published_at, ) def get_completion_delivery(self, delivery_id: str) -> dict[str, Any] | None: return self._completion_deliveries.get(delivery_id) def committed_completion_deliveries( self, *, root_frame_id: str | None = None, branch_id: str | None = None, limit: int = 1000, ) -> list[dict[str, Any]]: return self._completion_deliveries.committed( root_frame_id=root_frame_id, branch_id=branch_id, limit=limit, ) def completion_deliveries_for_session( self, root_frame_id: str, *, limit: int = 10_000, ) -> list[dict[str, Any]]: return self._completion_deliveries.for_session( root_frame_id, limit=limit, ) def bind_imported_completion_delivery( self, **fields: Any, ) -> dict[str, Any]: return self._completion_deliveries.bind_imported_message(**fields) def list_messages( self, root_frame_id: str, *, branch_id: str | None = None, start: int = 0, limit: int | None = 300, before_seq: int | None = None, newest_first: bool = False, ) -> list[dict]: messages = self._frames.list_messages( root_frame_id, branch_id=branch_id, start=start, limit=limit, before_seq=before_seq, newest_first=newest_first, ) return self._completion_deliveries.validate_message_projection( root_frame_id, messages, ) def list_message_boundaries( self, root_frame_id: str, *, branch_id: str | None = None, start: int = 0, limit: int | None = 300, ) -> list[dict]: messages = self._frames.list_message_boundaries( root_frame_id, branch_id=branch_id, start=start, limit=limit, ) return self._completion_deliveries.validate_message_projection( root_frame_id, messages, ) def list_branch_messages( self, root_frame_id: str, *, branch_id: str | None = None, start: int = 0, limit: int | None = 300, before_seq: int | None = None, newest_first: bool = False, boundaries: bool = False, ) -> list[dict]: """Project one branch's visible conversation without deleting rows.""" reader = self.list_message_boundaries if boundaries else self.list_messages projected = project_branch_records( self, root_frame_id, branch_id or self.active_session_branch(root_frame_id), list_local=lambda selected: reader( root_frame_id, branch_id=selected, limit=None, ), record_position=lambda message: int(message.get("seq") or 0), cursor_key="message_cursor", normalize_cursor=count_cursor, ) if newest_first or before_seq is not None: # Latest-first, walking backwards by `seq`. Opening a 640-message # session used to return messages 0-299 -- the *oldest* page, with # the newest 340 absent -- because the only order was ascending and # the only bound was a limit. A reader arriving at a long session # wants its end. # # Honest about what this does NOT do: the branch projection above # is whole-history by construction (it walks branch cursors to # decide what is visible at all), so this pages the projected list # rather than pushing a cursor into SQL. The user-visible defect -- # seeing the wrong end of the conversation -- is fixed; the read is # still O(branch). Making the projection incremental is a larger # change and is not claimed here. ordered = sorted( projected, key=lambda m: int(m.get("seq") or 0), reverse=True ) if before_seq is not None: ordered = [ m for m in ordered if int(m.get("seq") or 0) < int(before_seq) ] if limit is None: return ordered return ordered[: max(0, int(limit))] start = max(0, int(start)) if limit is None: return projected[start:] return projected[start : start + max(0, int(limit))] def list_branch_message_boundaries( self, root_frame_id: str, *, branch_id: str | None = None, start: int = 0, limit: int | None = 300, before_seq: int | None = None, newest_first: bool = False, ) -> list[dict]: return self.list_branch_messages( root_frame_id, branch_id=branch_id, start=start, limit=limit, before_seq=before_seq, newest_first=newest_first, boundaries=True, ) def message_count(self, root_frame_id: str) -> int: return self._frames.message_count(root_frame_id) def has_message_history(self) -> bool: return self._frames.has_message_history() def has_execution_history(self) -> bool: return self._frames.has_execution_history() def cell_count(self, root_frame_id: str) -> int: return self._frames.cell_count(root_frame_id) def latest_state_revision(self, root_frame_id: str) -> int: return self._frames.latest_state_revision(root_frame_id) # --- semantic activity steps (plan / search / env / skill / edit / …) ---- # Every visible host.* tool call becomes a persisted "step" so a reopened # session re-renders the same rich activity (not just the final prose). def add_step( self, *, step_id: str, frame_id: str, kind: str, title: str | None = None, input: dict | None = None, status: str = "running", ) -> dict: return self._frames.add_step( step_id=step_id, frame_id=frame_id, kind=kind, title=title, input=input, status=status, ) def update_step( self, step_id: str, *, status: str | None = None, output: dict | None = None, title: str | None = None, summary: str | None = None, ) -> None: self._frames.update_step( step_id, status=status, output=output, title=title, summary=summary, ) def list_steps( self, frame_id: str, *, start: int = 0, limit: int = 800 ) -> list[dict]: return self._frames.list_steps(frame_id, start=start, limit=limit) def step_count(self, frame_id: str) -> int: return self._frames.step_count(frame_id) # --- frame browse / detail / search -------------------------- def browse_frames( self, *, project_id: str | None = "default", status: str | None = None, roots_only: bool = True, limit: int = 50, before: tuple[int, str] | None = None, visible_to_user_id: str | None = None, ) -> list[dict]: return self._frames.browse_frames( project_id=project_id, status=status, roots_only=roots_only, limit=limit, before=before, visible_to_user_id=visible_to_user_id, ) def frame_detail( self, frame_id: str, *, page: int = 0, page_size: int = 50, visible_to_user_id: str | None = None, ) -> dict | None: return self._frames.frame_detail( frame_id, page=page, page_size=page_size, visible_to_user_id=visible_to_user_id, ) def search_frames( self, pattern: str, *, project_id: str | None = "default", limit: int = 50, visible_to_user_id: str | None = None, ) -> list[dict]: return self._frames.search_frames( pattern, project_id=project_id, limit=limit, visible_to_user_id=visible_to_user_id, ) # --- execution_log --------------------------------------------------- def log_cell( self, *, frame_id: str | None, code: str, result: dict, origin: str = "agent", cell_seq: int | None = None, project_id: str = "default", root_frame_id: str | None = None, cell_index: int | None = None, state_revision: int | None = None, kernel_id: str = "python", language: str = "python", visibility: str | None = None, pin: bool = False, replay_policy: str | None = None, figures: list | None = None, files_read: list | None = None, files_written: list | None = None, generation_id: str | None = None, ) -> str: return self._frames.log_cell( frame_id=frame_id, code=code, result=result, origin=origin, cell_seq=cell_seq, project_id=project_id, root_frame_id=root_frame_id, cell_index=cell_index, state_revision=state_revision, kernel_id=kernel_id, language=language, visibility=visibility, pin=pin, replay_policy=replay_policy, figures=figures, files_read=files_read, files_written=files_written, generation_id=generation_id, ) def list_cells( self, root_frame_id: str, *, branch_id: str | None = None ) -> list[dict]: return self._frames.list_cells(root_frame_id, branch_id=branch_id) def list_cell_outputs(self, root_frame_id: str) -> list[dict]: return self._frames.list_cell_outputs(root_frame_id) def cell_detail(self, producing_cell_id: str) -> dict | None: return self._frames.cell_detail(producing_cell_id) # --- canonical action ledger --------------------------------------- def append_action_group( self, *, root_frame_id: str, turn_id: str, kind: str, branch_id: str | None = None, ordinal: int | None = None, provider: str | None = None, model: str | None = None, wire_state: Any = None, assistant_content: str | None = None, assistant_message: Any = None, usage: dict[str, Any] | None = None, cost_usd: float | None = None, group_id: str | None = None, created_at: int | None = None, ) -> dict: return self._actions.append_group( root_frame_id=root_frame_id, branch_id=branch_id, turn_id=turn_id, ordinal=ordinal, kind=kind, provider=provider, model=model, wire_state=wire_state, assistant_content=assistant_content, assistant_message=assistant_message, usage=usage, cost_usd=cost_usd, group_id=group_id, created_at=created_at, ) def append_action_event( self, *, group_id: str, type: str, sequence: int | None = None, action_id: str | None = None, tool_call_id: str | None = None, wire_id: str | None = None, canonical_arguments: Any = None, raw_arguments: Any = None, result: Any = None, side_effect_class: str | None = None, resource_keys: list[str] | tuple[str, ...] | None = None, event_id: str | None = None, created_at: int | None = None, ) -> dict: return self._actions.append_event( group_id=group_id, type=type, sequence=sequence, action_id=action_id, tool_call_id=tool_call_id, wire_id=wire_id, canonical_arguments=canonical_arguments, raw_arguments=raw_arguments, result=result, side_effect_class=side_effect_class, resource_keys=resource_keys, event_id=event_id, created_at=created_at, ) def append_tool_action_group( self, *, root_frame_id: str, turn_id: str, events: list[dict[str, Any]], branch_id: str | None = None, ordinal: int | None = None, provider: str | None = None, model: str | None = None, wire_state: Any = None, assistant_content: str | None = None, assistant_message: Any = None, usage: dict[str, Any] | None = None, cost_usd: float | None = None, group_id: str | None = None, created_at: int | None = None, ) -> dict: return self._actions.append_tool_group( root_frame_id=root_frame_id, branch_id=branch_id, turn_id=turn_id, events=events, ordinal=ordinal, provider=provider, model=model, wire_state=wire_state, assistant_content=assistant_content, assistant_message=assistant_message, usage=usage, cost_usd=cost_usd, group_id=group_id, created_at=created_at, ) def append_action_group_with_events( self, *, root_frame_id: str, branch_id: str, turn_id: str, kind: str, events: list[dict[str, Any]], group_id: str, admission_operation: str, ) -> dict: """Atomically publish one non-provider lifecycle group and its events.""" return self._actions.append_tool_group( root_frame_id=root_frame_id, branch_id=branch_id, turn_id=turn_id, kind=kind, events=events, group_id=group_id, admission_operation=admission_operation, ) def get_action_group( self, group_id: str, *, include_events: bool = True ) -> dict | None: return self._actions.get_group(group_id, include_events=include_events) def latest_action_group_ordinal( self, root_frame_id: str, *, branch_id: str | None = None, kind: str | None = None, ) -> int | None: return self._actions.latest_group_ordinal( root_frame_id, branch_id=branch_id, kind=kind ) def list_action_groups( self, root_frame_id: str, *, branch_id: str | None = None, turn_id: str | None = None, after_ordinal: int | None = None, limit: int | None = None, include_events: bool = True, ) -> list[dict]: return self._actions.list_groups( root_frame_id, branch_id=branch_id, turn_id=turn_id, after_ordinal=after_ordinal, limit=limit, include_events=include_events, ) def list_action_events(self, group_id: str) -> list[dict]: return self._actions.list_events(group_id) def allocate_execution_attempt( self, *, group_id: str, producing_cell_id: str, state_revision: int | None = None, generation_id: str | None = None, owner_instance_id: str | None = None, replayed_from_cell_id: str | None = None, attempt_ordinal: int | None = None, attempt_id: str | None = None, allocated_at: int | None = None, ) -> dict: return self._actions.allocate_attempt( group_id=group_id, producing_cell_id=producing_cell_id, state_revision=state_revision, generation_id=generation_id, owner_instance_id=owner_instance_id, replayed_from_cell_id=replayed_from_cell_id, attempt_ordinal=attempt_ordinal, attempt_id=attempt_id, allocated_at=allocated_at, ) def mark_execution_attempt_started( self, attempt_id: str, *, started_at: int | None = None ) -> dict: return self._actions.mark_attempt_started(attempt_id, started_at=started_at) def bind_execution_attempt_generation( self, attempt_id: str, generation_id: str ) -> dict: return self._actions.bind_attempt_generation(attempt_id, generation_id) def abandon_incomplete_execution_attempts( self, *, owner_instance_id: str, finished_at: int | None = None, ) -> int: return self._actions.abandon_incomplete_attempts( owner_instance_id=owner_instance_id, finished_at=finished_at, ) def mark_execution_attempt_response( self, attempt_id: str, *, response_at: int | None = None ) -> dict: return self._actions.mark_attempt_response(attempt_id, response_at=response_at) def mark_execution_attempt_capture( self, attempt_id: str, *, capture_at: int | None = None ) -> dict: return self._actions.mark_attempt_capture(attempt_id, capture_at=capture_at) def finish_execution_attempt( self, attempt_id: str, *, terminal_state: str, error: Any = None, finished_at: int | None = None, ) -> dict: return self._actions.finish_attempt( attempt_id, terminal_state=terminal_state, error=error, finished_at=finished_at, ) def get_execution_attempt(self, attempt_id: str) -> dict | None: return self._actions.get_attempt(attempt_id) def list_execution_attempts( self, *, group_id: str | None = None, producing_cell_id: str | None = None, root_frame_id: str | None = None, branch_id: str | None = None, turn_id: str | None = None, ) -> list[dict]: return self._actions.list_attempts( group_id=group_id, producing_cell_id=producing_cell_id, root_frame_id=root_frame_id, branch_id=branch_id, turn_id=turn_id, ) # --- persistent kernel generations -------------------------------- def create_kernel_generation(self, **fields: Any) -> dict: return self._kernel_generations.create(**fields) def touch_kernel_generation(self, generation_id: str, **fields: Any) -> dict: return self._kernel_generations.touch(generation_id, **fields) def compare_and_swap_kernel_bootstrap( self, generation_id: str, *, expected_manifest_id: str | None, bootstrap: Any, at: int | None = None, ) -> dict | None: return self._kernel_generations.compare_and_swap_bootstrap( generation_id, expected_manifest_id=expected_manifest_id, bootstrap=bootstrap, at=at, ) def finish_kernel_generation( self, generation_id: str, *, state: str, reason: str, ended_at: int | None = None, ) -> dict: return self._kernel_generations.finish( generation_id, state=state, reason=reason, ended_at=ended_at, ) def abandon_live_kernel_generations( self, *, owner_instance_id: str, reason: str = "daemon_restart", ended_at: int | None = None, ) -> int: return self._kernel_generations.abandon_live( owner_instance_id=owner_instance_id, reason=reason, ended_at=ended_at, ) def get_kernel_generation(self, generation_id: str) -> dict | None: return self._kernel_generations.get(generation_id) def latest_kernel_generation( self, root_frame_id: str, language: str, *, branch_id: str | None = None, ) -> dict | None: return self._kernel_generations.latest( root_frame_id, language, branch_id=branch_id, ) def list_kernel_generations( self, root_frame_id: str, *, language: str | None = None, branch_id: str | None = None, ) -> list[dict]: return self._kernel_generations.list( root_frame_id, language=language, branch_id=branch_id, ) # --- durable Auto Mode state / audit events ------------------------ def reconcile_orphaned_auto_mode_candidates( self, *, now: int ) -> list[dict[str, Any]]: return self._completion_deliveries.reconcile_orphaned_candidates(now=now) def reconcile_orphaned_auto_mode_runs( self, *, owner_instance_id: str, now: int ) -> list[dict]: return self._auto_mode.reconcile_orphaned_runs( owner_instance_id=owner_instance_id, now=now ) def get_auto_mode_selection(self, scope_kind: str, scope_id: str) -> dict | None: return self._auto_mode.get_selection(scope_kind, scope_id) def set_auto_mode_selection( self, scope_kind: str, scope_id: str, values: dict, expected_revision: int, ) -> dict: return self._auto_mode.set_selection( scope_kind, scope_id, values, expected_revision=expected_revision, ) def start_auto_mode_run(self, **fields: Any) -> dict: return self._auto_mode.start_run(**fields) def record_auto_mode_candidate(self, run_id: str, **fields: Any) -> dict: return self._auto_mode.record_candidate(run_id, **fields) def start_auto_mode_review(self, run_id: str, **fields: Any) -> dict: return self._auto_mode.start_review(run_id, **fields) def complete_auto_mode_review(self, review_run_id: str, **fields: Any) -> dict: return self._auto_mode.complete_review(review_run_id, **fields) def start_auto_mode_repair(self, run_id: str, **fields: Any) -> dict: return self._auto_mode.start_repair(run_id, **fields) def complete_auto_mode_repair(self, repair_run_id: str, **fields: Any) -> dict: return self._auto_mode.complete_repair(repair_run_id, **fields) def bind_auto_mode_repair_execution_group( self, repair_run_id: str, **fields: Any ) -> dict: return self._auto_mode.bind_repair_execution_group(repair_run_id, **fields) def start_permission_review_assessment(self, run_id: str, **fields: Any) -> dict: return self._auto_mode.start_permission_review(run_id, **fields) def complete_permission_review_assessment( self, assessment_id: str, **fields: Any ) -> dict: return self._auto_mode.complete_permission_review(assessment_id, **fields) def terminate_auto_mode_run(self, run_id: str, **fields: Any) -> dict: return self._auto_mode.terminate_run(run_id, **fields) def auto_mode_event_cursor( self, root_frame_id: str, branch_id: str | None = None ) -> int: return self._auto_mode.event_cursor(root_frame_id, branch_id=branch_id) def list_auto_mode_events( self, root_frame_id: str, *, branch_id: str | None = None, after_cursor: int | None = None, upto_cursor: int | None = None, limit: int = 100_000, ) -> list[dict]: return self._auto_mode.list_events( root_frame_id, branch_id=branch_id, after_cursor=after_cursor, upto_cursor=upto_cursor, limit=limit, ) def project_auto_mode_run( self, root_frame_id: str, branch_id: str, upto_event_cursor: int | None = None, ) -> dict: return self._auto_mode.project_run( root_frame_id, branch_id, upto_event_cursor=upto_event_cursor, ) def list_auto_mode_audits( self, root_frame_id: str, branch_id: str, *, subject_kind: str | None = None, before: str | None = None, limit: int = 100, ) -> list[dict]: return self._auto_mode.list_audits( root_frame_id, branch_id, subject_kind=subject_kind, before=before, limit=limit, ) def export_auto_mode_projection( self, root_frame_id: str, *, branch_id: str | None = None, upto_event_cursor: int | None = None, ) -> dict: return self._auto_mode.export_projection( root_frame_id, branch_id=branch_id, upto_event_cursor=upto_event_cursor, ) def import_quarantined_auto_mode_projection( self, source: dict, **context: Any ) -> dict: return self._auto_mode.import_quarantined_projection(source, **context) def ensure_auto_mode_budget_state(self, run_id: str, **fields: Any) -> dict | None: return self._auto_mode.ensure_budget_state(run_id, **fields) def get_auto_mode_budget_state(self, run_id: str) -> dict | None: return self._auto_mode.get_budget_state(run_id) def list_auto_mode_budget_reservations(self, run_id: str) -> list[dict]: return self._auto_mode.list_budget_reservations(run_id) def reserve_auto_mode_budget(self, **fields: Any) -> dict: return self._auto_mode.reserve_budget(**fields) def commit_auto_mode_budget(self, admission_id: str, **fields: Any) -> dict: return self._auto_mode.commit_budget(admission_id, **fields) def release_auto_mode_budget(self, admission_id: str, **fields: Any) -> dict: return self._auto_mode.release_budget(admission_id, **fields) def mark_auto_mode_budget_unknown(self, admission_id: str) -> dict: return self._auto_mode.mark_budget_unknown(admission_id) def reconcile_auto_mode_budget(self, admission_id: str, **fields: Any) -> dict: return self._auto_mode.reconcile_budget(admission_id, **fields) def record_auto_mode_budget_delta(self, run_id: str, **fields: Any) -> dict: return self._auto_mode.record_budget_delta(run_id, **fields) def freeze_auto_mode_budget_initial_tokens( self, run_id: str, tokens: int, **fields: Any ) -> dict: return self._auto_mode.freeze_budget_initial_tokens(run_id, tokens, **fields) def trip_auto_mode_budget_circuit(self, run_id: str, **fields: Any) -> dict: return self._auto_mode.trip_budget_circuit(run_id, **fields) def project_auto_mode_budget(self, run_id: str) -> dict | None: return self._auto_mode.project_budget(run_id) # --- immutable session checkpoints / branches ---------------------- def ensure_session_branch(self, **fields: Any) -> dict: return self._session_snapshots.ensure_branch(**fields) def create_session_checkpoint(self, **fields: Any) -> dict: return self._session_snapshots.create_checkpoint(**fields) def get_checkpoint_state_snapshot( self, checkpoint_id: str, *, include_state: bool = False, ) -> dict | None: return self._checkpoint_states.get( checkpoint_id, include_state=include_state, ) def list_checkpoint_state_snapshots( self, root_frame_id: str, *, branch_id: str | None = None, limit: int = 100, ) -> list[dict]: return self._checkpoint_states.list( root_frame_id, branch_id=branch_id, limit=limit, ) def import_quarantined_checkpoint_state( self, source: dict, *, checkpoint_id: str, root_frame_id: str, branch_id: str, project_id: str, artifact_id_map: dict[str, str] | None = None, source_checkpoint_id: str | None = None, ) -> dict: return self._checkpoint_states.import_quarantined_snapshot( source, checkpoint_id=checkpoint_id, root_frame_id=root_frame_id, branch_id=branch_id, project_id=project_id, artifact_id_map=artifact_id_map, source_checkpoint_id=source_checkpoint_id, ) def validate_checkpoint_state_import( self, source: dict, *, include_state: bool = False, ) -> dict: return self._checkpoint_states.validate_checkpoint_state_import( source, include_state=include_state, ) def restore_checkpoint_state_snapshot( self, *, checkpoint_id: str, root_frame_id: str, project_id: str, ) -> dict: """Restore only the structured plan/review/memory projection. Normal branch activation calls the same repository inside its broader atomic publication transaction. This narrow facade exists for repair tooling and direct repository contract tests. """ return self._checkpoint_states.restore_checkpoint( checkpoint_id=checkpoint_id, root_frame_id=root_frame_id, project_id=project_id, ) def fork_session_branch(self, **fields: Any) -> dict: return self._session_snapshots.fork_branch(**fields) def get_session_checkpoint(self, checkpoint_id: str) -> dict | None: return self._session_snapshots.get_checkpoint(checkpoint_id) def get_session_checkpoint_for_source( self, root_frame_id: str, *, source_kind: str, source_id: str, ) -> dict | None: return self._session_snapshots.get_checkpoint_for_source( root_frame_id, source_kind=source_kind, source_id=source_id, ) def session_checkpoint_source_map( self, root_frame_id: str, *, source_kind: str ) -> dict[str, str]: return self._session_snapshots.checkpoint_source_map( root_frame_id, source_kind=source_kind, ) def list_session_checkpoints( self, root_frame_id: str, *, branch_id: str | None = None, limit: int = 100, ) -> list[dict]: return self._session_snapshots.list_checkpoints( root_frame_id, branch_id=branch_id, limit=limit, ) def retained_workspace_tree_ids(self) -> tuple[str, ...]: return self._session_snapshots.retained_tree_ids() def get_session_branch(self, branch_id: str) -> dict | None: return self._session_snapshots.get_branch(branch_id) def list_session_branches(self, root_frame_id: str) -> list[dict]: return self._session_snapshots.list_branches(root_frame_id) def ensure_active_session_branch(self, root_frame_id: str) -> str: return self._session_activation.ensure(root_frame_id) def active_session_branch(self, root_frame_id: str) -> str: return self._session_activation.current(root_frame_id) def session_export_guard(self, root_frame_id: str) -> dict: """Return one atomic branch/head/revert-marker export boundary.""" return self._session_activation.export_guard( root_frame_id, recovery_setting_key=revert_recovery_setting_key(root_frame_id), ) def activate_session_branch_checkpoint(self, **fields: Any) -> dict: return self._session_activation.activate_checkpoint(**fields) def record_snapshot_operation(self, **fields: Any) -> dict: return self._session_snapshots.record_operation(**fields) def get_snapshot_operation(self, operation_id: str) -> dict | None: return self._session_snapshots.get_operation(operation_id) def list_snapshot_operations( self, root_frame_id: str, *, branch_id: str | None = None, kind: str | None = None, status: str | None = None, limit: int = 100, ) -> list[dict]: return self._session_snapshots.list_operations( root_frame_id, branch_id=branch_id, kind=kind, status=status, limit=limit, ) # --- append-only Kernel recovery journal --------------------------- def append_recovery_event(self, **fields: Any) -> dict: return self._recovery_journal.append(**fields) def list_recovery_events( self, *, recovery_id: str | None = None, root_frame_id: str | None = None, branch_id: str | None = None, limit: int = 1000, newest: bool = False, ) -> list[dict]: return self._recovery_journal.list( recovery_id=recovery_id, root_frame_id=root_frame_id, branch_id=branch_id, limit=limit, newest=newest, ) # --- durable sub-agent delegation projection ---------------------- def restore_delegation_tree(self, **fields: Any) -> dict: return self._delegations.restore(**fields) def reserve_delegation_children(self, **fields: Any) -> dict: return self._delegations.reserve(**fields) def release_delegation_budget(self, **fields: Any) -> dict: return self._delegations.release(**fields) def persist_delegation_child(self, **fields: Any) -> dict | None: return self._delegations.persist_child(**fields) def continue_delegation_request(self, **fields: Any) -> dict: return self._delegations.continue_request(**fields) def delegation_request_for_child(self, **fields: Any) -> dict | None: return self._delegations.request_for_child(**fields) def delegation_tree(self, root_frame_id: str) -> dict: return self._delegations.project(root_frame_id) def delegation_budget(self, root_frame_id: str) -> dict | None: return self._delegations.budget(root_frame_id) def delete_frame(self, frame_id: str) -> dict[str, Any]: return self._frames.delete_frame(frame_id) def get_frame(self, frame_id: str) -> dict | None: return self._frames.get_frame(frame_id) def get_artifact(self, artifact_id: str) -> dict | None: return self._artifacts.get_artifact(artifact_id) def delete_artifact(self, artifact_id: str) -> list[str]: return self._artifacts.delete_artifact(artifact_id) def rename_artifact(self, artifact_id: str, filename: str) -> None: self._artifacts.rename_artifact(artifact_id, filename) def artifact_by_unique_filename( self, filename: str, root_frame_id: str | None = None ) -> dict | None: """A filename resolves only when it names exactly one artifact. Pass ``root_frame_id`` to ask within one frame (a report's siblings). """ return self._artifacts.artifact_by_unique_filename(filename, root_frame_id) def artifact_by_filename( self, filename: str, root_frame_id: str | None = None, *, strict: bool = False ) -> dict | None: return self._artifacts.artifact_by_filename( filename, root_frame_id, strict=strict, ) # --- artifacts ------------------------------------------------------- def _artifact_write_scope( self, *, frame_id: str | None, root_frame_id: str | None, project_id: str | None, ) -> tuple[bool, str | None, str]: return self._artifacts.artifact_write_scope( frame_id=frame_id, root_frame_id=root_frame_id, project_id=project_id, ) def save_artifact( self, *, path: str, filename: str, content_type: str | None, size_bytes: int, checksum: str | None, producing_cell_id: str | None = None, frame_id: str | None = None, root_frame_id: str | None = None, project_id: str | None = None, artifact_id: str | None = None, is_user_upload: bool = False, priority: int = 0, env_snapshot_id: str | None = None, snapshot_path: str | None = None, source: Any = None, ) -> dict: return self._artifacts.save_artifact( source=source, path=path, filename=filename, content_type=content_type, size_bytes=size_bytes, checksum=checksum, producing_cell_id=producing_cell_id, frame_id=frame_id, root_frame_id=root_frame_id, project_id=project_id, artifact_id=artifact_id, is_user_upload=is_user_upload, priority=priority, env_snapshot_id=env_snapshot_id, snapshot_path=snapshot_path, ) def record_cell_artifact( self, *, path: str, filename: str, content_type: str | None, size_bytes: int, checksum: str | None, producing_cell_id: str | None, frame_id: str | None, root_frame_id: str | None = None, project_id: str | None = None, env_snapshot_id: str | None = None, snapshot_path: str | None = None, source: Any = None, input_version_ids: list[str] | tuple[str, ...] | None = None, preserve_filename: bool = False, preserve_content_type: bool = False, reuse_policy: str = "any", reuse_matching_head: bool = False, ) -> dict: return self._artifacts.record_cell_artifact( path=path, filename=filename, content_type=content_type, size_bytes=size_bytes, checksum=checksum, producing_cell_id=producing_cell_id, frame_id=frame_id, root_frame_id=root_frame_id, project_id=project_id, env_snapshot_id=env_snapshot_id, snapshot_path=snapshot_path, source=source, input_version_ids=input_version_ids, preserve_filename=preserve_filename, preserve_content_type=preserve_content_type, reuse_policy=reuse_policy, reuse_matching_head=reuse_matching_head, ) def commit_artifact_upload(self, **fields) -> dict: """Thin facade for the upload repository's cross-store transaction.""" return self._artifacts.commit_artifact_upload(**fields) def artifact_by_scope_filename(self, *args, **kwargs) -> dict | None: """Thin facade for exact nullable-root upload lookup.""" return self._artifacts.artifact_by_scope_filename(*args, **kwargs) def rollback_artifact_upload(self, **fields) -> bool: """Thin facade used by durable upload-journal recovery.""" return self._artifacts.rollback_artifact_upload(**fields) def list_artifact_capture_observations( self, *, artifact_id: str | None = None, version_id: str | None = None, ) -> list[dict]: return self._artifacts.list_capture_observations( artifact_id=artifact_id, version_id=version_id, ) def artifact_capture_observation_cursor( self, *, root_frame_id: str | None = None, project_id: str | None = None, ) -> int: return self._artifacts.capture_observation_cursor( root_frame_id=root_frame_id, project_id=project_id, ) def artifact_capture_observations_since( self, cursor: int, *, root_frame_id: str | None, project_id: str, limit: int = 10_000, ) -> list[dict]: return self._artifacts.capture_observations_since( cursor, root_frame_id=root_frame_id, project_id=project_id, limit=limit, ) def record_artifact_restore( self, *, artifact_id: str, source_version_id: str, expected_latest_version_id: str, version_id: str, path: str, snapshot_path: str | None, size_bytes: int, checksum: str, frame_id: str | None, root_frame_id: str | None = None, project_id: str | None = None, publish=None, ) -> dict: return self._artifacts.record_artifact_restore( artifact_id=artifact_id, source_version_id=source_version_id, expected_latest_version_id=expected_latest_version_id, version_id=version_id, path=path, snapshot_path=snapshot_path, size_bytes=size_bytes, checksum=checksum, frame_id=frame_id, root_frame_id=root_frame_id, project_id=project_id, publish=publish, ) def materialise_artifact_version( self, *, source_version_id: str, artifact_id: str, version_id: str, filename: str, path: str, snapshot_path: str | None, frame_id: str | None, root_frame_id: str, project_id: str, producing_cell_id: str | None = None, publish=None, ) -> dict: return self._artifacts.materialise_artifact_version( source_version_id=source_version_id, artifact_id=artifact_id, version_id=version_id, filename=filename, path=path, snapshot_path=snapshot_path, frame_id=frame_id, root_frame_id=root_frame_id, project_id=project_id, producing_cell_id=producing_cell_id, publish=publish, ) def upsert_env_snapshot(self, snapshot: dict) -> str: return self._artifacts.upsert_env_snapshot(snapshot) def delete_env_snapshots_if_unreferenced(self, snapshot_ids) -> int: return self._artifacts.delete_env_snapshots_if_unreferenced(snapshot_ids) def get_env_snapshot(self, snapshot_id: str) -> dict | None: return self._artifacts.get_env_snapshot(snapshot_id) def env_snapshot_for_artifact( self, artifact_id: str, version_id: str | None = None ) -> dict | None: return self._artifacts.env_snapshot_for_artifact( artifact_id, version_id, ) def list_artifacts(self, filters: dict | None = None) -> list[dict]: return self._artifacts.list_artifacts(filters) def browse_artifacts( self, *, project_id: str, filename_query: str | None = None, content_type: str | None = None, origin: str | None = None, before: tuple[int, str] | None = None, limit: int = 50, visible_to_user_id: str | None = None, ) -> list[dict]: return self._artifacts.browse_artifacts( project_id=project_id, filename_query=filename_query, content_type=content_type, origin=origin, before=before, limit=limit, visible_to_user_id=visible_to_user_id, ) def list_artifact_names(self) -> list[dict]: return self._artifacts.list_artifact_names() def artifact_names_for_frame(self, frame_id: str) -> list[str]: return self._artifacts.artifact_names_for_frame(frame_id) def resolve_artifact_path(self, ident: str) -> str | None: return self._artifacts.resolve_artifact_path(ident) def version_for_path( self, path: str, *, root_frame_id: str | None, project_id: str ) -> str | None: return self._artifacts.version_for_path( path, root_frame_id=root_frame_id, project_id=project_id ) def version_meta(self, version_id: str) -> dict | None: return self._artifacts.version_meta(version_id) def set_version_source(self, version_id: str, source: Any) -> None: self._artifacts.set_version_source(version_id, source) def list_versions(self, artifact_id: str) -> list[dict]: return self._artifacts.list_versions(artifact_id) def update_version_path( self, version_id: str, path: str, size_bytes: int | None = None, checksum: str | None = None, ) -> None: self._artifacts.update_version_path( version_id, path, size_bytes=size_bytes, checksum=checksum, ) def set_version_snapshot(self, version_id: str, snapshot_path: str) -> None: self._artifacts.set_version_snapshot(version_id, snapshot_path) def set_priority(self, artifact_id: str, priority: int) -> dict | None: return self._artifacts.set_priority(artifact_id, priority) def set_latest_version(self, artifact_id: str, version_id: str) -> dict | None: return self._artifacts.set_latest_version(artifact_id, version_id) def add_lineage_edge( self, *, input_version_id: str, output_version_id: str, producing_cell_id: str | None = None, frame_id: str | None = None, ) -> None: self._artifacts.add_lineage_edge( input_version_id=input_version_id, output_version_id=output_version_id, producing_cell_id=producing_cell_id, frame_id=frame_id, ) def lineage_inputs( self, version_id: str, *, producing_cell_id: str | None = None, ) -> list[dict]: return self._artifacts.lineage_inputs( version_id, producing_cell_id=producing_cell_id, ) def lineage_edges_for(self, version_id: str, direction: str) -> list[dict]: return self._artifacts.lineage_edges_for(version_id, direction) def producing_cell_for_version(self, version_id: str) -> dict | None: return self._artifacts.producing_cell_for_version(version_id) # --- notes ----------------------------------------------------------- def add_note( self, *, project_id: str, content: str, title: str | None = None ) -> dict: return self._notes.add( project_id=project_id, content=content, title=title, ) def list_notes(self, project_id: str) -> list[dict]: return self._notes.list(project_id) def project_of_note(self, note_id: str) -> str | None: return self._notes.project_of(note_id) def delete_note(self, note_id: str) -> None: self._notes.delete(note_id) # --- settings (KV) --------------------------------------------------- def get_setting(self, key: str, default: str | None = None) -> str | None: return self._settings.get(key, default) def set_setting(self, key: str, value: str) -> None: self._settings.set(key, value) def delete_setting(self, key: str) -> None: self._settings.delete(key) def delete_setting_if_value(self, key: str, expected_value: str) -> bool: return self._settings.delete_if_value(key, expected_value) # --- web shares (public read-only snapshots) ------------------------- def get_share(self, share_id: str) -> dict | None: return self._shares.get(share_id) def active_share_for_frame(self, root_frame_id: str) -> dict | None: return self._shares.active_for_frame(root_frame_id) def list_shares_for_frame(self, root_frame_id: str) -> list[dict]: return self._shares.list_for_frame(root_frame_id) def list_shares(self, *, include_revoked: bool = False) -> list[dict]: return self._shares.list_all(include_revoked=include_revoked) def list_active_shares(self) -> list[dict]: return self._shares.list_active() def list_expired_shares(self, now_ms: int) -> list[dict]: return self._shares.list_expired(now_ms) def begin_share_publish( self, *, share_id: str, root_frame_id: str, title: str | None, pending_snapshot_id: str, expires_at: int | None = None, ) -> dict: return self._shares.begin_publish( share_id=share_id, root_frame_id=root_frame_id, title=title, pending_snapshot_id=pending_snapshot_id, expires_at=expires_at, ) def mark_share_ready( self, share_id: str, *, snapshot_id: str, bundle_sha256: str, bundle_size: int, projection_id: str, counts: dict | None = None, ) -> dict | None: return self._shares.mark_ready( share_id, snapshot_id=snapshot_id, bundle_sha256=bundle_sha256, bundle_size=bundle_size, projection_id=projection_id, counts=counts, ) def mark_share_failed(self, share_id: str) -> None: self._shares.mark_failed(share_id) def mark_share_revoked(self, share_id: str) -> None: self._shares.mark_revoked(share_id) def delete_share(self, share_id: str) -> None: self._shares.delete(share_id) # --- model profiles (saved LLM/API configs) -------------------------- # Stored as a JSON list under the `model_profiles` setting so users can keep # several full API configs (provider + base_url + model + key) side by side # and switch between them. Activating one writes the live `llm_*` settings. def list_model_profiles(self) -> list[dict]: return self._settings.list_model_profiles() def set_model_profiles(self, profiles: list[dict]) -> None: self._settings.set_model_profiles(profiles) def mutate_model_profiles(self, fn): return self._settings.mutate_model_profiles(fn) # --- permission rules (opencode-style tool-call gate) ---------------- def set_permission_rule( self, *, scope: str, scope_id: str = "", tool: str, pattern: str = "*", decision: str, ) -> str: return self._permissions.set_rule( scope=scope, scope_id=scope_id, tool=tool, pattern=pattern, decision=decision, ) def delete_permission_rule(self, rule_id: str) -> None: self._permissions.delete_rule(rule_id) def get_permission_rule(self, rule_id: str) -> dict | None: return self._permissions.get_rule(rule_id) def get_permission_rules(self, *, scope: str, scope_id: str = "") -> list[dict]: return self._permissions.get_rules(scope=scope, scope_id=scope_id) def list_permission_rules_for_frame( self, *, root_frame_id: str | None = None, project_id: str | None = None ) -> dict: return self._permissions.list_for_frame( root_frame_id=root_frame_id, project_id=project_id, ) def resolve_permission( self, *, root_frame_id: str | None = None, project_id: str | None = None, tool: str, pattern_input: str = "", ) -> str: return self._permissions.resolve( root_frame_id=root_frame_id, project_id=project_id, tool=tool, pattern_input=pattern_input, ) def seed_default_permission_rules(self, *, force: bool = False) -> None: self._permissions.seed_defaults(force=force) def create_permission_request(self, **request: Any) -> dict: return self._permissions.create_request(**request) def resolve_permission_request( self, decision_id: str, *, state: str, scope: str | None = None, pattern: str | None = None, message: str | None = None, resolution_context: str | None = None, continuation_required: bool = False, expected_action_digest: str | None = None, resolved_at: int | None = None, ) -> dict: return self._permissions.resolve_request( decision_id, state=state, scope=scope, pattern=pattern, message=message, resolution_context=resolution_context, continuation_required=continuation_required, expected_action_digest=expected_action_digest, resolved_at=resolved_at, ) def consume_restart_permission_grant( self, *, root_frame_id: str, tool: str, target: str = "", project_id: str | None = None, side_effect_class: str | None = None, resource_keys: list[str] | tuple[str, ...] | None = None, dangerous: bool = False, canonical_arguments: Any = None, consumed_at: int | None = None, ) -> dict | None: return self._permissions.consume_restart_once_grant( root_frame_id=root_frame_id, tool=tool, target=target, project_id=project_id, side_effect_class=side_effect_class, resource_keys=resource_keys, dangerous=dangerous, canonical_arguments=canonical_arguments, consumed_at=consumed_at, ) def activate_restart_permission_continuation( self, decision_id: str, *, expires_at: int | None = None, ) -> dict: return self._permissions.activate_restart_continuation( decision_id, expires_at=expires_at, ) def get_permission_request(self, decision_id: str) -> dict | None: return self._permissions.get_request(decision_id) def permission_request_action_digest(self, decision_id: str) -> str: return self._permissions.request_action_digest(decision_id) def list_permission_requests( self, *, root_frame_id: str | None = None, state: str | None = None, ) -> list[dict]: return self._permissions.list_requests( root_frame_id=root_frame_id, state=state, ) # --- plans (structured plan → approve → auto-execute) ---------------- def _plan_row(self, row) -> dict: return self._plans.normalize_row(row) def create_plan( self, *, frame_id: str, project_id: str = "default", title: str | None, rationale: str | None, confidence: str | None, steps: list[dict], artifact_id: str | None = None, status: str = "draft", ) -> dict: return self._plans.create( frame_id=frame_id, project_id=project_id, title=title, rationale=rationale, confidence=confidence, steps=steps, artifact_id=artifact_id, status=status, ) def get_plan(self, plan_id: str) -> dict | None: return self._plans.get(plan_id) def get_plan_by_frame(self, frame_id: str) -> dict | None: """The most recent (non-discarded) plan for a frame, else the newest.""" return self._plans.get_by_frame(frame_id) def pause_orphaned_executing_plans(self) -> int: """Startup reconciliation: no turn survives the process that ran it.""" return self._plans.pause_orphaned_executing() def list_plans(self, frame_id: str, *, limit: int = 50) -> list[dict]: return self._plans.list_for_frame(frame_id, limit=limit) def update_plan( self, plan_id: str, *, title: str | None = None, rationale: str | None = None, confidence: str | None = None, steps: list[dict] | None = None, status: str | None = None, step_status: dict | None = None, artifact_id: str | None = None, ) -> None: self._plans.update( plan_id, title=title, rationale=rationale, confidence=confidence, steps=steps, status=status, step_status=step_status, artifact_id=artifact_id, ) def compare_and_set_plan_status( self, plan_id: str, *, expected: str, new_status: str ) -> bool: """Claim a plan transition: True for the one caller that performed it. ``update_plan(status=...)`` writes whatever it is given, so a caller that checks the status first has already let go of the row by the time it writes. Anything that must happen once -- resuming a paused plan -- goes through here instead. """ return self._plans.compare_and_set_status( plan_id, expected=expected, new_status=new_status ) def set_plan_step_status( self, plan_id: str, step_id: str, status: str, note: str | None = None ) -> dict | None: """Merge one step's status into the plan's ``step_status`` JSON. Returns the updated plan **row**, not a folded view: ``steps[]`` still carries whatever the plan was created with, and each step's live status lives in the separate ``step_status`` map keyed by step id. The folding is done by ``server/plans.py::public_plan`` on the way to the client. This said "with steps[] status folded in", which is a description of a different function. A caller that believed it would read `steps[i] ["status"]`, find nothing, and conclude no step had progressed. """ return self._plans.set_step_status(plan_id, step_id, status, note) def delete_plans_for_frame(self, frame_id: str) -> None: self._plans.delete_for_frame(frame_id) # --- folders (session grouping within a project) -------------------- def create_folder(self, *, project_id: str, name: str) -> dict: return self._folders.create(project_id=project_id, name=name) def list_folders(self, project_id: str) -> list[dict]: return self._folders.list(project_id) def project_of_folder(self, folder_id: str) -> str | None: return self._folders.project_of(folder_id) def rename_folder(self, folder_id: str, name: str) -> None: self._folders.rename(folder_id, name) def delete_folder(self, folder_id: str) -> None: self._folders.delete(folder_id) def set_frame_folder(self, frame_id: str, folder_id: str | None) -> None: self._folders.set_frame_folder(frame_id, folder_id) # --- memories -------------------------------------------------------- def add_memory( self, *, content: str, block: str = "general", project_id: str = "default" ) -> dict: return self._memories.add( content=content, block=block, project_id=project_id, ) def update_memory( self, memory_id: str, *, content: str | None = None, block: str | None = None, project_id: str | None = None, ) -> dict | None: return self._memories.update( memory_id, content=content, block=block, project_id=project_id, ) def list_memories( self, project_id: str | None = None, block: str | None = None ) -> list[dict]: return self._memories.list(project_id=project_id, block=block) def resolve_memories( self, project_id: str | None = None, block: str | None = None ) -> dict: """`list_memories` plus how many items inheritance added or hid.""" return self._memories.resolve(project_id=project_id, block=block) def delete_memory(self, memory_id: str, project_id: str | None = None) -> bool: """Delete within one scope; True when a row went. The scope is required because an id-only delete crosses project boundaries silently.""" return self._memories.delete(memory_id, project_id=project_id) def memory_blocks(self, project_id: str | None = None) -> list[dict]: return self._memories.blocks(project_id) # --- feedback (per message) ----------------------------------------- def set_feedback(self, frame_id: str, key: str, rating: str | None) -> None: self._settings.set_feedback(frame_id, key, rating) def list_feedback(self, frame_id: str) -> dict: return self._settings.list_feedback(frame_id) # --- image annotations (figure review) ------------------------------ def add_annotation( self, *, root_frame_id: str, artifact_id: str, artifact_name: str | None, rel_x: float, rel_y: float, body: str, version_id: str | None = None, checksum: str | None = None, kind: str | None = None, locator: str | None = None, ) -> dict: return self._annotations.add( root_frame_id=root_frame_id, artifact_id=artifact_id, artifact_name=artifact_name, rel_x=rel_x, rel_y=rel_y, body=body, version_id=version_id, checksum=checksum, kind=kind, locator=locator, ) def get_annotation(self, annotation_id: str) -> dict | None: return self._annotations.get(annotation_id) def list_annotations( self, root_frame_id: str, *, artifact_id: str | None = None, status: str | None = None, ) -> list[dict]: return self._annotations.list_for_frame( root_frame_id, artifact_id=artifact_id, status=status, ) def update_annotation( self, annotation_id: str, *, body: str | None = None, status: str | None = None, expect_status: str | None = None, ) -> dict | None: return self._annotations.update( annotation_id, body=body, status=status, expect_status=expect_status, ) def mark_annotations_sent(self, annotation_ids: list[str]) -> None: self._annotations.mark_sent(annotation_ids) # --- admission ledger ------------------------------------------------- def reserve_with_admission( self, *, reservation_id: str, root_frame_id: str, annotation_ids: list[str], ) -> tuple[bool, list[dict]]: """Claim the id and the pins in ONE transaction. Two commits is two outcomes. Reserving and then recording separately means a ledger insert that fails leaves pins `reserved` with nothing to reconcile them against -- held forever, invisible in the composer, and with no row that a recovery pass could even find. So the ledger insert and the status change are one `BEGIN IMMEDIATE`, and either both happen or neither does. The ledger's PRIMARY KEY is what makes an id globally unique: a second request naming an existing id loses on the insert and gets nothing -- it does not coexist in another frame, and it does not overwrite the first request's row. `annotations(reservation_id, annotation_id)` could never have provided that, because `annotation_id` is already the primary key and the pair is unique for free. """ ids: list[str] = [] seen: set[str] = set() for annotation_id in annotation_ids or []: if ( type(annotation_id) is str and annotation_id and annotation_id not in seen ): seen.add(annotation_id) ids.append(annotation_id) now = _now_ms() with self._lock: self._conn.execute("BEGIN IMMEDIATE") try: self._conn.execute( "INSERT INTO annotation_admissions(reservation_id," "root_frame_id,annotation_ids,request_id,job_id,message_id," "state,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?)", ( reservation_id, root_frame_id, json.dumps(ids), None, None, None, "reserved", now, now, ), ) except sqlite3.IntegrityError: self._conn.rollback() return False, [] except BaseException: self._conn.rollback() raise try: claimed: list[dict] = [] if ids: placeholders = ",".join("?" * len(ids)) self._conn.execute( f"UPDATE annotations SET status='reserved', " f"reservation_id=?, updated_at={now} " f"WHERE root_frame_id=? AND status='open' " f"AND annotation_id IN ({placeholders})", (reservation_id, root_frame_id, *ids), ) claimed = [ dict(row) for row in self._conn.execute( "SELECT * FROM annotations WHERE reservation_id=? " "AND root_frame_id=? ORDER BY number", (reservation_id, root_frame_id), ).fetchall() ] # A claim that got nothing is not a live reservation. Left # `reserved`, it is a permanent row that recovery keeps # finding and a reconcile keeps reporting as in-flight -- # for pins this request never held. The concurrent loser is # the ordinary way to reach this. self._conn.execute( "UPDATE annotation_admissions SET annotation_ids=?, " "state=? WHERE reservation_id=?", ( json.dumps([r["annotation_id"] for r in claimed]), "reserved" if claimed else "released", reservation_id, ), ) else: self._conn.execute( "UPDATE annotation_admissions SET state='released' " "WHERE reservation_id=?", (reservation_id,), ) self._conn.commit() except BaseException: self._conn.rollback() raise return True, claimed def update_admission( self, reservation_id: str, *, root_frame_id: str, state: str | None = None, request_id: str | None = None, job_id: str | None = None, ) -> bool: """Advance an admission this frame owns. Scoped, so an id alone is not authority over somebody else's row.""" sets = ["updated_at=?"] params: list[Any] = [_now_ms()] for column, value in ( ("state", state), ("request_id", request_id), ("job_id", job_id), ): if value is not None: sets.append(f"{column}=?") params.append(value) params.extend([reservation_id, root_frame_id]) # A state change is a CAS on the non-terminal states; a # correlation-only write is not. `sent` and `released` are what the # turn did and no later caller gets to rewrite them -- but recording # *which* request and job an already-terminal admission belonged to is # exactly the correlation a lost 202 needs, so it stays unconditional. guard = " AND state IN ('reserved','pending')" if state is not None else "" with self._lock: cursor = self._conn.execute( f"UPDATE annotation_admissions SET {','.join(sets)} " f"WHERE reservation_id=? AND root_frame_id=?{guard}", tuple(params), ) self._conn.commit() return bool(cursor.rowcount) def abandon_admission( self, reservation_id: str, *, root_frame_id: str, job_id: str ) -> bool: """Undo an admission whose turn was correlated and then never started. Correlation is written before `Thread.start`, so that a client whose 202 was lost can tell an accepted turn from a refusal. That ordering opens its own hole at the other end: if the start then fails, plain release leaves `released` *with* a request and a job id -- the exact signature of accepted work, for a turn that never ran. A reconcile would report it as accepted and the client would not resend. A CAS on the job id, so this can only ever retract the turn it was called for: a start that succeeded owns its row and nothing here matches it. Pins and ledger move in one transaction, for the same reason every other terminal transition does. """ now = _now_ms() with self._lock: self._conn.execute("BEGIN IMMEDIATE") try: changed = self._conn.execute( "UPDATE annotation_admissions SET state='released', " "request_id=NULL, job_id=NULL, updated_at=? " "WHERE reservation_id=? AND root_frame_id=? AND job_id=? " "AND state IN ('reserved','pending','released')", (now, reservation_id, root_frame_id, job_id), ).rowcount if not changed: self._conn.rollback() return False self._conn.execute( "UPDATE annotations SET status='open', reservation_id=NULL, " "updated_at=? WHERE reservation_id=? AND root_frame_id=? " "AND status='reserved'", (now, reservation_id, root_frame_id), ) self._conn.commit() except BaseException: self._conn.rollback() raise return True def record_admission( self, *, reservation_id: str, root_frame_id: str, annotation_ids: list[str], request_id: str | None = None, job_id: str | None = None, message_id: str | None = None, state: str = "reserved", ) -> None: now = _now_ms() with self._lock: self._conn.execute( "INSERT INTO annotation_admissions(reservation_id,root_frame_id," "annotation_ids,request_id,job_id,message_id,state,created_at," "updated_at) VALUES(?,?,?,?,?,?,?,?,?) " "ON CONFLICT(reservation_id) DO UPDATE SET " "request_id=excluded.request_id, job_id=excluded.job_id, " "message_id=excluded.message_id, state=excluded.state, " "updated_at=excluded.updated_at", ( reservation_id, root_frame_id, json.dumps(list(annotation_ids)), request_id, job_id, message_id, state, now, now, ), ) self._conn.commit() def set_admission_state(self, reservation_id: str, state: str) -> None: with self._lock: self._conn.execute( "UPDATE annotation_admissions SET state=?, updated_at=? " "WHERE reservation_id=?", (state, _now_ms(), reservation_id), ) self._conn.commit() def get_admission( self, reservation_id: str, *, root_frame_id: str | None = None ) -> dict | None: sql = "SELECT * FROM annotation_admissions WHERE reservation_id=?" params: list[Any] = [reservation_id] if root_frame_id: sql += " AND root_frame_id=?" params.append(root_frame_id) with self._lock: row = self._conn.execute(sql, tuple(params)).fetchone() if row is None: return None record = dict(row) record["annotation_ids"] = json.loads(record["annotation_ids"] or "[]") return record def list_admissions(self, root_frame_id: str) -> list[dict]: with self._lock: rows = self._conn.execute( "SELECT * FROM annotation_admissions WHERE root_frame_id=? " "ORDER BY created_at", (root_frame_id,), ).fetchall() out = [] for row in rows: record = dict(row) record["annotation_ids"] = json.loads(record["annotation_ids"] or "[]") out.append(record) return out def recover_stranded_admissions(self) -> int: """Release reservations no live request can still be holding. A process that dies between reserve and finalize leaves `reserved` rows that nothing will ever release: they are neither sent nor available, and the comments are invisible in the composer forever. At startup no request is in flight by definition, so anything still `reserved` is stranded. """ recovered = 0 with self._lock: # Read, release and stamp under ONE write transaction, per row. # # This used to read every candidate, drop the lock, and then for # each one call `release` (its own transaction) followed by an # unconditional `set_admission_state(..., "released")`. Both halves # were wrong in the same direction. A live request can finalize # between the read and the release -- the read says `reserved`, the # turn sends, and the late recovery pass then stamps `released` # over `sent`. That is terminal evidence going backwards: the # message carried the comments and the ledger now says it did not, # so a client reconciling after a lost 202 is told to send them # again. # # `BEGIN IMMEDIATE` takes the write lock before the read, so the # candidate cannot move underneath it, and the stamp is a CAS on # the non-terminal states rather than an assignment. self._conn.execute("BEGIN IMMEDIATE") try: rows = self._conn.execute( "SELECT reservation_id, root_frame_id, annotation_ids FROM " "annotation_admissions WHERE state IN ('reserved','pending')" ).fetchall() for row in rows: reservation_id = row["reservation_id"] root = row["root_frame_id"] try: expected = set(json.loads(row["annotation_ids"] or "[]")) except ValueError: expected = set() held = { held_row["annotation_id"] for held_row in self._conn.execute( "SELECT annotation_id FROM annotations WHERE " "reservation_id=? AND status='reserved' " "AND root_frame_id=?", (reservation_id, root), ).fetchall() } # The ledger moves only when the exact set it names really # goes back. Anything else -- nothing held because a live # turn finalised them first, or a partial set because # something moved underneath -- is not this pass's news to # report, and stamping `released` over it would overwrite # the `sent` that turn just wrote. Terminal evidence that # can go backwards is not evidence. if not held or held != expected: continue freed = self._conn.execute( "UPDATE annotations SET status='open', " "reservation_id=NULL, updated_at=? " "WHERE reservation_id=? AND status='reserved' " "AND root_frame_id=?", (_now_ms(), reservation_id, root), ).rowcount if freed != len(held): continue self._conn.execute( "UPDATE annotation_admissions SET state='released', " "updated_at=? WHERE reservation_id=? " "AND state IN ('reserved','pending')", (_now_ms(), reservation_id), ) recovered += 1 self._conn.commit() except BaseException: self._conn.rollback() raise return recovered def reserve_annotations( self, *, root_frame_id: str, annotation_ids: list[str], reservation_id: str ) -> list[dict]: return self._annotations.reserve( root_frame_id=root_frame_id, annotation_ids=annotation_ids, reservation_id=reservation_id, ) def release_annotations( self, reservation_id: str, *, root_frame_id: str | None = None ) -> int: return self._annotations.release(reservation_id, root_frame_id=root_frame_id) def finalize_annotations_sent( self, reservation_id: str, *, expected_ids: list[str] | None = None, root_frame_id: str | None = None, request_id: str | None = None, job_id: str | None = None, ) -> bool: return self._annotations.finalize_sent( reservation_id, expected_ids=expected_ids, root_frame_id=root_frame_id, request_id=request_id, job_id=job_id, ) def annotation_is_reserved(self, annotation_id: str) -> bool: return self._annotations.is_reserved(annotation_id) def delete_annotation(self, annotation_id: str) -> None: self._annotations.delete(annotation_id) def delete_unreserved_annotation(self, annotation_id: str) -> bool: return self._annotations.delete_unreserved(annotation_id) # --- complete DataPro content index ------------------------------- def index_datapro_result( self, query: str, structured_content: Mapping[str, Any], frame_id: str | None = None, artifact_id: str | None = None, occurrence_id: str | None = None, source_content: Any | None = None, ) -> dict[str, Any]: """Atomically index every field returned by one successful search.""" scope = self.resolve_frame_scope(frame_id) return self._datapro_index.ingest( query=str(query), structured_content=structured_content, project_id=str(scope["project_id"] or "default"), root_frame_id=( str(scope["root_frame_id"]) if scope.get("root_frame_id") else None ), artifact_id=artifact_id, occurrence_id=occurrence_id, source_content=source_content, ) def link_datapro_index_artifact( self, batch_id: str, artifact_id: str | None ) -> dict[str, Any]: """Bind a batch to its saved result, refusing a dead Artifact. `ingest` commits the batch with `artifact_id` NULL, and the Artifact is created and linked afterwards. A delete landing in that window matched nothing -- the batch was not yet attributed to the Artifact -- and this UPDATE then pointed it at an id that no longer exists, resurrecting state the delete was supposed to remove. Nothing would collect it again, because `delete_for_artifact` only ever runs once per Artifact, so the batch stayed palette-visible forever. Linking to a missing Artifact therefore drops the batch instead: its owner is gone, so the index content has no lifecycle left to share. """ if artifact_id is not None and self.get_artifact(artifact_id) is None: self._datapro_index.delete_batch(batch_id) raise KeyError(f"no such artifact {artifact_id!r}") return self._datapro_index.link_artifact(batch_id, artifact_id) def search_datapro_index( self, query: str, limit: int = 20, frame_id: str | None = None, project_id: str | None = None, include_context: bool = False, ) -> dict[str, Any]: """Search literal DataPro text globally or within an explicit scope.""" root_frame_id: str | None = None resolved_project = project_id if frame_id is not None: scope = self.resolve_frame_scope( frame_id, fallback_project=project_id or "default" ) root_frame_id = ( str(scope["root_frame_id"]) if scope.get("root_frame_id") else None ) resolved_project = str(scope["project_id"] or "default") return self._datapro_index.search( query, limit=limit, project_id=resolved_project, root_frame_id=root_frame_id, include_context=include_context, ) def get_datapro_index_batch(self, batch_id: str) -> dict[str, Any] | None: return self._datapro_index.get_batch(batch_id) def delete_datapro_index_batch(self, batch_id: str) -> None: self._datapro_index.delete_batch(batch_id) # --- global search (command palette) -------------------------------- def search( self, query: str, limit: int = 20, *, visible_to_user_id: str | None = None ) -> dict: """Search sessions, artifacts, and indexed DataPro content for ⌘K. `visible_to_user_id` narrows the queries themselves. It used to be a post-filter in the route, applied *after* `LIMIT 20` -- so on a team where the twenty most recently updated matches belong to colleagues, every one of them was dropped and the caller was told their own session does not exist. `visible_session_clause` is the same rule the frame listings use, and it exists precisely because "filter after the read" turns a full page of hidden rows into a phantom empty result. """ q = f"%{query.strip()}%" frame_scope = artifact_scope = "" frame_params: list = [] artifact_params: list = [] if visible_to_user_id is not None: clause, frame_params = visible_session_clause(visible_to_user_id) frame_scope = " AND " + clause clause, artifact_params = visible_session_clause( visible_to_user_id, table="artifacts", session_expr="artifacts.root_frame_id", ) artifact_scope = " AND " + clause with self._lock: frames = self._conn.execute( "SELECT frame_id,project_id,name,task_summary,updated_at FROM frames " "WHERE parent_id IS NULL AND (name LIKE ? OR task_summary LIKE ?)" f"{frame_scope} " "ORDER BY updated_at DESC LIMIT ?", (q, q, *frame_params, limit), ).fetchall() arts = self._conn.execute( "SELECT artifact_id,filename,content_type,root_frame_id,project_id " f"FROM artifacts WHERE filename LIKE ?{artifact_scope} " "ORDER BY created_at DESC " "LIMIT ?", (q, *artifact_params, limit), ).fetchall() return { "sessions": [ { "id": r["frame_id"], "project_id": r["project_id"], "name": r["name"], "task_summary": r["task_summary"], } for r in frames ], "artifacts": [ { "id": r["artifact_id"], "filename": r["filename"], "content_type": r["content_type"], "root_frame_id": r["root_frame_id"], "project_id": r["project_id"], } for r in arts ], # Command-palette hits need the child record and Artifact link, not # a potentially-megabyte wrapper copied onto every matching child. "datapro": self.search_datapro_index( query, limit=limit, include_context=False )["items"], } # --- agents / specialists ------------------------------------------- def capability_state( self, *, project_id: str | None = None, session_id: str | None = None, ) -> CapabilityStateService: return self._capabilities.scoped( project_id=project_id, session_id=session_id, ) def skill_versions(self) -> SkillVersionRepository: """Return the Store-owned immutable Skill package repository.""" return self._skill_versions def set_capability_enabled( self, kind: str, name: str, enabled: bool, *, scope: str = "global", scope_id: str = "", metadata: dict | None = None, ) -> dict: return self._capabilities.set_enabled( kind, name, enabled, scope=scope, scope_id=scope_id, metadata=metadata, ) def capability_snapshot( self, kind: str, names, *, project_id: str | None = None, session_id: str | None = None, ) -> dict[str, dict]: return self.capability_state( project_id=project_id, session_id=session_id, ).snapshot(kind, names) def list_explicit_capability_states( self, kind: str | None = None, *, scope: str | None = None, scope_id: str | None = None, ) -> list[dict]: return self._capability_repository.explicit_states( kind, scope=scope, scope_id=scope_id, ) def list_agents( self, *, project_id: str | None = None, session_id: str | None = None, include_disabled: bool = False, ) -> list[dict]: return self.specialist_profiles( project_id=project_id, session_id=session_id, ).list(include_disabled=include_disabled) def specialist_profiles( self, *, project_id: str | None = None, session_id: str | None = None, ) -> SpecialistProfileService: """Return the shared resolver/filter seam for custom and built-ins.""" return self._specialists.scoped( project_id=project_id, session_id=session_id, ) def get_agent( self, name: str, *, project_id: str | None = None, session_id: str | None = None, include_disabled: bool = False, ) -> dict | None: return self.specialist_profiles( project_id=project_id, session_id=session_id, ).resolve(name, include_disabled=include_disabled) def upsert_agent( self, *, name: str, description: str = "", system_prompt: str = "", skill_names: list | None = None, connectors: list | None = None, unrestricted: bool = True, ) -> dict: return self._agents.upsert( name=name, description=description, system_prompt=system_prompt, skill_names=skill_names, connectors=connectors, unrestricted=unrestricted, ) def update_agent(self, name: str, **fields: Any) -> dict | None: """Partial update: only the supplied columns change. Returns None if the specialist does not exist.""" return self._agents.update(name, **fields) def delete_agent(self, name: str) -> None: self._agents.delete(name) # --- connectors (MCP servers) --------------------------------------- def list_connectors(self) -> list[dict]: return self._connectors.list() def get_connector(self, connector_id: str) -> dict | None: return self._connectors.get(connector_id) def upsert_connector( self, *, connector_id: str, name: str, command, description: str = "", args=None, env=None, enabled: bool = True, ) -> dict: # Brokered here rather than in the repository: this facade owns the # SecretBroker, and the repository must keep returning the real env to # the callers that launch the server. env = broker_connector_env(self, connector_id, env) return self._connectors.upsert( connector_id=connector_id, name=name, command=command, description=description, args=args, env=env, enabled=enabled, ) def patch_connector( self, connector_id: str, *, name=None, description=None, command=None, args=None, enabled=None, env_updates=None, remove_env=None, ) -> dict | None: """Update connector metadata without exposing or erasing env secrets. Public connector projections contain env names but never values. A Web editor therefore cannot round-trip a complete env mapping. Treat env as an explicit patch: absent names retain their broker references, supplied values replace one name, and ``remove_env`` deletes selected names. """ current = self._connectors.get(connector_id) if current is None: return None updates = env_updates if env_updates is not None else {} removals = remove_env if remove_env is not None else [] if not isinstance(updates, dict): raise ValueError("env_updates must be an object") if any( not isinstance(key, str) or not key or "=" in key or "\x00" in key for key in updates ): raise ValueError("env_updates names must be valid environment names") if any(value is None or "\x00" in str(value) for value in updates.values()): raise ValueError("env_updates values cannot be null or contain NUL") # A patch carries *new* values only -- absent names keep their existing # references. So a value that already looks like a broker reference did # not come from us: `broker_connector_env` passes `is_ref` text through # untouched, which would let a caller who cannot read any secret paste # another connector's reference into an env it also controls the # command of, and have it resolved at spawn. if any(is_ref(str(value)) for value in updates.values()): raise ValueError( "env_updates values must be literal values, not secret references" ) if not isinstance(removals, list) or any( not isinstance(item, str) or not item or "=" in item or "\x00" in item for item in removals ): raise ValueError("remove_env must contain valid environment names") if set(updates) & set(removals): raise ValueError("an env name cannot be updated and removed together") previous_env = current.get("env") merged_env = dict(previous_env) if isinstance(previous_env, dict) else {} brokered = broker_connector_env(self, connector_id, updates) retired: list[str] = [] for key in removals: old = merged_env.pop(key, None) if isinstance(old, str) and old: retired.append(old) for key, value in brokered.items(): old = merged_env.get(key) merged_env[key] = value if isinstance(old, str) and old and old != value: retired.append(old) updated = self._connectors.upsert( connector_id=connector_id, name=current["name"] if name is None else name, description=( current.get("description", "") if description is None else description ), command=current.get("command") if command is None else command, args=current.get("args") if args is None else args, env=merged_env, enabled=current.get("enabled", True) if enabled is None else bool(enabled), ) for value in retired: if is_ref(value) and value not in merged_env.values(): try: self.secrets.delete(value) except Exception: # noqa: BLE001 - the connector update succeeded pass return updated def set_connector_enabled(self, connector_id: str, enabled: bool) -> None: self._connectors.set_enabled(connector_id, enabled) # --- compute jobs ---------------------------------------------------- def create_compute_job(self, **kw) -> dict: return self._compute_jobs.create(**kw) def update_compute_job(self, job_id: str, **fields) -> dict | None: return self._compute_jobs.update(job_id, **fields) def get_compute_job(self, job_id: str) -> dict | None: return self._compute_jobs.get(job_id) def artifact_write_scope( self, *, frame_id: str | None = None, root_frame_id: str | None = None, project_id: str | None = None, ) -> tuple[bool, str | None, str]: """The scope a write *would* land in, resolved without writing. The repository has always had this and `save_artifact` calls it -- but by the time `save_artifact` runs, `ArtifactManager.upload` has already rewritten the live file, so a conflicting `project_id` refused *after* the previous version's bytes were gone. Nothing needed to be built; the resolution needed to be asked for first. Public so the upload path can. """ return self._artifacts.artifact_write_scope( frame_id=frame_id, root_frame_id=root_frame_id, project_id=project_id ) def compute_job_by_idempotency_key( self, key: str, owner_key: str | None = None, *, scoped: bool = True ) -> dict | None: return self._compute_jobs.by_idempotency_key(key, owner_key, scoped=scoped) def live_compute_jobs( self, owner_key: str | None = None, scoped: bool = False ) -> list[dict]: return self._compute_jobs.live(owner_key=owner_key, scoped=scoped) def list_compute_jobs(self, limit: int = 200) -> list[dict]: return self._compute_jobs.list(limit) def compute_jobs_for_owner( self, owner_key: str | None, limit: int = 200 ) -> list[dict]: """One owner's remote jobs, live and finished. Never installation-wide.""" return self._compute_jobs.for_owner(owner_key, limit) def append_compute_job_event(self, job_id: str, kind: str, payload=None) -> int: return self._compute_jobs.append_event(job_id, kind, payload) def compute_job_events(self, job_id: str) -> list[dict]: return self._compute_jobs.events(job_id) def delete_compute_job(self, job_id: str) -> None: self._compute_jobs.delete(job_id) def delete_connector(self, connector_id: str) -> None: # Drop the credentials with the row. Otherwise a connector the user # removed leaves its env secrets in the keychain with nothing left in # the app that refers to them. forget_connector_env(self, self._connectors.get(connector_id)) self._connectors.delete(connector_id) def connector_env(self, connector: dict) -> dict: """The env a connector's process is launched with, references resolved.""" return resolve_connector_env(self, connector) # --- compaction ------------------------------------------------------ def archive_compaction( self, *, frame_id: str | None, summary: str, compacted: list[dict], project_id: str = "default", **metadata: Any, ) -> str: return self._compactions.archive( frame_id=frame_id, summary=summary, compacted=compacted, project_id=project_id, **metadata, ) def list_compaction_archives(self, frame_id: str, *, limit: int = 50) -> list[dict]: return self._compactions.list(frame_id, limit=limit) # --- endpoints ---------------------------------------------- def upsert_endpoint(self, name: str, **fields: Any) -> None: self._endpoints.upsert(name, **fields) def list_endpoints(self) -> list[dict]: return self._endpoints.list() # --- host_call audit ---------------------------------------- def log_host_call( self, *, method: str, args: list, ok: bool, frame_id: str | None = None, result: Any = None, action_group_id: str | None = None, action_id: str | None = None, permission_decision_id: str | None = None, side_effect_class: str | None = None, resource_keys: list[str] | tuple[str, ...] | None = None, ) -> None: self._host_calls.log( method=method, args=args, ok=ok, frame_id=frame_id, result=result, action_group_id=action_group_id, action_id=action_id, permission_decision_id=permission_decision_id, side_effect_class=side_effect_class, resource_keys=resource_keys, ) def has_successful_bash_receipt( self, *, producing_cell_id: str, command_sha256: str, root_frame_id: str, branch_id: str, turn_id: str, ) -> bool: """Return the exact Host-authorized command receipt for code evidence.""" return self._host_calls.has_successful_bash_receipt( producing_cell_id=producing_cell_id, command_sha256=command_sha256, root_frame_id=root_frame_id, branch_id=branch_id, turn_id=turn_id, ) # --- generic read-only query (host.query backing) ------------------- def _refresh_scoped_views( self, conn: sqlite3.Connection, scope: Mapping[str, Any] ) -> None: """Publish the `my_*` views for one caller's scope, at most once per scope. The internal artifact tables cannot simply be denied: a bundled Skill legitimately reads `artifact_versions.source` to confirm the retrieval provenance it just attached. So the base tables are closed to direct access and reachable only through these views, which carry the caller's `root_frame_id`/`project_id` baked in as literals. Two things here are load-bearing, and the first version of this method got both wrong. `executescript` is not used, and this is a correctness point rather than a performance one: it issues an implicit COMMIT before running its script, so every `host.query` ended whatever transaction the caller had open. On a database that holds an audit ledger, a read that commits someone else's half-finished write is not a read. `test_agent_sql_does_not_commit_the_ callers_transaction` is the check. The definitions are also cached against the scope that produced them, rather than rebuilt per query. Measured on this machine that is 0.090 ms -> 0.005 ms per query, 17x on this path -- worth having and not more than that. It is recorded because the number is small: an earlier note here blamed a slow test suite on it, which was wrong, and 0.09 ms per query could not have done that. The scope values are internal ids the caller never supplies -- they come from `resolve_frame_scope` -- and they are quoted anyway, because a value interpolated into DDL is worth quoting whatever its provenance. """ root = str(scope.get("root_frame_id") or "") project = str(scope.get("project_id") or "") if getattr(self, "_view_scope", None) == (root, project): return root_sql = _sql_quote(root) project_sql = _sql_quote(project) statements = ( f"""CREATE TEMP VIEW my_artifacts AS SELECT * FROM main.artifacts WHERE root_frame_id = {root_sql} AND project_id = {project_sql}""", f"""CREATE TEMP VIEW my_artifact_versions AS SELECT v.* FROM main.artifact_versions v JOIN main.artifacts a ON a.artifact_id = v.artifact_id WHERE a.root_frame_id = {root_sql} AND a.project_id = {project_sql}""", f"""CREATE TEMP VIEW my_artifact_capture_observations AS SELECT o.* FROM main.artifact_capture_observations o JOIN main.artifacts a ON a.artifact_id = o.artifact_id WHERE a.root_frame_id = {root_sql} AND a.project_id = {project_sql}""", f"""CREATE TEMP VIEW my_lineage_edges AS SELECT e.* FROM main.lineage_edges e JOIN main.artifact_versions v ON v.version_id = e.output_version_id JOIN main.artifacts a ON a.artifact_id = v.artifact_id WHERE a.root_frame_id = {root_sql} AND a.project_id = {project_sql}""", f"""CREATE TEMP VIEW my_frames AS SELECT * FROM main.frames WHERE frame_id = {root_sql} OR root_frame_id = {root_sql}""", f"""CREATE TEMP VIEW my_env_snapshots AS SELECT s.* FROM main.env_snapshots s JOIN main.frames f ON f.frame_id = s.frame_id WHERE f.frame_id = {root_sql} OR f.root_frame_id = {root_sql}""", # Both tables carry `root_frame_id` themselves, so this is a # filter and not a join. The join was tried first and returned # nothing: a message's `frame_id` is nullable (a turn's messages # are keyed by the root), so joining on it dropped every row -- # a scoped view that is always empty is indistinguishable from a # working guard until somebody asserts the owner can still read. f"""CREATE TEMP VIEW my_messages AS SELECT * FROM main.messages WHERE root_frame_id = {root_sql}""", f"""CREATE TEMP VIEW my_execution_log AS SELECT * FROM main.execution_log WHERE root_frame_id = {root_sql}""", ) for name in _SCOPED_VIEWS: conn.execute(f"DROP VIEW IF EXISTS temp.{name}") for statement in statements: conn.execute(statement) self._view_scope = (root, project) def _query_view_only(self) -> frozenset[str]: """Which base tables `host.query` may reach only through a view. Team mode adds the conversation/execution family. Off, the set is byte-identical to what it has always been, which is what keeps a single-user install's documented `SELECT * FROM frames` working (INV-1). Read from the environment rather than from a Config because `Store` is constructed in places that have no Config, and a guard that silently loosened when its config was unavailable would be the wrong failure. Reading the env is fine; *re-implementing the truthiness rule* was not. This used to accept only ("1","true","yes","on") while `Config.team_mode` accepts everything outside ("0","false","no","off",""), so `OPENAI4S_TEAM_MODE=enabled` -- or `y`, or `2` -- booted a daemon with login forced and ownership filtering on while this guard returned the single-user set and `host.query` read every member's prompts. One rule, one place. """ from openai4s.config import _env_flag if _env_flag("OPENAI4S_TEAM_MODE", False): return _VIEW_ONLY_TABLES | _TEAM_VIEW_ONLY_TABLES return _VIEW_ONLY_TABLES def query( self, sql: str, params: list | None = None, limit: int | None = None, timeout_s: float = 5.0, scope: Mapping[str, Any] | None = None, ) -> list[dict]: """Run a read-only SELECT/CTE under a real SQLite authorizer. The authorizer is the enforcement; the text checks below are kept as a cheap first refusal with a clearer message. That ordering matters: the text checks were previously the *only* enforcement, and they cannot see a table named in a bound parameter, spelled `"artifacts"`, `[artifacts]` or `main.artifacts`, or reached through `pragma_table_list`. SQLite hands the authorizer the resolved table name after parsing, so none of those spellings are different to it. `scope`, when supplied, publishes the `my_*` views for that caller. It used to be accepted by the SDK and dropped on the floor here. """ lowered = sql.lower() deny_scan = _strip_sql_literals(lowered) bad = _DENY_WORD_RE.search(deny_scan) if bad is not None: raise PermissionError(f"host.query: table '{bad.group(0)}' is not readable") shadow = _cte_shadow_name(sql) if shadow is not None: # See `_cte_shadow_name`: naming a CTE after a scoped view makes # SQLite hand the authorizer that name as the view responsible for # the read, which is how the escape hatch below came to trust a # string the caller wrote. raise PermissionError( f"host.query: {shadow!r} is a scoped view and cannot " f"be used as a CTE name" ) stripped = lowered.lstrip() if not (stripped.startswith("select") or stripped.startswith("with")): raise ValueError("host.query only allows read-only SELECT/CTE") # The daemon's one connection, with the authorizer installed for exactly # the duration of this statement and removed in `finally` -- the same # shape as the existing `set_progress_handler` bracket directly below. # # A separate `mode=ro` connection was tried first and reverted: it would # add a second connection lifetime, a second lock discipline and # multi-process interaction to a compatibility facade, for defence in # depth the authorizer already provides. The authorizer denies every # action code that is not a read or a plain SELECT, so writes are refused # by rule and not by keyword. conn = self._conn with self._lock: guard = _QueryAuthorizer( view_only=self._query_view_only(), published_views=_SCOPED_VIEWS if scope else frozenset(), ) try: # Views first, with the guard off: creating them is a privileged # setup step, not part of the caller's statement. _clear_authorizer(conn) if scope: self._refresh_scoped_views(conn, scope) conn.set_authorizer(guard) conn.set_progress_handler(_TimeoutGuard(timeout_s), 10000) cur = conn.execute(sql, tuple(params or ())) rows = cur.fetchmany(limit) if limit else cur.fetchall() except sqlite3.DatabaseError as error: if guard.denied: raise PermissionError( f"host.query: {', '.join(guard.denied)} is not readable " f"from agent SQL. Scoped rows are available through " f"my_artifacts, my_artifact_versions and " f"my_lineage_edges." ) from error raise finally: conn.set_progress_handler(None, 10000) _clear_authorizer(conn) return [dict(r) for r in rows] def schema(self) -> dict[str, list[str]]: with self._lock: tables = self._conn.execute( "SELECT name FROM sqlite_master WHERE type='table'" ).fetchall() out: dict[str, list[str]] = {} for t in tables: name = t["name"] if ( name in QUERY_DENYLIST or name.startswith("sqlite_") or name in {"datapro_index_batches", "datapro_index_entries"} ): continue cols = self._conn.execute(f"PRAGMA table_info({name})").fetchall() out[name] = [c["name"] for c in cols] return out class _TimeoutGuard: """Progress-handler callback that aborts a query after timeout_s (5s).""" def __init__(self, timeout_s: float): self._deadline = time.time() + timeout_s def __call__(self) -> int: return 1 if time.time() > self._deadline else 0 _STORES: dict[str, Store] = {} _STORES_LOCK = threading.Lock() def _discard_store(store: Store) -> None: """Remove a closed singleton without evicting a newer replacement.""" key = str(store.db_path.resolve()) with _STORES_LOCK: if _STORES.get(key) is store: _STORES.pop(key, None) def get_store(db_path: Path) -> Store: """Process-wide singleton Store per db path.""" key = str(Path(db_path).resolve()) with _STORES_LOCK: st = _STORES.get(key) if st is None or st._closed: st = Store(Path(db_path)) _STORES[key] = st return st