diff --git a/adapters/chatgpt/metadata/chatgpt-adapter.json b/adapters/chatgpt/metadata/chatgpt-adapter.json index e630be8..9fc1849 100644 --- a/adapters/chatgpt/metadata/chatgpt-adapter.json +++ b/adapters/chatgpt/metadata/chatgpt-adapter.json @@ -7,7 +7,7 @@ "adapter_path": "adapters/chatgpt/selective-intelligence", "transformation": "runtime_only_with_nested_role_entrypoints_as_role_references", "single_skill_entrypoint": "SKILL.md", - "runtime_file_count": 59, + "runtime_file_count": 61, "role_path_map": { "subskills/si-aligner/SKILL.md": "subskills/si-aligner/ROLE.md", "subskills/si-intake/SKILL.md": "subskills/si-intake/ROLE.md", @@ -18,7 +18,7 @@ "subskills/si-worker/SKILL.md": "subskills/si-worker/ROLE.md" }, "behavioral_contract": "source_preserved_not_installed_behavior_proof", - "source_manifest_sha256": "e1a569d7f2c6cccae1d652b81011100db01790ae3168da9d43a5d2a23e05abc7", + "source_manifest_sha256": "09aac47c8c11717defd8f837391ec4126e7e62696813dd428e7eeb14180f89bf", "source_file_sha256": { "SKILL.md": "afb4c6e096aa83683105c45b5525ac34ee436654be4dacc9c2a44f01bf2033b1", "VERSION": "326dd5d8641f399990de42def50ac29eb4c9222601c0a0a3fc3c947dd3bec65d", @@ -31,7 +31,8 @@ "references/activation-and-adoption.md": "04df2423ec610be344366dd646b427e3e9062f90244b0e4143a6fb02218b5f74", "references/actual-intent-alignment.md": "3bc1fac79f7fed56c098c5dd01ce7c5e49a973c9a2455606f1ff7c495373a9ad", "references/architecture-reuse.md": "5d1c4d09ca8e7407d19c7fc9d68587b2ea0b61be61e5c3448890af2c8766138d", - "references/continuity-and-impact.md": "78f93ccaaed14e58544c5a0c78b6bf933922b4c576a597b13d63ac74d0de53e0", + "references/continuity-and-impact.md": "ffd9f6b96147ba77d251854f4b3fd2ffbb3cd4b9deccf98b8fcfac849fd109f7", + "references/postgres-session-storage.md": "db0d16851060a6563366b85e37657554b4108475d78f30b8215bb570b02c8dce", "references/durable-progress-and-recovery.md": "1457eed72572aa6e0f34ab6773b81735e3f1ee22e5427e44a238619b034a609d", "references/evidence-and-completion.md": "f05424d0e1b8c68624aab05d3e03361e312a08a92bc3b3338b86e29a1971569f", "references/failure-patterns-and-gates.md": "90b4b2fc15069a33cdef3fc39d653b718f4ef4e27345c0db5dab3d7fca5408b5", @@ -65,8 +66,9 @@ "scripts/feedback.py": "90bfdc574d03b630ea5e68f22185c848d37ab93e3edca768b0d2b501f7924bbb", "scripts/intent_contract.py": "a3bcb84336179e5dc9e18f55c7f02c27bbe777a116aa0942bee807ea82eb58a6", "scripts/lane_registry.py": "f9417ae2ac0a3911220c75b9f83cbe5204a30142f948b54611df6e4712846e35", - "scripts/lane_session.py": "d20f57cc7c00d84b4e91399d2c2051b130f6f5144331907206b01efd085b17a8", + "scripts/lane_session.py": "1119e62d04825221ef8c823bdc9e6ac8ac351458715a5e4e40932fb899bbf89d", "scripts/policy_guard.py": "bc8be4706a053114973371ce9bfeb94a13881e3fa42779ac4abea550a1229bbe", + "scripts/postgres_sessions.py": "e19ea9cffa6872b62d2870c59e4e08a10f231c2ecfab77018bf0e46341c3b297", "scripts/progress_checkpoint.py": "3660538c309ff0c01f87e615037e83a7cb8bd5fc651f66f22569da4cf599e0aa", "scripts/project_index.py": "496eb8e316fa7baf85c3c0b35ead76053d41c842846367163a5eadb7e94b755c", "scripts/prompt_queue.py": "1f5423659c07bcf842454931e28a0547c3020e19d5afeb58fee5adf6fa1c4f33", @@ -92,7 +94,8 @@ "references/activation-and-adoption.md": "04df2423ec610be344366dd646b427e3e9062f90244b0e4143a6fb02218b5f74", "references/actual-intent-alignment.md": "3bc1fac79f7fed56c098c5dd01ce7c5e49a973c9a2455606f1ff7c495373a9ad", "references/architecture-reuse.md": "5d1c4d09ca8e7407d19c7fc9d68587b2ea0b61be61e5c3448890af2c8766138d", - "references/continuity-and-impact.md": "78f93ccaaed14e58544c5a0c78b6bf933922b4c576a597b13d63ac74d0de53e0", + "references/continuity-and-impact.md": "ffd9f6b96147ba77d251854f4b3fd2ffbb3cd4b9deccf98b8fcfac849fd109f7", + "references/postgres-session-storage.md": "db0d16851060a6563366b85e37657554b4108475d78f30b8215bb570b02c8dce", "references/durable-progress-and-recovery.md": "1457eed72572aa6e0f34ab6773b81735e3f1ee22e5427e44a238619b034a609d", "references/evidence-and-completion.md": "f05424d0e1b8c68624aab05d3e03361e312a08a92bc3b3338b86e29a1971569f", "references/failure-patterns-and-gates.md": "90b4b2fc15069a33cdef3fc39d653b718f4ef4e27345c0db5dab3d7fca5408b5", @@ -126,8 +129,9 @@ "scripts/feedback.py": "90bfdc574d03b630ea5e68f22185c848d37ab93e3edca768b0d2b501f7924bbb", "scripts/intent_contract.py": "a3bcb84336179e5dc9e18f55c7f02c27bbe777a116aa0942bee807ea82eb58a6", "scripts/lane_registry.py": "f9417ae2ac0a3911220c75b9f83cbe5204a30142f948b54611df6e4712846e35", - "scripts/lane_session.py": "d20f57cc7c00d84b4e91399d2c2051b130f6f5144331907206b01efd085b17a8", + "scripts/lane_session.py": "1119e62d04825221ef8c823bdc9e6ac8ac351458715a5e4e40932fb899bbf89d", "scripts/policy_guard.py": "bc8be4706a053114973371ce9bfeb94a13881e3fa42779ac4abea550a1229bbe", + "scripts/postgres_sessions.py": "e19ea9cffa6872b62d2870c59e4e08a10f231c2ecfab77018bf0e46341c3b297", "scripts/progress_checkpoint.py": "3660538c309ff0c01f87e615037e83a7cb8bd5fc651f66f22569da4cf599e0aa", "scripts/project_index.py": "496eb8e316fa7baf85c3c0b35ead76053d41c842846367163a5eadb7e94b755c", "scripts/prompt_queue.py": "1f5423659c07bcf842454931e28a0547c3020e19d5afeb58fee5adf6fa1c4f33", @@ -155,6 +159,7 @@ "scripts/lane_registry.py", "scripts/lane_session.py", "scripts/policy_guard.py", + "scripts/postgres_sessions.py", "scripts/text_gate.py" ] } diff --git a/adapters/chatgpt/selective-intelligence/references/continuity-and-impact.md b/adapters/chatgpt/selective-intelligence/references/continuity-and-impact.md index 1432374..c7c41eb 100644 --- a/adapters/chatgpt/selective-intelligence/references/continuity-and-impact.md +++ b/adapters/chatgpt/selective-intelligence/references/continuity-and-impact.md @@ -2,6 +2,8 @@ Use this reference when work spans sessions, agents, branches, worktrees, repositories, interrupted runs, or more than one build. The purpose is to preserve one product truth while allowing safe progress. +For native SI workers that need a shared durable session store, see [optional PostgreSQL session persistence](postgres-session-storage.md). This is an opt-in persistence adapter; it does not replace SI governance or move workspace artifacts. + ## Contents - [Resume from evidence](#resume-from-evidence) diff --git a/adapters/chatgpt/selective-intelligence/references/postgres-session-storage.md b/adapters/chatgpt/selective-intelligence/references/postgres-session-storage.md new file mode 100644 index 0000000..300e34d --- /dev/null +++ b/adapters/chatgpt/selective-intelligence/references/postgres-session-storage.md @@ -0,0 +1,83 @@ +# Optional shared session persistence + +Use this for shared checkpoints with **one active executor per workspace and +session**. A replacement process may recover only after the previous executor +is confirmed terminated. The existing file backend remains the default and +requires no database, API key, paid service, or Python dependency. + +SI retains interpretation, approval, task, receipt, and verification ownership. +PostgreSQL stores the same validated JSON session and monotonic revision. This +does not turn another product, gateway, or model into an SI engine. + +## Configure a native SI worker + +Install the optional `psycopg[binary]>=3.2,<4` dependency in that worker's Python +environment. Set these server-side variables using the deployment secret store: + +| Variable | Value | +| --- | --- | +| `SI_SESSION_BACKEND` | `postgres` | +| `SI_SESSION_EXECUTION_MODE` | `single-active`; explicitly acknowledge the execution prerequisite below | +| `SI_SESSION_DATABASE_URL` | Direct PostgreSQL connection URI for the approved database | +| `SI_SESSION_NAMESPACE` | Explicit project identifier, such as `si-worker-preview` | + +Use a direct Neon endpoint, **not** its `-pooler` transaction-pooling endpoint. +The engine holds a session advisory lock while a governed operation runs. Each +checkpoint commits independently so it survives a killed worker. After confirming +that worker has terminated, a replacement acquires the lock and reads the +persisted revision. Remote +connections require `sslmode=verify-full`; system trust roots are the default. + +Initialize the table explicitly on the intended isolated database: + +```bash +python scripts/lane_session.py initialize-store +``` + +This creates `public.si_runtime_sessions`. Normal session reads/writes never +provision a database or create a table. Give the runtime role only the database +and table privileges it needs; use a separate migration identity if appropriate. +Namespaces prevent accidental record collisions, **not** access by another +holder of the same database credentials. Separate product trust boundaries need +separate roles/databases or an independently enforced authorization layer. + +The existing engine, CLI and MCP callers continue through `lane_session`; no +second checkpoint API or duplicated orchestration engine is introduced. Keep +workspaces and artifact evidence accessible at the paths recorded in the session. +Shared session storage alone does not move local files between machines. + +## Recovery contract + +- A database advisory lock serializes cooperative checkpoint updates while the + connection is alive. It is not a worker lease or a scheduler. +- Stale revisions cannot overwrite newer approval, correction, or evidence. +- A missing store, lost connection, invalid configuration, or database error + fails closed; it never creates a local replacement session. +- Process death releases its database lock. Completed evidence remains durable. +- A network failure can release that lock while the old executor still runs. + PostgreSQL revision checks do not fence workspace writes, rollback, or external + tools. **Do not automatically launch a replacement on lock release, connection + loss, or timeout. Confirm the old executor has terminated first.** The explicit + `single-active` setting acknowledges this prerequisite; it does not enforce + process termination or make active-active execution safe. +- Database durability is not exactly-once execution of arbitrary external tools. + Preserve existing pending/unknown receipts, reconciliation and idempotency + rules before retrying an interrupted external action. +- Selecting `file` again does not migrate sessions back. Do not change a live + worker's backend or namespace as a recovery shortcut. + +## Verify before adopting + +Set `SI_TEST_POSTGRES_URL` to a **disposable local or approved isolated test** +database and run the PostgreSQL session tests. They create the session table and +use temporary unique namespaces. Never supply a customer production database. + +```bash +python -B -m unittest discover -s tests -p test_postgres_sessions.py +``` + +Those tests cover process restart, lock release after process death, concurrent +writers, stale revisions, namespace separation and recovery of real engine +verification evidence. Local Postgres proof does not establish a Neon +deployment, installed client adoption, cross-machine workspace availability or +provider-side spending limits. Record those states separately. diff --git a/adapters/chatgpt/selective-intelligence/scripts/lane_session.py b/adapters/chatgpt/selective-intelligence/scripts/lane_session.py index f47dd7c..4b344f9 100644 --- a/adapters/chatgpt/selective-intelligence/scripts/lane_session.py +++ b/adapters/chatgpt/selective-intelligence/scripts/lane_session.py @@ -16,6 +16,7 @@ from intent_contract import classify_intent, concept_tokens, merge_active_contract import checkpoint as CP +from postgres_sessions import configured_store try: import lane_registry as reg # type: ignore @@ -72,9 +73,13 @@ def sessions_dir() -> Path: return directory.resolve() -def session_path(session_id: str) -> Path: +def _validate_session_id(session_id: str) -> None: if not session_id or any(ch not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_" for ch in session_id): raise ValueError("invalid session id") + + +def session_path(session_id: str) -> Path: + _validate_session_id(session_id) return sessions_dir() / f"{session_id}.session.json" @@ -89,11 +94,17 @@ class SessionConflictError(RuntimeError): @contextmanager def session_lock(session_id: str): - """Serialize cooperative SI writers across threads and local processes. + """Serialize cooperative SI writers through the configured storage lock. Reentrant in one thread. Locks are advisory and do not authenticate hostile - filesystem writers or provide distributed/network-filesystem transactions. + filesystem writers or fence external actions after a lost database lock. """ + _validate_session_id(session_id) + store = configured_store() + if store is not None: + with store.lock(session_id): + yield + return path = session_path(session_id).with_suffix(".lock") key = str(path) with _THREAD_LOCKS_GUARD: @@ -140,12 +151,13 @@ def _revision(session: dict[str, Any]) -> int: def save_session(session: dict[str, Any]) -> None: - """Compare revisions under the session lock, then atomically replace JSON. + """Compare revisions under the session lock, then atomically persist JSON. Missing revisions on historical records start at zero; saving upgrades storage only and never restores checkpoint or task execution authority. """ - target = session_path(session["sessionId"]) + _validate_session_id(session["sessionId"]) + store = configured_store() expected = _revision(session) with session_lock(session["sessionId"]): current = load_session(session["sessionId"]) @@ -158,6 +170,13 @@ def save_session(session: dict[str, Any]) -> None: staged["updatedAt"] = _now() staged["persistenceRevision"] = actual + 1 encoded = json.dumps(staged, indent=2, sort_keys=True, allow_nan=False) + if store is not None: + if not store.save(session["sessionId"], expected, staged): + raise SessionConflictError("session changed since it was loaded; reload before saving") + session["updatedAt"] = staged["updatedAt"] + session["persistenceRevision"] = staged["persistenceRevision"] + return + target = session_path(session["sessionId"]) temp = target.with_suffix(f".{uuid.uuid4().hex}.tmp") try: with temp.open("x", encoding="utf-8") as handle: @@ -174,19 +193,26 @@ def save_session(session: dict[str, Any]) -> None: def load_session(session_id: str) -> dict[str, Any] | None: try: - path = session_path(session_id) + _validate_session_id(session_id) except ValueError: return None - if path.is_symlink(): - raise ValueError("session file must not be a symlink") - try: - data = json.loads(path.read_text(encoding="utf-8")) - except FileNotFoundError: - return None + store = configured_store() + if store is not None: + data = store.load(session_id) + if data is None: + return None + else: + path = session_path(session_id) + if path.is_symlink(): + raise ValueError("session file must not be a symlink") + try: + data = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return None if not isinstance(data, dict) or data.get("schemaVersion") != SCHEMA_VERSION: raise ValueError("unsupported session schema") if data.get("sessionId") != session_id: - raise ValueError("saved session identity differs from its file") + raise ValueError("saved session identity differs from its storage key") _revision(data) return data @@ -688,7 +714,15 @@ def main() -> int: show.add_argument("--session", required=True) ready = sub.add_parser("ready") ready.add_argument("--session", required=True) + sub.add_parser("initialize-store", help="Create the optional Postgres session table on the configured database") args = parser.parse_args() + if args.command == "initialize-store": + store = configured_store() + if store is None: + parser.error("initialize-store requires SI_SESSION_BACKEND=postgres") + store.initialize() + print(json.dumps({"initialized": True, "backend": "postgres", "namespace": store.namespace})) + return 0 session = load_session(args.session) if not session: print(json.dumps({"error": "session not found"})) diff --git a/adapters/chatgpt/selective-intelligence/scripts/postgres_sessions.py b/adapters/chatgpt/selective-intelligence/scripts/postgres_sessions.py new file mode 100644 index 0000000..a5ee356 --- /dev/null +++ b/adapters/chatgpt/selective-intelligence/scripts/postgres_sessions.py @@ -0,0 +1,172 @@ +"""Optional PostgreSQL persistence for the canonical SI session owner. + +No SI decisions live here. JSON state and revision checks remain in lane_session. +Use a direct connection: session advisory locks cannot use transaction pooling. +""" +from __future__ import annotations + +from contextlib import contextmanager +from functools import lru_cache +import hashlib +import json +import os +import re +import threading +from typing import Any + + +class SessionStorageError(RuntimeError): + """A configured store is unavailable; never silently switch persistence.""" + + +_HELD = threading.local() +_TABLE = "public.si_runtime_sessions" + + +class PostgresSessionStore: + def __init__(self, database_url: str, namespace: str): + if not database_url: + raise SessionStorageError("SI_SESSION_DATABASE_URL is required for postgres sessions") + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,99}", namespace): + raise SessionStorageError("SI_SESSION_NAMESPACE must be an explicit 1-100 character project identifier") + try: + import psycopg + from psycopg.conninfo import conninfo_to_dict + except ImportError: + raise SessionStorageError("Postgres sessions require the optional psycopg[binary] dependency") from None + try: + options = conninfo_to_dict(database_url) + except psycopg.Error: + raise SessionStorageError("Invalid SI_SESSION_DATABASE_URL") from None + host = options.get("host", "").lower() + if not host or "," in host or "-pooler." in host: + raise SessionStorageError("Postgres sessions require one explicit direct database host, without transaction pooling") + local = host in {"localhost", "127.0.0.1", "::1"} or host.startswith("/") + if not local: + if options.get("sslmode", "verify-full") != "verify-full": + raise SessionStorageError("Remote session databases require sslmode=verify-full") + options["sslmode"] = "verify-full" + options.setdefault("sslrootcert", "system") + options.setdefault("connect_timeout", "10") + options["application_name"] = "selective-intelligence-sessions" + self._driver = psycopg + self._options = options + self.namespace = namespace + self._identity = hashlib.sha256(database_url.encode()).hexdigest() + + def _key(self, session_id: str) -> tuple: + # Forked children must never reuse an inherited connection/lock record. + return (os.getpid(), self._identity, self.namespace, session_id) + + def _connect(self): + connection = None + try: + connection = self._driver.connect(**self._options, autocommit=True) + connection.execute("SET statement_timeout = '30s'") + connection.execute("SET lock_timeout = '30s'") + return connection + except self._driver.Error: + if connection is not None: + connection.close() + raise SessionStorageError("Cannot connect to the configured Postgres session store") from None + + def _execute(self, connection, statement: str, values=()): + try: + return connection.execute(statement, values) + except self._driver.Error: + raise SessionStorageError("Postgres session operation failed; no local fallback was used") from None + + @contextmanager + def lock(self, session_id: str): + held = getattr(_HELD, "connections", None) + if held is None: + held = _HELD.connections = {} + key = self._key(session_id) + if key in held: + connection = held[key] + if connection.closed or connection.broken: + raise SessionStorageError("Postgres session lock was lost; stop and recover before retrying") + yield connection + return + connection = self._connect() + lock_id = int.from_bytes(hashlib.sha256((self.namespace + "\0" + session_id).encode()).digest()[:8], "big", signed=True) + try: + self._execute(connection, "SELECT pg_advisory_lock(%s::bigint)", (lock_id,)) + held[key] = connection + try: + yield connection + finally: + del held[key] + finally: + # Closing releases the session lock even after an error. No reconnect + # here: a replacement connection would no longer hold this lock. + connection.close() + + def initialize(self) -> None: + """Explicit schema setup, never an implicit side effect of a read.""" + connection = self._connect() + try: + self._execute(connection, f""" + CREATE TABLE IF NOT EXISTS {_TABLE} ( + namespace text NOT NULL, + session_id text NOT NULL, + revision bigint NOT NULL CHECK (revision > 0), + payload jsonb NOT NULL, + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (namespace, session_id), + CHECK (jsonb_typeof(payload) = 'object'), + CHECK (payload->>'sessionId' = session_id), + CHECK ((payload->>'persistenceRevision')::bigint = revision) + ) + """) + finally: + connection.close() + + def load(self, session_id: str) -> dict[str, Any] | None: + with self.lock(session_id) as connection: + row = self._execute(connection, + f"SELECT revision, payload FROM {_TABLE} WHERE namespace = %s AND session_id = %s", + (self.namespace, session_id)).fetchone() + if row is None: + return None + revision, payload = row + if not isinstance(payload, dict) or payload.get("persistenceRevision") != revision: + raise SessionStorageError("Postgres session payload and stored revision disagree") + return payload + + def save(self, session_id: str, expected: int, staged: dict[str, Any]) -> bool: + encoded = json.dumps(staged, sort_keys=True, allow_nan=False) + with self.lock(session_id) as connection: + if expected == 0: + cursor = self._execute(connection, f""" + INSERT INTO {_TABLE} (namespace, session_id, revision, payload) + VALUES (%s, %s, 1, %s::jsonb) + ON CONFLICT (namespace, session_id) DO NOTHING + RETURNING revision + """, (self.namespace, session_id, encoded)) + else: + cursor = self._execute(connection, f""" + UPDATE {_TABLE} SET revision = %s, payload = %s::jsonb, updated_at = now() + WHERE namespace = %s AND session_id = %s AND revision = %s + RETURNING revision + """, (expected + 1, encoded, self.namespace, session_id, expected)) + return cursor.fetchone() is not None + + +@lru_cache(maxsize=8) +def _store(database_url: str, namespace: str) -> PostgresSessionStore: + return PostgresSessionStore(database_url, namespace) + + +def configured_store() -> PostgresSessionStore | None: + backend = os.environ.get("SI_SESSION_BACKEND", "file").strip().lower() + if backend == "file": + return None + if backend != "postgres": + raise SessionStorageError("SI_SESSION_BACKEND must be file or postgres") + if os.environ.get("SI_SESSION_EXECUTION_MODE") != "single-active": + raise SessionStorageError( + "Postgres checkpoint storage requires SI_SESSION_EXECUTION_MODE=single-active; " + "confirm the prior executor has stopped before recovery. Automatic takeover is unsupported" + ) + return _store(os.environ.get("SI_SESSION_DATABASE_URL", ""), os.environ.get("SI_SESSION_NAMESPACE", "")) diff --git a/skills/selective-intelligence/metadata/distribution.json b/skills/selective-intelligence/metadata/distribution.json index 8c59d39..76652ce 100644 --- a/skills/selective-intelligence/metadata/distribution.json +++ b/skills/selective-intelligence/metadata/distribution.json @@ -155,6 +155,7 @@ "references/actual-intent-alignment.md", "references/architecture-reuse.md", "references/continuity-and-impact.md", + "references/postgres-session-storage.md", "references/durable-progress-and-recovery.md", "references/evidence-and-completion.md", "references/failure-patterns-and-gates.md", @@ -190,6 +191,7 @@ "scripts/lane_registry.py", "scripts/lane_session.py", "scripts/policy_guard.py", + "scripts/postgres_sessions.py", "scripts/progress_checkpoint.py", "scripts/project_index.py", "scripts/prompt_queue.py", @@ -238,6 +240,7 @@ "references/application-patterns.md", "references/architecture-reuse.md", "references/continuity-and-impact.md", + "references/postgres-session-storage.md", "references/correction-harvesting.md", "references/distribution-and-discoverability.md", "references/durable-progress-and-recovery.md", @@ -283,6 +286,7 @@ "scripts/lane_registry.py", "scripts/lane_session.py", "scripts/policy_guard.py", + "scripts/postgres_sessions.py", "scripts/progress_checkpoint.py", "scripts/project_index.py", "scripts/prompt_queue.py", @@ -316,6 +320,7 @@ "tests/test_product_model_execution_integrity.py", "tests/test_product_model_plan_import.py", "tests/test_product_model_transactions.py", + "tests/test_postgres_sessions.py", "tests/test_product_model_worker_binding.py", "tests/test_product_model_handoff_completeness.py", "tests/test_product_model_repair_handoff.py", diff --git a/skills/selective-intelligence/references/continuity-and-impact.md b/skills/selective-intelligence/references/continuity-and-impact.md index 1432374..c7c41eb 100644 --- a/skills/selective-intelligence/references/continuity-and-impact.md +++ b/skills/selective-intelligence/references/continuity-and-impact.md @@ -2,6 +2,8 @@ Use this reference when work spans sessions, agents, branches, worktrees, repositories, interrupted runs, or more than one build. The purpose is to preserve one product truth while allowing safe progress. +For native SI workers that need a shared durable session store, see [optional PostgreSQL session persistence](postgres-session-storage.md). This is an opt-in persistence adapter; it does not replace SI governance or move workspace artifacts. + ## Contents - [Resume from evidence](#resume-from-evidence) diff --git a/skills/selective-intelligence/references/postgres-session-storage.md b/skills/selective-intelligence/references/postgres-session-storage.md new file mode 100644 index 0000000..300e34d --- /dev/null +++ b/skills/selective-intelligence/references/postgres-session-storage.md @@ -0,0 +1,83 @@ +# Optional shared session persistence + +Use this for shared checkpoints with **one active executor per workspace and +session**. A replacement process may recover only after the previous executor +is confirmed terminated. The existing file backend remains the default and +requires no database, API key, paid service, or Python dependency. + +SI retains interpretation, approval, task, receipt, and verification ownership. +PostgreSQL stores the same validated JSON session and monotonic revision. This +does not turn another product, gateway, or model into an SI engine. + +## Configure a native SI worker + +Install the optional `psycopg[binary]>=3.2,<4` dependency in that worker's Python +environment. Set these server-side variables using the deployment secret store: + +| Variable | Value | +| --- | --- | +| `SI_SESSION_BACKEND` | `postgres` | +| `SI_SESSION_EXECUTION_MODE` | `single-active`; explicitly acknowledge the execution prerequisite below | +| `SI_SESSION_DATABASE_URL` | Direct PostgreSQL connection URI for the approved database | +| `SI_SESSION_NAMESPACE` | Explicit project identifier, such as `si-worker-preview` | + +Use a direct Neon endpoint, **not** its `-pooler` transaction-pooling endpoint. +The engine holds a session advisory lock while a governed operation runs. Each +checkpoint commits independently so it survives a killed worker. After confirming +that worker has terminated, a replacement acquires the lock and reads the +persisted revision. Remote +connections require `sslmode=verify-full`; system trust roots are the default. + +Initialize the table explicitly on the intended isolated database: + +```bash +python scripts/lane_session.py initialize-store +``` + +This creates `public.si_runtime_sessions`. Normal session reads/writes never +provision a database or create a table. Give the runtime role only the database +and table privileges it needs; use a separate migration identity if appropriate. +Namespaces prevent accidental record collisions, **not** access by another +holder of the same database credentials. Separate product trust boundaries need +separate roles/databases or an independently enforced authorization layer. + +The existing engine, CLI and MCP callers continue through `lane_session`; no +second checkpoint API or duplicated orchestration engine is introduced. Keep +workspaces and artifact evidence accessible at the paths recorded in the session. +Shared session storage alone does not move local files between machines. + +## Recovery contract + +- A database advisory lock serializes cooperative checkpoint updates while the + connection is alive. It is not a worker lease or a scheduler. +- Stale revisions cannot overwrite newer approval, correction, or evidence. +- A missing store, lost connection, invalid configuration, or database error + fails closed; it never creates a local replacement session. +- Process death releases its database lock. Completed evidence remains durable. +- A network failure can release that lock while the old executor still runs. + PostgreSQL revision checks do not fence workspace writes, rollback, or external + tools. **Do not automatically launch a replacement on lock release, connection + loss, or timeout. Confirm the old executor has terminated first.** The explicit + `single-active` setting acknowledges this prerequisite; it does not enforce + process termination or make active-active execution safe. +- Database durability is not exactly-once execution of arbitrary external tools. + Preserve existing pending/unknown receipts, reconciliation and idempotency + rules before retrying an interrupted external action. +- Selecting `file` again does not migrate sessions back. Do not change a live + worker's backend or namespace as a recovery shortcut. + +## Verify before adopting + +Set `SI_TEST_POSTGRES_URL` to a **disposable local or approved isolated test** +database and run the PostgreSQL session tests. They create the session table and +use temporary unique namespaces. Never supply a customer production database. + +```bash +python -B -m unittest discover -s tests -p test_postgres_sessions.py +``` + +Those tests cover process restart, lock release after process death, concurrent +writers, stale revisions, namespace separation and recovery of real engine +verification evidence. Local Postgres proof does not establish a Neon +deployment, installed client adoption, cross-machine workspace availability or +provider-side spending limits. Record those states separately. diff --git a/skills/selective-intelligence/scripts/lane_session.py b/skills/selective-intelligence/scripts/lane_session.py index f47dd7c..4b344f9 100644 --- a/skills/selective-intelligence/scripts/lane_session.py +++ b/skills/selective-intelligence/scripts/lane_session.py @@ -16,6 +16,7 @@ from intent_contract import classify_intent, concept_tokens, merge_active_contract import checkpoint as CP +from postgres_sessions import configured_store try: import lane_registry as reg # type: ignore @@ -72,9 +73,13 @@ def sessions_dir() -> Path: return directory.resolve() -def session_path(session_id: str) -> Path: +def _validate_session_id(session_id: str) -> None: if not session_id or any(ch not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_" for ch in session_id): raise ValueError("invalid session id") + + +def session_path(session_id: str) -> Path: + _validate_session_id(session_id) return sessions_dir() / f"{session_id}.session.json" @@ -89,11 +94,17 @@ class SessionConflictError(RuntimeError): @contextmanager def session_lock(session_id: str): - """Serialize cooperative SI writers across threads and local processes. + """Serialize cooperative SI writers through the configured storage lock. Reentrant in one thread. Locks are advisory and do not authenticate hostile - filesystem writers or provide distributed/network-filesystem transactions. + filesystem writers or fence external actions after a lost database lock. """ + _validate_session_id(session_id) + store = configured_store() + if store is not None: + with store.lock(session_id): + yield + return path = session_path(session_id).with_suffix(".lock") key = str(path) with _THREAD_LOCKS_GUARD: @@ -140,12 +151,13 @@ def _revision(session: dict[str, Any]) -> int: def save_session(session: dict[str, Any]) -> None: - """Compare revisions under the session lock, then atomically replace JSON. + """Compare revisions under the session lock, then atomically persist JSON. Missing revisions on historical records start at zero; saving upgrades storage only and never restores checkpoint or task execution authority. """ - target = session_path(session["sessionId"]) + _validate_session_id(session["sessionId"]) + store = configured_store() expected = _revision(session) with session_lock(session["sessionId"]): current = load_session(session["sessionId"]) @@ -158,6 +170,13 @@ def save_session(session: dict[str, Any]) -> None: staged["updatedAt"] = _now() staged["persistenceRevision"] = actual + 1 encoded = json.dumps(staged, indent=2, sort_keys=True, allow_nan=False) + if store is not None: + if not store.save(session["sessionId"], expected, staged): + raise SessionConflictError("session changed since it was loaded; reload before saving") + session["updatedAt"] = staged["updatedAt"] + session["persistenceRevision"] = staged["persistenceRevision"] + return + target = session_path(session["sessionId"]) temp = target.with_suffix(f".{uuid.uuid4().hex}.tmp") try: with temp.open("x", encoding="utf-8") as handle: @@ -174,19 +193,26 @@ def save_session(session: dict[str, Any]) -> None: def load_session(session_id: str) -> dict[str, Any] | None: try: - path = session_path(session_id) + _validate_session_id(session_id) except ValueError: return None - if path.is_symlink(): - raise ValueError("session file must not be a symlink") - try: - data = json.loads(path.read_text(encoding="utf-8")) - except FileNotFoundError: - return None + store = configured_store() + if store is not None: + data = store.load(session_id) + if data is None: + return None + else: + path = session_path(session_id) + if path.is_symlink(): + raise ValueError("session file must not be a symlink") + try: + data = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return None if not isinstance(data, dict) or data.get("schemaVersion") != SCHEMA_VERSION: raise ValueError("unsupported session schema") if data.get("sessionId") != session_id: - raise ValueError("saved session identity differs from its file") + raise ValueError("saved session identity differs from its storage key") _revision(data) return data @@ -688,7 +714,15 @@ def main() -> int: show.add_argument("--session", required=True) ready = sub.add_parser("ready") ready.add_argument("--session", required=True) + sub.add_parser("initialize-store", help="Create the optional Postgres session table on the configured database") args = parser.parse_args() + if args.command == "initialize-store": + store = configured_store() + if store is None: + parser.error("initialize-store requires SI_SESSION_BACKEND=postgres") + store.initialize() + print(json.dumps({"initialized": True, "backend": "postgres", "namespace": store.namespace})) + return 0 session = load_session(args.session) if not session: print(json.dumps({"error": "session not found"})) diff --git a/skills/selective-intelligence/scripts/postgres_sessions.py b/skills/selective-intelligence/scripts/postgres_sessions.py new file mode 100644 index 0000000..a5ee356 --- /dev/null +++ b/skills/selective-intelligence/scripts/postgres_sessions.py @@ -0,0 +1,172 @@ +"""Optional PostgreSQL persistence for the canonical SI session owner. + +No SI decisions live here. JSON state and revision checks remain in lane_session. +Use a direct connection: session advisory locks cannot use transaction pooling. +""" +from __future__ import annotations + +from contextlib import contextmanager +from functools import lru_cache +import hashlib +import json +import os +import re +import threading +from typing import Any + + +class SessionStorageError(RuntimeError): + """A configured store is unavailable; never silently switch persistence.""" + + +_HELD = threading.local() +_TABLE = "public.si_runtime_sessions" + + +class PostgresSessionStore: + def __init__(self, database_url: str, namespace: str): + if not database_url: + raise SessionStorageError("SI_SESSION_DATABASE_URL is required for postgres sessions") + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,99}", namespace): + raise SessionStorageError("SI_SESSION_NAMESPACE must be an explicit 1-100 character project identifier") + try: + import psycopg + from psycopg.conninfo import conninfo_to_dict + except ImportError: + raise SessionStorageError("Postgres sessions require the optional psycopg[binary] dependency") from None + try: + options = conninfo_to_dict(database_url) + except psycopg.Error: + raise SessionStorageError("Invalid SI_SESSION_DATABASE_URL") from None + host = options.get("host", "").lower() + if not host or "," in host or "-pooler." in host: + raise SessionStorageError("Postgres sessions require one explicit direct database host, without transaction pooling") + local = host in {"localhost", "127.0.0.1", "::1"} or host.startswith("/") + if not local: + if options.get("sslmode", "verify-full") != "verify-full": + raise SessionStorageError("Remote session databases require sslmode=verify-full") + options["sslmode"] = "verify-full" + options.setdefault("sslrootcert", "system") + options.setdefault("connect_timeout", "10") + options["application_name"] = "selective-intelligence-sessions" + self._driver = psycopg + self._options = options + self.namespace = namespace + self._identity = hashlib.sha256(database_url.encode()).hexdigest() + + def _key(self, session_id: str) -> tuple: + # Forked children must never reuse an inherited connection/lock record. + return (os.getpid(), self._identity, self.namespace, session_id) + + def _connect(self): + connection = None + try: + connection = self._driver.connect(**self._options, autocommit=True) + connection.execute("SET statement_timeout = '30s'") + connection.execute("SET lock_timeout = '30s'") + return connection + except self._driver.Error: + if connection is not None: + connection.close() + raise SessionStorageError("Cannot connect to the configured Postgres session store") from None + + def _execute(self, connection, statement: str, values=()): + try: + return connection.execute(statement, values) + except self._driver.Error: + raise SessionStorageError("Postgres session operation failed; no local fallback was used") from None + + @contextmanager + def lock(self, session_id: str): + held = getattr(_HELD, "connections", None) + if held is None: + held = _HELD.connections = {} + key = self._key(session_id) + if key in held: + connection = held[key] + if connection.closed or connection.broken: + raise SessionStorageError("Postgres session lock was lost; stop and recover before retrying") + yield connection + return + connection = self._connect() + lock_id = int.from_bytes(hashlib.sha256((self.namespace + "\0" + session_id).encode()).digest()[:8], "big", signed=True) + try: + self._execute(connection, "SELECT pg_advisory_lock(%s::bigint)", (lock_id,)) + held[key] = connection + try: + yield connection + finally: + del held[key] + finally: + # Closing releases the session lock even after an error. No reconnect + # here: a replacement connection would no longer hold this lock. + connection.close() + + def initialize(self) -> None: + """Explicit schema setup, never an implicit side effect of a read.""" + connection = self._connect() + try: + self._execute(connection, f""" + CREATE TABLE IF NOT EXISTS {_TABLE} ( + namespace text NOT NULL, + session_id text NOT NULL, + revision bigint NOT NULL CHECK (revision > 0), + payload jsonb NOT NULL, + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (namespace, session_id), + CHECK (jsonb_typeof(payload) = 'object'), + CHECK (payload->>'sessionId' = session_id), + CHECK ((payload->>'persistenceRevision')::bigint = revision) + ) + """) + finally: + connection.close() + + def load(self, session_id: str) -> dict[str, Any] | None: + with self.lock(session_id) as connection: + row = self._execute(connection, + f"SELECT revision, payload FROM {_TABLE} WHERE namespace = %s AND session_id = %s", + (self.namespace, session_id)).fetchone() + if row is None: + return None + revision, payload = row + if not isinstance(payload, dict) or payload.get("persistenceRevision") != revision: + raise SessionStorageError("Postgres session payload and stored revision disagree") + return payload + + def save(self, session_id: str, expected: int, staged: dict[str, Any]) -> bool: + encoded = json.dumps(staged, sort_keys=True, allow_nan=False) + with self.lock(session_id) as connection: + if expected == 0: + cursor = self._execute(connection, f""" + INSERT INTO {_TABLE} (namespace, session_id, revision, payload) + VALUES (%s, %s, 1, %s::jsonb) + ON CONFLICT (namespace, session_id) DO NOTHING + RETURNING revision + """, (self.namespace, session_id, encoded)) + else: + cursor = self._execute(connection, f""" + UPDATE {_TABLE} SET revision = %s, payload = %s::jsonb, updated_at = now() + WHERE namespace = %s AND session_id = %s AND revision = %s + RETURNING revision + """, (expected + 1, encoded, self.namespace, session_id, expected)) + return cursor.fetchone() is not None + + +@lru_cache(maxsize=8) +def _store(database_url: str, namespace: str) -> PostgresSessionStore: + return PostgresSessionStore(database_url, namespace) + + +def configured_store() -> PostgresSessionStore | None: + backend = os.environ.get("SI_SESSION_BACKEND", "file").strip().lower() + if backend == "file": + return None + if backend != "postgres": + raise SessionStorageError("SI_SESSION_BACKEND must be file or postgres") + if os.environ.get("SI_SESSION_EXECUTION_MODE") != "single-active": + raise SessionStorageError( + "Postgres checkpoint storage requires SI_SESSION_EXECUTION_MODE=single-active; " + "confirm the prior executor has stopped before recovery. Automatic takeover is unsupported" + ) + return _store(os.environ.get("SI_SESSION_DATABASE_URL", ""), os.environ.get("SI_SESSION_NAMESPACE", "")) diff --git a/skills/selective-intelligence/tests/test_postgres_sessions.py b/skills/selective-intelligence/tests/test_postgres_sessions.py new file mode 100644 index 0000000..a15a78d --- /dev/null +++ b/skills/selective-intelligence/tests/test_postgres_sessions.py @@ -0,0 +1,235 @@ +"""Optional-store boundaries and opt-in real PostgreSQL recovery tests. + +The integration class requires SI_TEST_POSTGRES_URL for a disposable database. +An absent database is reported as skipped, never simulated as a native pass. +""" +from __future__ import annotations + +import copy +import importlib.util +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import time +import unittest +from unittest import mock +import uuid + +SCRIPTS = Path(__file__).resolve().parents[1] / "scripts" +if str(SCRIPTS) not in sys.path: + sys.path.insert(0, str(SCRIPTS)) +import build_engine as E +import checkpoint as CP +import lane_session as LS +import postgres_sessions as PG + +HAS_DRIVER = importlib.util.find_spec("psycopg") is not None +TEST_URL = os.environ.get("SI_TEST_POSTGRES_URL") + + +class StoreConfigurationTests(unittest.TestCase): + def test_default_needs_no_driver_or_database(self): + with mock.patch.dict(os.environ, {}, clear=True): + self.assertIsNone(PG.configured_store()) + + def test_unknown_backend_cannot_fall_back_to_files(self): + with mock.patch.dict(os.environ, {"SI_SESSION_BACKEND": "postgrse"}): + with self.assertRaisesRegex(PG.SessionStorageError, "must be file or postgres"): + LS.load_session("si-test") + + def test_database_is_explicit(self): + with self.assertRaisesRegex(PG.SessionStorageError, "DATABASE_URL is required"): + PG.PostgresSessionStore("", "test") + + def test_postgres_requires_single_active_execution_contract(self): + for mode in ("", "active-active", "automatic-failover"): + with self.subTest(mode=mode): + with mock.patch.dict(os.environ, {"SI_SESSION_BACKEND": "postgres", "SI_SESSION_EXECUTION_MODE": mode}): + with self.assertRaisesRegex(PG.SessionStorageError, "Automatic takeover is unsupported"): + LS.load_session("si-test") + + def test_namespace_is_explicit_and_restricted(self): + for namespace in ("", "../other", "a b", "a" * 101): + with self.subTest(namespace=namespace): + with self.assertRaisesRegex(PG.SessionStorageError, "NAMESPACE"): + PG.PostgresSessionStore("postgresql://localhost/test", namespace) + + @unittest.skipUnless(HAS_DRIVER, "optional psycopg dependency is not installed") + def test_remote_database_requires_verified_tls(self): + for mode in ("disable", "prefer", "require", "verify-ca"): + with self.subTest(mode=mode): + with self.assertRaisesRegex(PG.SessionStorageError, "verify-full"): + PG.PostgresSessionStore(f"postgresql://db.example/test?sslmode={mode}", "test") + store = PG.PostgresSessionStore("postgresql://db.example/test", "test") + self.assertEqual(store._options["sslmode"], "verify-full") + self.assertEqual(store._options["sslrootcert"], "system") + + @unittest.skipUnless(HAS_DRIVER, "optional psycopg dependency is not installed") + def test_transaction_pooler_or_implicit_host_is_rejected(self): + for url in ("postgresql://ep-test-pooler.us-east-1.aws.neon.tech/test", + "postgresql://EP-TEST-POOLER.us-east-1.aws.neon.tech/test", + "dbname=test", "host=first,second dbname=test"): + with self.subTest(url=url): + with self.assertRaisesRegex(PG.SessionStorageError, "direct database host"): + PG.PostgresSessionStore(url, "test") + + +@unittest.skipUnless(TEST_URL and HAS_DRIVER, "requires SI_TEST_POSTGRES_URL and optional psycopg") +class PostgresRecoveryTests(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory(prefix="si-postgres-") + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name) + self.workspace = self.root / "work" + self.workspace.mkdir() + self.namespace = "test-" + uuid.uuid4().hex + environment = mock.patch.dict(os.environ, { + "SI_SESSION_BACKEND": "postgres", + "SI_SESSION_EXECUTION_MODE": "single-active", + "SI_SESSION_DATABASE_URL": TEST_URL, + "SI_SESSION_NAMESPACE": self.namespace, + "SI_SESSION_DIR": str(self.root / "must-not-be-used"), + }) + environment.start() + self.addCleanup(environment.stop) + self.store = PG.configured_store() + self.store.initialize() + self.addCleanup(self.clean_rows) + self.session = LS.new_session("Preserve approved membership access.", workspace=str(self.workspace)) + CP.approve_checkpoint(self.session, self.session["currentCheckpointId"]) + self.task = LS.add_task(self.session, title="Preserve membership access", queue="ready", metadata={"kind": "worker"}) + LS.save_session(self.session) + self.sid = self.session["sessionId"] + + def clean_rows(self): + connection = self.store._connect() + try: + connection.execute("DELETE FROM public.si_runtime_sessions WHERE namespace = %s", (self.namespace,)) + connection.execute("DELETE FROM public.si_runtime_sessions WHERE namespace = %s", (self.namespace + "-other",)) + finally: + connection.close() + + def command(self, code, *args): + return [sys.executable, "-B", "-c", "import sys;sys.path.insert(0,sys.argv[1]);" + code, + str(SCRIPTS), *args] + + def child(self, code, *args): + result = subprocess.run(self.command(code, *args), capture_output=True, text=True, timeout=40) + self.assertEqual(result.returncode, 0, result.stderr) + return json.loads(result.stdout) + + def test_stale_revision_and_insert_collision_are_rejected(self): + stale = copy.deepcopy(self.session) + LS.record_event(self.session, "newer", {}) + LS.save_session(self.session) + with self.assertRaises(LS.SessionConflictError): + LS.save_session(stale) + stale["persistenceRevision"] = 0 + with self.assertRaises(LS.SessionConflictError): + LS.save_session(stale) + saved = LS.load_session(self.sid) + self.assertEqual(saved["persistenceRevision"], 2) + self.assertEqual(saved["events"][-1]["eventType"], "newer") + + def test_namespaces_separate_identical_session_ids(self): + with mock.patch.dict(os.environ, {"SI_SESSION_NAMESPACE": self.namespace + "-other"}): + self.assertIsNone(LS.load_session(self.sid)) + other = copy.deepcopy(self.session) + other["persistenceRevision"] = 0 + other["objective"] = "Other project." + LS.save_session(other) + self.assertEqual(LS.load_session(self.sid)["objective"], "Other project.") + self.assertEqual(LS.load_session(self.sid)["objective"], self.session["objective"]) + + def test_concurrent_processes_preserve_all_checkpoints(self): + code = """ +import lane_session as LS +with LS.session_lock(sys.argv[2]): + session = LS.load_session(sys.argv[2]) + LS.record_event(session, 'child-checkpoint', {'child': sys.argv[3]}) + LS.save_session(session) +""" + children = [subprocess.Popen(self.command(code, self.sid, str(n)), stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True) for n in range(8)] + try: + for child in children: + _, error = child.communicate(timeout=40) + self.assertEqual(child.returncode, 0, error) + finally: + for child in children: + if child.poll() is None: + child.kill() + child.wait() + saved = LS.load_session(self.sid) + events = [e for e in saved["events"] if e["eventType"] == "child-checkpoint"] + self.assertEqual({e["payload"]["child"] for e in events}, {str(n) for n in range(8)}) + self.assertEqual(len(events), 8) + self.assertEqual(saved["persistenceRevision"], 9) + + def test_killed_process_releases_lock_and_keeps_committed_checkpoint(self): + marker = self.root / "checkpoint-saved" + code = """ +import lane_session as LS +from pathlib import Path +import time +with LS.session_lock(sys.argv[2]): + session = LS.load_session(sys.argv[2]) + LS.record_event(session, 'before-process-death', {}) + LS.save_session(session) + Path(sys.argv[3]).write_text('saved') + time.sleep(60) +""" + child = subprocess.Popen(self.command(code, self.sid, str(marker)), stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True) + try: + deadline = time.monotonic() + 15 + while not marker.exists() and child.poll() is None and time.monotonic() < deadline: + time.sleep(0.02) + self.assertTrue(marker.exists(), "child did not commit its checkpoint") + finally: + if child.poll() is None: + child.kill() + child.communicate(timeout=5) + recovered = self.child("import lane_session as LS,json;print(json.dumps(LS.load_session(sys.argv[2])))", self.sid) + self.assertEqual(recovered["persistenceRevision"], 2) + self.assertEqual(recovered["events"][-1]["eventType"], "before-process-death") + self.assertEqual(recovered["authorizedCheckpointId"], self.session["authorizedCheckpointId"]) + + def test_lost_lock_does_not_advance_state_or_fall_back(self): + with self.store.lock(self.sid) as connection: + connection.close() + with self.assertRaisesRegex(PG.SessionStorageError, "lock was lost"): + LS.save_session(self.session) + self.assertEqual(self.session["persistenceRevision"], 1) + self.assertEqual(LS.load_session(self.sid)["persistenceRevision"], 1) + self.assertFalse((self.root / "must-not-be-used").exists()) + + def test_fresh_process_recovers_verified_engine_evidence(self): + artifact = { + "sessionId": self.sid, "taskId": self.task["taskId"], + "authorized_checkpoint_id": self.session["authorizedCheckpointId"], + "authorized_intent_hash": self.session["authorizedIntentHash"], + "producer": {"adapterId": "synthetic", "surface": "test", "generatedAt": "2026-09-17"}, + "files": {"test_fixture.py": "import unittest\nclass Fixture(unittest.TestCase):\n def test_access(self):\n self.assertEqual(2 + 2, 4)\n"}, + } + E.apply_worker_artifact(session_id=self.sid, task_id=self.task["taskId"], artifact=artifact) + result = E.verify_task(session_id=self.sid, task_id=self.task["taskId"], command={ + "argv": [sys.executable, "-m", "unittest", "discover", "-s", ".", "-p", "test_fixture.py"]}) + self.assertTrue(result["passed"]) + before = LS.load_session(self.sid) + recovered = self.child(""" +import lane_session as LS,json +session = LS.load_session(sys.argv[2]) +print(json.dumps({'state': LS.global_state(session), 'session': session})) +""", self.sid) + self.assertEqual(recovered["state"], "VERIFIED_COMPLETE") + self.assertEqual(recovered["session"], before) + E.interrupt_project(session_id=self.sid, correction="Membership access must require a new explicit approval.") + self.assertNotEqual(LS.global_state(LS.load_session(self.sid)), "VERIFIED_COMPLETE") + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/build_chatgpt_adapter.py b/tools/build_chatgpt_adapter.py index 9f3d457..8226533 100644 --- a/tools/build_chatgpt_adapter.py +++ b/tools/build_chatgpt_adapter.py @@ -102,6 +102,8 @@ def write_text(path: Path, text: str, mode: int) -> None: "scripts/build_engine.py", "scripts/capabilities.py", "scripts/checkpoint.py", "scripts/context_budget.py", "scripts/feedback.py", "scripts/intent_contract.py", "scripts/lane_registry.py", "scripts/lane_session.py", "scripts/policy_guard.py", + "scripts/postgres_sessions.py", + "scripts/progress_checkpoint.py", "scripts/project_index.py", "scripts/text_gate.py", "lanes/si.execution.json", "lanes/si.planning.json", "schemas/lane.schema.json", }) diff --git a/tools/public_plugin.py b/tools/public_plugin.py index ba938f7..f8fe414 100644 --- a/tools/public_plugin.py +++ b/tools/public_plugin.py @@ -38,9 +38,10 @@ MAX_ENTRY = 100 * 1024 * 1024 MAX_SEGMENTS = 20 MAX_PATH_BYTES = 1_024 -# 59 canonical runtime inputs (including delivery evidence) plus manifest and icon. +# 61 canonical runtime inputs (including optional storage and delivery evidence) +# plus manifest and icon. # Keep the byte budget and exact-content checks; never omit execution to fit. -MAX_RUNTIME_ENTRIES = 61 +MAX_RUNTIME_ENTRIES = 63 MAX_RUNTIME_UNCOMPRESSED = 1 * 1024 * 1024 SUPPORTED_CATEGORIES = { "Productivity", diff --git a/tools/test_chatgpt_adapter.py b/tools/test_chatgpt_adapter.py index 9041a84..a103ff1 100644 --- a/tools/test_chatgpt_adapter.py +++ b/tools/test_chatgpt_adapter.py @@ -27,9 +27,10 @@ def main() -> int: files = sorted(path.relative_to(ADAPTER_ROOT).as_posix() for path in ADAPTER_ROOT.rglob("*") if path.is_file()) skill_entrypoints = [path for path in files if Path(path).name == "SKILL.md"] require(skill_entrypoints == ["SKILL.md"], f"expected one SKILL.md, found {skill_entrypoints}") - # Keep all 50 existing runtime files and add eight missing execution owners - # and dependencies. This is a repository budget, not platform acceptance. - require(len(files) <= 59, f"runtime adapter exceeds the explicit delivery budget: {len(files)} files") + # The 59-file execution bundle also carries the optional Postgres adapter + # and its configuration reference. This is a repository budget, not + # platform acceptance; exact source hashes and required owners follow. + require(len(files) <= 61, f"runtime adapter exceeds the explicit delivery budget: {len(files)} files") require(BUILD.EXECUTION_RUNTIME_FILES.issubset(files), "runtime execution dependency is missing") # Re-derive the exact allowlist, bytes and source identities. File count or # version labels alone cannot establish that a repair reached this package. diff --git a/tools/test_chatgpt_execution_delivery.py b/tools/test_chatgpt_execution_delivery.py index 0aa56cc..f65eb47 100644 --- a/tools/test_chatgpt_execution_delivery.py +++ b/tools/test_chatgpt_execution_delivery.py @@ -22,7 +22,8 @@ REPO = Path(__file__).resolve().parents[1] SOURCE = REPO / "skills" / "selective-intelligence" CORE = ("build_engine", "capabilities", "checkpoint", "context_budget", "feedback", - "intent_contract", "lane_session", "policy_guard", "text_gate") + "intent_contract", "lane_session", "policy_guard", "postgres_sessions", + "progress_checkpoint", "project_index", "text_gate") REQUIRED = {"scripts/" + name + ".py" for name in CORE} | { "scripts/lane_registry.py", "lanes/si.execution.json", "lanes/si.planning.json", "schemas/lane.schema.json",