From d44dc7a4ff008d02bb53a32aeb0a09c430fc0eef Mon Sep 17 00:00:00 2001 From: GUOGUO Date: Sat, 17 Oct 2026 12:20:03 +0800 Subject: [PATCH] feat: publish store-owned candidate bundles in an authorized local store Create/Update write complete bundles to one explicitly selected SQLite store outside the installed package and any checkout. Package and checkout IDs are reserved, adaptations need a distinct ID with derived_from lineage, entrypoints must be UTF-8, and duplicate/stale/revert writes are explicit conflicts. History keeps the original resources. Store items join Collect under qualified store refs. Agent-assisted-by: Claude Code (Opus) (implementer) Agent-assisted-by: Cursor Agent (Claude Opus) (implementer) Signed-off-by: GUOGUO Co-authored-by: Cursor --- docs/knowledge-base.md | 15 +- docs/knowledge-store.md | 85 ++++++++ src/omnirsi/knowledge_operations.py | 13 +- src/omnirsi/knowledge_store.py | 234 ++++++++++++++++++++++ tests/test_knowledge_store.py | 300 ++++++++++++++++++++++++++++ 5 files changed, 639 insertions(+), 8 deletions(-) create mode 100644 docs/knowledge-store.md create mode 100644 src/omnirsi/knowledge_store.py create mode 100644 tests/test_knowledge_store.py diff --git a/docs/knowledge-base.md b/docs/knowledge-base.md index 0f04988d..0596a0ba 100644 --- a/docs/knowledge-base.md +++ b/docs/knowledge-base.md @@ -37,12 +37,13 @@ flowchart TB P -. "source view" .-> M1 C -. "coverage view" .-> M2["SearchSpace Matrix"] O["K2: normalized reads / Collect"] -. "reuse" .-> R - W["Follow-up: Create / Update"] -. "authorized local store only" .-> B["Complete revision bundles"] + W["K2: Create / Update"] -. "store-owned candidates only" .-> B["Complete revision bundles"] ``` -Solid edges describe the K0/K1 foundation; the dashed -[normalized reads / Collect](knowledge-operations.md) box reuses those readers, -and Create / Update remains a follow-up. No mandatory ContextPack, conversation manager, crawler or universal +Solid edges describe the K0/K1 foundation; dashed boxes are the K2 +[normalized reads / Collect](knowledge-operations.md) and +[authorized local Create / Update](knowledge-store.md) operations, which reuse +those readers. No mandatory ContextPack, conversation manager, crawler or universal performance loop is introduced. A generated catalog is a projection, not a third metadata authority or maintained matrix. @@ -82,7 +83,9 @@ dependency and peer remain independent repository relationships. The legacy readers retain their command contracts. [Normalized reads](knowledge-operations.md) identify complete declared dependency bundles and reject unavailable old snapshot -revisions. An old revision never silently resolves newer resource bytes. Typed source +revisions. The [authorized store](knowledge-store.md) retains old metadata/resources +and verifies their bundle digest. An old revision never silently resolves newer +resource bytes. Typed source examples are present for `audit-vllm-kv-lifetimes` (repository), `calibrate-serving-simulation` (reviewed Vidur v2 mechanism), and the internal case. Other source records retain their existing evidence rather than fabricated fields. @@ -150,7 +153,7 @@ available after noneditable installation; local `skills/` is not wheel data. |---|---|---| | PR #12, K0/K1 | Ownership, preserved content, metadata projection, explicit instruction/resource/reference reads, both matrices and reusable case | No normalized Collect or mutation API claim | | K2 reads | [`KnowledgeReader`](knowledge-operations.md) facade, typed sources, byte-derived bundle identity, authority-qualified refs with ambiguity rejection, exact revisions/resources and bounded Collect | Historical snapshot material must exist; no crawling or standalone export | -| PR 3, K2 writes | Separate follow-up | Supplied store-owned candidate Create/Update in one authorized store, adapter-identity collision rejection, duplicate/stale rejection, serialized writes, complete publication and immutable history | +| K2 writes | Python `KnowledgeStore` for store-owned candidates, adapter-identity reservation, `derived_from` adaptations, SQLite writer serialization, duplicate/stale rejection, complete publication and immutable history | One authorized local directory only; no repository writer, promotion or distributed writer guarantee | The writer will not edit site-packages, frozen evidence or source checkouts. A version field alone provides no concurrency guarantee. No automatic promotion diff --git a/docs/knowledge-store.md b/docs/knowledge-store.md new file mode 100644 index 00000000..3ecb345a --- /dev/null +++ b/docs/knowledge-store.md @@ -0,0 +1,85 @@ +# Authorized local Create / Update + +This K2 write slice builds on the [normalized read contract](knowledge-operations.md). +It adds one backend: SQLite in an **existing, explicitly authorized local directory**. +Create and Update operate only on **store-owned candidate bundles**. Package and +checkout content stay read-only; they are never written, and their IDs cannot be +reused in the store. Amending repository-owned content follows the normal scoped +PR path. + +```python +from pathlib import Path +from omnirsi.knowledge_operations import Bundle, KnowledgeReader +from omnirsi.knowledge_store import KnowledgeStore + +skills = Path("skills") +store = KnowledgeStore(Path("/authorized/local/knowledge-workspace"), skill_root=skills) +source = KnowledgeReader(skill_root=skills).read("calibrate-serving-simulation") +draft = Bundle( + {**source.metadata, "id": "calibrate-serving-simulation-local", "state": "candidate", + "derived_from": [{"ref": "checkout:calibrate-serving-simulation", + "revision": source.revision}]}, + dict(source.content), dict(source.dependencies), +) +revision = store.create(draft) +# Supply a complete amended Bundle, preserving source attribution and resources: +# next_revision = store.update(amended, expected=revision, +# reason="Clarify applicability", evidence=[source_pointer]) +old = store.read("calibrate-serving-simulation-local", revision=revision) +collected = store.collect(ids=["store:calibrate-serving-simulation-local"], + revisions={"store:calibrate-serving-simulation-local": revision}, + material=True) +``` + +`skill_root` is required: pass the checkout skills root whose IDs must be reserved, +or `None` to declare package-only use. `store.reader()` / `store.collect` combine +package, checkout and store items with qualified `store:` refs. An ID owned by +the store and a read-only adapter is reported as ambiguous, never shadowed. + +## Create and Update + +- Create registers a supplied **candidate** and rejects duplicate identity. It + rejects an ID owned by package or checkout content. An adaptation of existing + content uses a distinct ID and records `derived_from` entries with a qualified + `ref` and the source content revision (or `null` if unknown). +- Create does not relabel imported content or generate guidance. Update requires + the expected current revision, a reason and an explicit evidence list (empty + means none supplied). It publishes a successor without changing the content's + state; promotion is not implemented. +- An Update whose bundle reproduces a retained revision is an explicit conflict. + Existing history preserves old metadata, provenance, resources, parent revision, + reason and evidence even if a later interpretation changes. +- If a read-only adapter later gains a store item's ID, reads report ambiguity and + further Updates fail with a collision until the owners resolve it. + +## Writer and publication guarantees + +- Scope: cooperating processes/threads using this API on one local SQLite file; + no network-filesystem, distributed-store or hostile filesystem-owner guarantee. +- The workspace must already exist and lie outside the installed `omnirsi` + package, the checkout containing `skill_root` and, for a source or editable + install, the checkout containing the package (these hold evidence and both + matrices). That package checkout is the nearest ancestor holding both + `pyproject.toml` and `.git`; with `skill_root=None` it is the only checkout + guarded, so a checkout lacking either marker is not protected. The entrypoint of a stored bundle must be UTF-8 text. Workspace and database symlinks are rejected. No destination + directory is guessed or automatically created; the caller controls permissions. +- SQLite `BEGIN IMMEDIATE` serializes writers before duplicate/stale checks. + Resources, revision metadata and the current pointer commit in one transaction. + A database error rolls the whole transaction back; partial usable bundles are + never published. A five-second lock timeout raises an explicit SQLite error. +- Content/resource/dependency validation runs before writing. Dependencies are + included as bytes and checked against declared digests. Old reads load retained + bytes and verify their complete bundle identity; they never fall through to a + current checkout. No external mutable dependency lookup occurs during reads. +- Resource keys cannot be absolute paths or traversal paths; they are database + keys, never filesystem write destinations. +- Runtime operations write only the store database. They do not edit adapter + sources, raw evidence or either matrix, and a store candidate is not repository + coverage until exported through a reviewed repository change. +- Normal SQLite local-filesystem transaction guarantees apply. This implementation + does not claim tested power-loss durability, malicious tamper resistance or an + automatic recovery campaign. Integrity mismatches fail visibly. + +Tests use real bundles, SQLite transactions, competing writer connections and a +real rejecting database trigger to prove rollback after an insert starts. They do +not use hardware, synthetic performance claims or a fake storage implementation. diff --git a/src/omnirsi/knowledge_operations.py b/src/omnirsi/knowledge_operations.py index 4f137e17..749819f8 100644 --- a/src/omnirsi/knowledge_operations.py +++ b/src/omnirsi/knowledge_operations.py @@ -3,6 +3,7 @@ import hashlib import json import re +import sqlite3 from dataclasses import dataclass from importlib.resources import files from pathlib import Path, PurePosixPath @@ -177,8 +178,11 @@ def _tree(node, prefix: str) -> dict[str, bytes]: class KnowledgeReader: - def __init__(self, *, skill_root: Path | None = None): + def __init__(self, *, skill_root: Path | None = None, store=None): self.skill_root = skill_root.resolve() if skill_root is not None else None + if store is not None and store.skill_root != self.skill_root: + raise ValueError("Store must reserve identities from the same checkout") + self.store = store self._records: dict[str, dict] | None = None self._library: tuple[dict[str, bytes], str] | None = None @@ -210,6 +214,9 @@ def owners(self) -> dict[str, list[str]]: raise IdentityError("collision", "Duplicate identity inside the checkout") for identity in local: result.setdefault(identity, []).append("checkout") + if self.store is not None: + for identity in self.store.identities(): + result.setdefault(identity, []).append("store") return result def identities(self) -> list[str]: @@ -250,6 +257,8 @@ def read(self, reference: str, *, revision: str | None = None) -> Bundle: def _read(self, qualified: str, revision: str | None) -> Bundle: authority, _, identity = qualified.partition(":") + if authority == "store": + return self.store.read(identity, revision=revision) bundle = self._checkout(identity) if authority == "checkout" else self._knowledge(identity) if revision is not None and bundle.revision != revision: raise RevisionConflict( @@ -455,7 +464,7 @@ def matches(metadata: dict, authority: str) -> bool: {"id": reference, "reason": "unavailable_revision", "detail": str(exc)} ) continue - except (ValueError, KeyError, TypeError, AttributeError, OSError) as exc: + except (ValueError, KeyError, TypeError, AttributeError, OSError, sqlite3.Error) as exc: omitted.append({"id": reference, "reason": "invalid", "detail": str(exc)}) continue if not matches(view, authority): diff --git a/src/omnirsi/knowledge_store.py b/src/omnirsi/knowledge_store.py new file mode 100644 index 00000000..ce971a26 --- /dev/null +++ b/src/omnirsi/knowledge_store.py @@ -0,0 +1,234 @@ +from __future__ import annotations + +import json +import re +import sqlite3 +from contextlib import closing +from importlib.resources import files +from pathlib import Path + +from .knowledge_operations import ( + AUTHORITIES, + IDENTITY, + Bundle, + IdentityError, + KnowledgeReader, + RevisionConflict, + _json, + safe_path, +) + + +def _protected(skill_root: Path | None) -> list[Path]: + package = Path(str(files("omnirsi"))).resolve() + roots = [package] + # A source or editable install lives inside a checkout that also holds evidence + # and both matrices; protect that checkout even without an explicit skill_root. + for parent in package.parents: + if (parent / "pyproject.toml").is_file() and (parent / ".git").exists(): + roots.append(parent) + break + if skill_root is not None: + roots.append(Path(skill_root).resolve().parent) + return roots + + +class KnowledgeStore: + """One authorized writable store for store-owned candidate bundles. + + Package and checkout content stay read-only; their IDs are reserved here so a + store candidate cannot shadow them. Repository-owned amendments use PRs. + """ + + def __init__(self, workspace: Path, *, skill_root: Path | None): + # The caller authorizes an existing dedicated directory; none is created or guessed. + workspace = Path(workspace) + if workspace.is_symlink() or not workspace.is_dir(): + raise ValueError("Store requires an existing authorized workspace directory") + self.workspace = workspace.resolve() + for root in _protected(skill_root): + if self.workspace.is_relative_to(root): + raise ValueError(f"Store workspace must be outside read-only sources: {root}") + self.skill_root = Path(skill_root).resolve() if skill_root is not None else None + self.adapters = KnowledgeReader(skill_root=self.skill_root) + self.database = self.workspace / "knowledge.sqlite3" + if self.database.is_symlink(): + raise ValueError("Store database cannot be a symlink") + with closing(self._connect()) as connection, connection: + connection.executescript( + "CREATE TABLE IF NOT EXISTS revisions (" + "id TEXT NOT NULL, revision TEXT NOT NULL, metadata TEXT NOT NULL, " + "dependencies TEXT NOT NULL, parent TEXT, reason TEXT NOT NULL, " + "evidence TEXT NOT NULL, PRIMARY KEY (id, revision));" + "CREATE TABLE IF NOT EXISTS resources (" + "id TEXT NOT NULL, revision TEXT NOT NULL, path TEXT NOT NULL, data BLOB NOT NULL, " + "PRIMARY KEY (id, revision, path), " + "FOREIGN KEY (id, revision) REFERENCES revisions(id, revision));" + "CREATE TABLE IF NOT EXISTS current (id TEXT PRIMARY KEY, revision TEXT NOT NULL, " + "FOREIGN KEY (id, revision) REFERENCES revisions(id, revision));" + ) + + def _connect(self): + if self.database.is_symlink() or self.database.resolve().parent != self.workspace: + raise ValueError("Store database escapes the authorized workspace") + connection = sqlite3.connect(self.database, timeout=5) + connection.execute("PRAGMA foreign_keys=ON") + return connection + + def identities(self) -> list[str]: + with closing(self._connect()) as connection: + return [row[0] for row in connection.execute("SELECT id FROM current ORDER BY id")] + + def reader(self) -> KnowledgeReader: + return KnowledgeReader(skill_root=self.skill_root, store=self) + + def collect(self, query: str = "", **options) -> dict: + return self.reader().collect(query, **options) + + def create(self, bundle: Bundle) -> str: + return self._publish(bundle, expected=None, reason="create", evidence=[]) + + def update(self, bundle: Bundle, *, expected: str, reason: str, evidence: list[dict]) -> str: + if not expected or not reason.strip(): + raise ValueError("Update requires expected old revision and reason") + return self._publish(bundle, expected=expected, reason=reason, evidence=evidence) + + def _lineage(self, metadata: dict) -> None: + lineage = metadata.get("derived_from", []) + if not isinstance(lineage, list): + raise ValueError("derived_from must be a list of qualified source revisions") + for source in lineage: + if not isinstance(source, dict) or not isinstance(source.get("ref"), str): + raise ValueError("derived_from entries require a qualified ref") + authority, _, identity = source["ref"].partition(":") + if authority not in AUTHORITIES or not re.fullmatch(IDENTITY, identity): + raise ValueError(f"derived_from ref must be :: {source['ref']}") + revision = source.get("revision") + if revision is not None and ( + not isinstance(revision, str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", revision) + ): + raise ValueError("derived_from revision must be a content revision or null") + if identity == metadata["id"]: + raise IdentityError( + "collision", "An adaptation needs a distinct identity from its source" + ) + + def _publish(self, bundle: Bundle, *, expected: str | None, reason: str, evidence: list) -> str: + # Copy caller data, validate it, then store the whole bundle and current + # pointer in one transaction. + bundle = Bundle( + json.loads(_json(bundle.metadata)), dict(bundle.content), dict(bundle.dependencies) + ) + revision = bundle.revision + identity = bundle.metadata["id"] + try: + bundle.content[bundle.metadata["entrypoint"]].decode("utf-8") + except UnicodeDecodeError as exc: + raise ValueError("Bundle entrypoint must be UTF-8 text") from exc + self._lineage(bundle.metadata) + owners = self.adapters.owners().get(identity) + if owners: + raise IdentityError( + "collision", + f"Identity {identity} is owned by read-only {', '.join(owners)} content; " + "amend it through a repository PR or create an adaptation with a distinct ID", + ) + if expected is None and bundle.metadata["state"] != "candidate": + raise ValueError( + "Create registers a candidate; imported or active content is not promoted" + ) + with closing(self._connect()) as connection, connection: + connection.execute("BEGIN IMMEDIATE") + current = connection.execute( + "SELECT revision FROM current WHERE id=?", (identity,) + ).fetchone() + if expected is None: + if current is not None: + raise RevisionConflict(f"Duplicate identity: {identity}") + elif current is None or current[0] != expected: + raise RevisionConflict(f"Stale or missing revision for {identity}") + if expected is not None: + previous = connection.execute( + "SELECT metadata FROM revisions WHERE id=? AND revision=?", (identity, expected) + ).fetchone() + old = json.loads(previous[0]) + if bundle.metadata["state"] != old["state"]: + raise ValueError("Update cannot promote content state") + if revision == expected: + raise ValueError("Update must change the bundle") + if connection.execute( + "SELECT 1 FROM revisions WHERE id=? AND revision=?", (identity, revision) + ).fetchone(): + raise RevisionConflict( + f"Update reproduces retained revision {revision}; " + "change the bundle so the successor has its own revision" + ) + connection.execute( + "INSERT INTO revisions VALUES (?, ?, ?, ?, ?, ?, ?)", + ( + identity, + revision, + _json(bundle.metadata).decode(), + _json(bundle.dependencies).decode(), + expected, + reason, + _json(evidence).decode(), + ), + ) + connection.executemany( + "INSERT INTO resources VALUES (?, ?, ?, ?)", + [(identity, revision, path, data) for path, data in bundle.content.items()], + ) + connection.execute( + "INSERT INTO current VALUES (?, ?) " + "ON CONFLICT(id) DO UPDATE SET revision=excluded.revision", + (identity, revision), + ) + return revision + + def read(self, identity: str, *, revision: str | None = None) -> Bundle: + with closing(self._connect()) as connection: + if revision is None: + row = connection.execute( + "SELECT revision FROM current WHERE id=?", (identity,) + ).fetchone() + if row is None: + raise IdentityError("unknown_identity", f"Unknown identity: store:{identity}") + revision = row[0] + row = connection.execute( + "SELECT metadata, dependencies FROM revisions WHERE id=? AND revision=?", + (identity, revision), + ).fetchone() + if row is None: + raise RevisionConflict(f"Unknown revision: store:{identity} at {revision}") + resources = connection.execute( + "SELECT path, data FROM resources WHERE id=? AND revision=?", (identity, revision) + ).fetchall() + bundle = Bundle(json.loads(row[0]), dict(resources), json.loads(row[1])) + if bundle.revision != revision: + raise RevisionConflict("Stored bundle integrity mismatch") + return bundle + + def resource(self, identity: str, path: str, *, revision: str) -> bytes: + safe_path(path) + bundle = self.read(identity, revision=revision) + if path not in bundle.content: + raise ValueError(f"Unknown resource: {path}") + return bundle.content[path] + + def history(self, identity: str) -> list[dict]: + with closing(self._connect()) as connection: + rows = connection.execute( + "SELECT revision, parent, reason, evidence FROM revisions " + "WHERE id=? ORDER BY rowid", + (identity,), + ).fetchall() + return [ + { + "revision": revision, + "parent": parent, + "reason": reason, + "evidence": json.loads(evidence), + } + for revision, parent, reason, evidence in rows + ] diff --git a/tests/test_knowledge_store.py b/tests/test_knowledge_store.py new file mode 100644 index 00000000..9f4104e2 --- /dev/null +++ b/tests/test_knowledge_store.py @@ -0,0 +1,300 @@ +import hashlib +import os +import sqlite3 +import subprocess +import sys +import time +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest + +import omnirsi +from omnirsi.knowledge_operations import Bundle, IdentityError, KnowledgeReader, RevisionConflict +from omnirsi.knowledge_store import KnowledgeStore + +REPO = Path(__file__).resolve().parents[1] +ROOT = REPO / "skills" +NOTES = "calibrate-serving-simulation/references/source-notes.md" +LOCAL = "calibrate-serving-simulation-local" + + +def adaptation(source="calibrate-serving-simulation", identity=LOCAL): + bundle = KnowledgeReader(skill_root=ROOT).read(source) + metadata = { + **bundle.metadata, + "id": identity, + "state": "candidate", + "derived_from": [{"ref": f"checkout:{source}", "revision": bundle.revision}], + } + return Bundle(metadata, dict(bundle.content), dict(bundle.dependencies)) + + +def changed(bundle, suffix): + content = dict(bundle.content) + content[NOTES] += suffix + return Bundle(dict(bundle.metadata), content, dict(bundle.dependencies)) + + +def store_at(path): + return KnowledgeStore(path, skill_root=ROOT) + + +def test_duplicate_stale_update_and_immutable_resources(tmp_path): + store = store_at(tmp_path) + original = adaptation() + revision = store.create(original) + with pytest.raises(RevisionConflict, match="Duplicate"): + store.create(original) + successor = changed(original, b"\nScoped interpretation amendment.\n") + next_revision = store.update(successor, expected=revision, reason="clarify", evidence=[]) + assert next_revision != revision + with pytest.raises(RevisionConflict, match="Stale"): + store.update(original, expected=revision, reason="stale", evidence=[]) + with pytest.raises(RevisionConflict, match="retained revision"): + store.update(original, expected=next_revision, reason="revert", evidence=[]) + assert store.resource(LOCAL, NOTES, revision=revision) == original.content[NOTES] + assert store.read(LOCAL).revision == next_revision + assert store.history(LOCAL)[1]["parent"] == revision + collected = store.collect(ids=[LOCAL], revisions={LOCAL: revision}, material=True) + item = collected["items"][0] + assert (item["revision"], item["authority"]) == (revision, "store") + assert item["ref"] == f"store:{LOCAL}" + + +def test_store_cannot_shadow_read_only_identities(tmp_path): + store = store_at(tmp_path) + for source in ("calibrate-serving-simulation", "case.admission-drain-occupancy"): + bundle = KnowledgeReader(skill_root=ROOT).read(source) + copied = Bundle({**bundle.metadata, "state": "candidate"}, bundle.content, {}) + with pytest.raises(IdentityError, match="read-only checkout"): + store.create(copied) + package_id = KnowledgeReader().identities()[0].partition(":")[2] + record = KnowledgeReader().read(package_id) + copied = Bundle({**record.metadata, "state": "candidate"}, record.content, record.dependencies) + with pytest.raises(IdentityError, match="read-only package"): + store.create(copied) + looped = adaptation() + looped.metadata["derived_from"] = [{"ref": f"store:{LOCAL}", "revision": None}] + with pytest.raises(IdentityError, match="distinct identity"): + store.create(looped) + malformed = adaptation() + malformed.metadata["derived_from"] = [{"ref": "calibrate-serving-simulation"}] + with pytest.raises(ValueError, match="authority"): + store.create(malformed) + assert store.identities() == [] + + +def test_identity_added_later_to_checkout_is_ambiguous(tmp_path): + skills = tmp_path / "checkout" / "skills" + (skills / "later").mkdir(parents=True) + (skills / "later" / "SKILL.md").write_text( + "---\nname: later\ndescription: test\n---\nBody", encoding="utf-8" + ) + workspace = tmp_path / "store" + workspace.mkdir() + store = KnowledgeStore(workspace, skill_root=skills) + store.create(adaptation(identity="arrives-later")) + (skills / "arrives-later").mkdir() + (skills / "arrives-later" / "SKILL.md").write_text( + "---\nname: arrives-later\ndescription: test\n---\nBody", encoding="utf-8" + ) + reader = store.reader() + with pytest.raises(IdentityError, match="Ambiguous"): + reader.read("arrives-later") + assert reader.read("store:arrives-later").metadata["state"] == "candidate" + assert reader.collect(ids=["arrives-later"])["omissions"][0]["reason"] == "ambiguous_identity" + with pytest.raises(IdentityError, match="read-only checkout"): + current = store.read("arrives-later") + store.update(changed(current, b"x"), expected=current.revision, reason="r", evidence=[]) + with pytest.raises(ValueError, match="same checkout"): + KnowledgeReader(skill_root=ROOT, store=store) + + +def test_concurrent_writers_compare_expected_inside_transaction(tmp_path, monkeypatch): + store = store_at(tmp_path) + original = adaptation() + revision = store.create(original) + lock_attempts = [] + connect = store._connect + + def traced_connect(): + connection = connect() + connection.set_trace_callback( + lambda sql: lock_attempts.append(sql) if sql == "BEGIN IMMEDIATE" else None + ) + return connection + + monkeypatch.setattr(store, "_connect", traced_connect) + + def update(index): + try: + return store.update( + changed(original, str(index).encode()), + expected=revision, + reason="concurrent amendment", + evidence=[], + ) + except RevisionConflict: + return "conflict" + + # Hold the write lock so both writers pass any pre-transaction check before + # either can publish; only an in-transaction comparison yields one conflict. + blocker = sqlite3.connect(store.database, isolation_level=None) + blocker.execute("BEGIN IMMEDIATE") + with ThreadPoolExecutor(max_workers=2) as workers: + pending = [workers.submit(update, index) for index in (1, 2)] + # Both writers must be waiting on the lock, well inside its 5 s timeout. + deadline = time.monotonic() + 2 + while len(lock_attempts) < 2 and time.monotonic() < deadline: + time.sleep(0.01) + assert len(lock_attempts) == 2 + assert not any(future.done() for future in pending) + blocker.execute("ROLLBACK") + blocker.close() + results = [future.result() for future in pending] + assert results.count("conflict") == 1 + assert len(store.history(LOCAL)) == 2 + + +def test_database_rejection_rolls_back_complete_publication(tmp_path): + store = store_at(tmp_path) + original = adaptation() + revision = store.create(original) + with sqlite3.connect(store.database) as connection: + connection.execute( + "CREATE TRIGGER reject_resource BEFORE INSERT ON resources " + "WHEN NEW.path LIKE '%source-notes.md' " + "BEGIN SELECT RAISE(ABORT, 'resource rejected'); END" + ) + with pytest.raises(sqlite3.IntegrityError, match="resource rejected"): + store.update(changed(original, b"new"), expected=revision, reason="blocked", evidence=[]) + assert store.read(LOCAL).revision == revision + assert len(store.history(LOCAL)) == 1 + + +def test_incomplete_unsafe_and_promotion_writes_fail(tmp_path): + store = store_at(tmp_path) + original = adaptation() + incomplete = dict(original.content) + del incomplete[NOTES] + with pytest.raises(ValueError, match="Missing required"): + store.create(Bundle(original.metadata, incomplete, {})) + bad_metadata = {**original.metadata, "required_resources": ["missing.md"]} + with pytest.raises(ValueError, match="Missing required"): + store.create(Bundle(bad_metadata, original.content, {})) + with pytest.raises(ValueError, match="Unsafe"): + store.create(Bundle(original.metadata, {**original.content, "../outside": b"no"}, {})) + with pytest.raises(ValueError, match="Missing dependency"): + store.create(Bundle(original.metadata, original.content, {"missing": "sha256:" + "0" * 64})) + with pytest.raises(ValueError, match="candidate"): + store.create(Bundle({**original.metadata, "state": "active"}, original.content, {})) + entry = original.metadata["entrypoint"] + with pytest.raises(ValueError, match="UTF-8"): + store.create(Bundle(original.metadata, {**original.content, entry: b"\xff\xfe"}, {})) + bad_lineage = {**original.metadata, "derived_from": [{"ref": "checkout:x", "revision": 7}]} + with pytest.raises(ValueError, match="content revision"): + store.create(Bundle(bad_lineage, original.content, {})) + revision = store.create(original) + with pytest.raises(ValueError, match="promote"): + store.update( + Bundle({**original.metadata, "state": "active"}, original.content, {}), + expected=revision, + reason="promote", + evidence=[], + ) + assert len(store.history(LOCAL)) == 1 + + +def test_workspace_location_and_symlinks(tmp_path): + with pytest.raises(ValueError, match="existing authorized"): + store_at(tmp_path / "not-authorized-or-created") + for skill_root in (ROOT, None): + with pytest.raises(ValueError, match="outside read-only"): + KnowledgeStore(REPO / "evidence", skill_root=skill_root) + with pytest.raises(ValueError, match="outside read-only"): + KnowledgeStore(Path(omnirsi.__file__).parent / "data", skill_root=None) + target = tmp_path / "target" + target.mkdir() + link = tmp_path / "link" + try: + link.symlink_to(target, target_is_directory=True) + except OSError as exc: + pytest.skip(f"symlink unavailable: {exc}") + with pytest.raises(ValueError, match="existing authorized"): + store_at(link) + workspace = tmp_path / "workspace" + workspace.mkdir() + (workspace / "knowledge.sqlite3").symlink_to(target / "elsewhere.sqlite3") + with pytest.raises(ValueError, match="symlink"): + store_at(workspace) + + +def test_real_shared_resource_bundle_keeps_old_dependency_bytes(tmp_path): + original = adaptation("inference-benchmark-design", "inference-benchmark-design-local") + store = store_at(tmp_path) + revision = store.create(original) + shared = "inference-optimization-casebook/cases/admission-drain-occupancy.md" + identity = original.metadata["id"] + assert store.resource(identity, shared, revision=revision) == original.content[shared] + altered = dict(original.content) + altered[shared] += b"changed" + with pytest.raises(RevisionConflict, match="Dependency bytes"): + store.update( + Bundle(original.metadata, altered, original.dependencies), + expected=revision, + reason="stale dependency", + evidence=[], + ) + + +def _digest(paths): + digest = hashlib.sha256() + for root in paths: + files = sorted(p for p in root.rglob("*") if p.is_file()) if root.is_dir() else [root] + for path in files: + digest.update(path.relative_to(REPO).as_posix().encode()) + digest.update(path.read_bytes()) + return digest.hexdigest() + + +def test_operations_leave_sources_evidence_and_matrices_unchanged(tmp_path): + watched = [ + ROOT, + REPO / "src" / "omnirsi" / "data", + REPO / "evidence", + REPO / "docs" / "reference-matrix.md", + REPO / "docs" / "searchspace-matrix.md", + ] + before = _digest(watched) + store = store_at(tmp_path) + revision = store.create(adaptation()) + store.update(changed(adaptation(), b"amend"), expected=revision, reason="r", evidence=[]) + store.collect(limit=500, material=True) + store.reader().collect("calibration", material=True) + store.read(LOCAL, revision=revision) + assert _digest(watched) == before + assert sorted(p.name for p in tmp_path.iterdir()) == ["knowledge.sqlite3"] + + +def test_package_only_collect_without_checkout_or_yaml(tmp_path): + script = ( + "import sys; sys.modules['yaml'] = None\n" + "from omnirsi.knowledge_operations import KnowledgeReader\n" + "reader = KnowledgeReader()\n" + "result = reader.collect(tier='dependency', material=True, limit=2)\n" + "item = result['items'][0]\n" + "assert item['authority'] == 'package', item\n" + "assert reader.read(item['ref'], revision=item['revision']).revision == item['revision']\n" + "assert 'omnirsi.skill_catalog' not in sys.modules\n" + ) + env = {**os.environ, "PYTHONPATH": str(REPO / "src"), "PYTHONUTF8": "1"} + result = subprocess.run( + [sys.executable, "-c", script], + cwd=tmp_path, + env=env, + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 0, result.stderr