工具 (0) 刷新
@@ -702,6 +717,30 @@
运行日志
// ────────────────────────────────────────────────────────
// Sidebar loaders
// ────────────────────────────────────────────────────────
+function retentionInputs() {
+ $('retentionCount').hidden = $('retentionMode').value !== 'recent';
+}
+async function loadRetention() {
+ try {
+ const data = await api('/api/workspace');
+ $('retentionMode').value = data.retention.mode;
+ $('retentionCount').value = data.retention.keep_last;
+ retentionInputs();
+ $('retentionStatus').textContent = '下次发布后按此策略清理。正在查询和手动保护的版本会保留。';
+ } catch (err) { $('retentionStatus').textContent = err.message; }
+}
+$('retentionMode').addEventListener('change', retentionInputs);
+$('saveRetentionBtn').addEventListener('click', async () => {
+ const button = $('saveRetentionBtn'); button.disabled = true;
+ try {
+ await api('/api/workspace/retention', {method:'POST', body:JSON.stringify({
+ mode:$('retentionMode').value, keep_last:Number($('retentionCount').value)})});
+ await loadRetention();
+ logLine('已保存 Workspace 历史保留方式');
+ } catch (err) { $('retentionStatus').textContent = err.message; }
+ finally { button.disabled = false; }
+});
+
async function loadHealth() {
try {
state.health = await api('/api/health');
@@ -1338,7 +1377,7 @@
运行日志
// Bootstrap
// ────────────────────────────────────────────────────────
(async () => {
- await Promise.all([loadHealth(), loadTools(), loadGraph(), loadKb()]);
+ await Promise.all([loadHealth(), loadTools(), loadGraph(), loadKb(), loadRetention()]);
$('chatInput').focus();
})();
diff --git a/datamind/tests/test_cli_codex.py b/datamind/tests/test_cli_codex.py
index 4ed7d92..ac5e806 100644
--- a/datamind/tests/test_cli_codex.py
+++ b/datamind/tests/test_cli_codex.py
@@ -69,3 +69,20 @@ def test_root_http_launcher_exports_the_same_app():
root=Path(__file__).resolve().parents[2]
module=runpy.run_path(str(root/'server.py'),run_name='datamind_root_server_test')
assert module['app'] is app
+
+
+def test_workspace_cli_policy_persists_and_gc_defaults_to_preview(monkeypatch,tmp_path):
+ import json
+ cfg=DataSettings(_env_file=None,data={'base_dir':tmp_path})
+ monkeypatch.setattr(cli,'_settings',lambda:cfg)
+ runner=CliRunner()
+ result=runner.invoke(cli.app,['workspace-policy','recent','--keep-last','3'])
+ assert result.exit_code==0,result.output
+ status=runner.invoke(cli.app,['workspace-status','--usage'])
+ assert status.exit_code==0,status.output
+ value=json.loads(status.output)
+ assert value['retention']=={'mode':'recent','keep_last':3}
+ preview=runner.invoke(cli.app,['workspace-gc'])
+ assert preview.exit_code==0,preview.output
+ assert json.loads(preview.output)['dry_run'] is True
+ assert runner.invoke(cli.app,['workspace-policy','invalid']).exit_code!=0
diff --git a/datamind/tests/test_db_versioning.py b/datamind/tests/test_db_versioning.py
index 295083c..af6e8ae 100644
--- a/datamind/tests/test_db_versioning.py
+++ b/datamind/tests/test_db_versioning.py
@@ -282,3 +282,20 @@ async def test_historical_join_and_cte_pin_all_tables_and_preserve_literals(inge
current = await db_service.query_sql(sql)
assert old.rows == [["old left", "old right", "FROM right_table"]]
assert current.rows == [["new left", "new right", "FROM right_table"]]
+
+
+@pytest.mark.parametrize('error_type', ['InterfaceError', 'DatabaseError'])
+def test_dolt_startup_retries_connection_refusal_from_both_connector_engines(monkeypatch,error_type):
+ connector=pytest.importorskip('mysql.connector')
+ from types import SimpleNamespace
+ from datamind.capabilities.db.providers import dolt
+ calls=[]
+ def connect(**kwargs):
+ calls.append('connect')
+ if len(calls)<3:
+ raise getattr(connector.errors,error_type)(errno=2003,msg="Can't connect to MySQL server")
+ return SimpleNamespace(close=lambda:calls.append('closed'))
+ monkeypatch.setattr(connector,'connect',connect)
+ monkeypatch.setattr(dolt.time,'sleep',lambda _:None)
+ DoltDialect()._wait_ready('127.0.0.1',3307,'root','',tries=3)
+ assert calls==['connect','connect','connect','closed']
diff --git a/datamind/tests/test_direct_data_plane.py b/datamind/tests/test_direct_data_plane.py
index dce4f6e..e8ab8ff 100644
--- a/datamind/tests/test_direct_data_plane.py
+++ b/datamind/tests/test_direct_data_plane.py
@@ -18,6 +18,7 @@ def settings(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
monkeypatch.delenv('DATAMIND__LLM__API_KEY', raising=False)
monkeypatch.setenv('DATAMIND_DATA_ROOT', str(tmp_path))
+ monkeypatch.setenv('DATAMIND__WORKSPACE__RETENTION','all')
monkeypatch.delenv('DATAMIND_PROFILE', raising=False)
from datamind import runtime as options
def forbidden(*a, **kw):
@@ -203,7 +204,7 @@ async def publish(owner):
assert all(r['status'] == 'stored' for r in result['result']['results']), result
manifest = result['result']['results'][0]['details']
assert manifest['metrics']['graph_extraction_calls'] == 0
- assert manifest['applied_skills'][0]['version'] == '1.2.0'
+ assert manifest['applied_skills'][0]['version'] == '1.3.0'
return manifest['workspace_revision']
first = await publish('OldOwner')
reader = await begin('read')
@@ -271,3 +272,20 @@ async def test_mcp_capability_set_cannot_be_expanded_by_agent(settings,monkeypat
await call(token,'db_list_tables')
with pytest.raises(PermissionError,match='capability set'):
await call(token,'workspace_publish',path='.',surfaces=['db'])
+
+
+def test_legacy_operation_schema_upgrades_safely_for_concurrent_clients(tmp_path):
+ import sqlite3
+ from concurrent.futures import ThreadPoolExecutor
+ from datamind.operations import OperationStore
+ cfg=DataSettings(_env_file=None,data={'base_dir':tmp_path})
+ cfg.ensure_dirs()
+ path=cfg.data.storage_dir/'operations.sqlite3'
+ with sqlite3.connect(path) as db:
+ db.execute('CREATE TABLE operations (id TEXT PRIMARY KEY, session TEXT NOT NULL, revision TEXT, mode TEXT NOT NULL)')
+ def begin(_):return OperationStore(cfg).begin(mode='write')['operation_id']
+ with ThreadPoolExecutor(max_workers=8) as pool:tokens=list(pool.map(begin,range(8)))
+ assert len(set(tokens))==8
+ with sqlite3.connect(path) as db:
+ assert db.execute('SELECT COUNT(*) FROM operations').fetchone()[0]==8
+ assert {'lease','request_id'}<={r[1] for r in db.execute('PRAGMA table_info(operations)')}
diff --git a/datamind/tests/test_http_api.py b/datamind/tests/test_http_api.py
index 177efdb..49290cd 100644
--- a/datamind/tests/test_http_api.py
+++ b/datamind/tests/test_http_api.py
@@ -222,3 +222,37 @@ async def test_chat_converts_stream_failure_to_sse_error(configured_app):
assert '"type": "text"' in response.text
assert '"type": "error"' in response.text
assert '"message": "stream failed: RuntimeError"' in response.text
+
+
+@pytest.mark.asyncio
+async def test_workspace_policy_routes_validate_and_persist(configured_app):
+ test_app,settings=configured_app
+ async with httpx.AsyncClient(transport=httpx.ASGITransport(app=test_app),base_url='http://test') as client:
+ status=await client.get('/api/workspace?usage=true')
+ assert status.json()['retention']['mode']=='latest'
+ assert 'storage' in status.json()
+ result=await client.post('/api/workspace/retention',json={'mode':'recent','keep_last':3})
+ assert result.json()=={'mode':'recent','keep_last':3}
+ assert (await client.get('/api/workspace')).json()['retention']['keep_last']==3
+ assert (await client.post('/api/workspace/retention',json={'mode':'none'})).status_code==422
+ assert (await client.post('/api/workspace/retention',json={'mode':'recent','keep_last':0})).status_code==422
+ preview=await client.post('/api/workspace/gc',json={})
+ assert preview.json()['dry_run'] is True
+ missing=await client.post('/api/workspace/rollback',json={'revision':'unknown'})
+ assert missing.status_code==409
+
+
+@pytest.mark.asyncio
+async def test_http_startup_with_private_workspace_dolt_needs_no_live_dsn(tmp_path,monkeypatch):
+ from datamind import server
+ cfg=Settings(_env_file=None,data={'base_dir':tmp_path},db={'dialect':'dolt'},
+ workspace={'storage_mode':'incremental'})
+ class Host:
+ def __init__(self,settings):pass
+ async def aclose(self):pass
+ monkeypatch.setattr(server,'DataSettings',lambda:cfg)
+ monkeypatch.setattr(server,'CodexHost',Host)
+ async with server._lifespan(server.app):
+ state=server.app.state.datamind
+ assert state.data.services.db is None
+ assert 'workspace_publish' in state.data.tools
diff --git a/datamind/tests/test_workspace_incremental.py b/datamind/tests/test_workspace_incremental.py
new file mode 100644
index 0000000..b46854b
--- /dev/null
+++ b/datamind/tests/test_workspace_incremental.py
@@ -0,0 +1,489 @@
+"""Incremental storage, native snapshots, retention and cross-call leases."""
+
+import asyncio
+import hashlib
+
+import pytest
+
+from datamind.config import Settings
+from datamind.core.errors import CapabilityError
+from datamind.workspace import Workspace
+
+
+class Embed:
+ name = "fixture"
+ dimension = 4
+ calls = 0
+
+ async def embed_texts(self, texts):
+ self.calls += len(texts)
+ return [[1.0, 0.0, 0.0, 0.0] for _ in texts]
+
+ async def embed_query(self, text):
+ return [1.0, 0.0, 0.0, 0.0]
+
+
+@pytest.fixture
+def incremental(tmp_path, monkeypatch):
+ from datamind import runtime
+
+ embed = Embed()
+ monkeypatch.setattr(runtime, "build_embedding", lambda *a, **kw: embed)
+ root = (tmp_path / "sources").resolve()
+ root.mkdir()
+ cfg = Settings(
+ _env_file=None,
+ data={"base_dir": tmp_path / "runtime"},
+ embedding={"model": "fixture", "dimension": 4},
+ kb={"vector_store": "lancedb"},
+ workspace={"storage_mode": "incremental", "retention": "all", "auto_gc": False},
+ )
+ return Workspace(cfg), root, embed
+
+
+async def publish(workspace, root, owner="Alice", surfaces=("graph",)):
+ doc = root / "owner.md"
+ doc.write_text(owner + " owns Project")
+ return await workspace.update(
+ str(root),
+ surfaces=surfaces,
+ graph_sources=[
+ {
+ "uri": str(doc),
+ "version": "sha256:" + hashlib.sha256(doc.read_bytes()).hexdigest(),
+ "triples": [
+ {"subject": owner, "relation": "owns", "object": "Project"}
+ ],
+ }
+ ],
+ )
+
+
+@pytest.mark.asyncio
+async def test_graph_objects_reused_deleted_and_rollback_remains_pinned(incremental):
+ w, root, _ = incremental
+ a = await publish(w, root)
+ b = await publish(w, root)
+ assert a["sources"] == b["sources"]
+ assert b["metrics"]["object_bytes_written"] == 0
+ assert b["metrics"]["sources_reused"] == 1
+ assert not (w.root / "revisions" / b["workspace_revision"] / "storage").exists()
+ async with w.reader(a["workspace_revision"]) as old:
+ c = await publish(w, root, "Bob")
+ assert "Alice" in str(await old.call("graph_neighbors", entity="Project"))
+ assert "Bob" not in str(await old.call("graph_neighbors", entity="Project"))
+ assert (
+ "Alice"
+ in (await old.call("raw_file_read", path=str(root / "owner.md")))["text"]
+ )
+ assert (await old.call("workspace_inspect", path=str(root)))["count"] == 1
+ w.rollback(a["workspace_revision"], expected_revision=c["workspace_revision"])
+ async with w.reader() as reader:
+ assert "Alice" in str(await reader.call("graph_neighbors", entity="Project"))
+ (root / "owner.md").unlink()
+ await w.update(str(root), surfaces=["graph"], graph_sources=[])
+ async with w.reader() as reader:
+ assert (await reader.call("graph_schema"))["nodes"] == 0
+
+
+@pytest.mark.asyncio
+async def test_native_kb_versions_share_storage_and_survive_cleanup(incremental):
+ pytest.importorskip("lancedb")
+ w, root, embed = incremental
+ a = await publish(w, root, surfaces=("kb", "graph"))
+ b = await publish(w, root, surfaces=("kb", "graph"))
+ assert a["snapshots"]["kb"] == b["snapshots"]["kb"]
+ assert embed.calls == 1
+ assert b["metrics"]["kb_rows_upserted"] == 0
+ assert b["metrics"]["object_bytes_written"] == 0
+ async with w.reader(a["workspace_revision"]) as old:
+ c = await publish(w, root, "Bob", surfaces=("kb", "graph"))
+ assert a["snapshots"]["kb"]["path"] == c["snapshots"]["kb"]["path"]
+ assert a["snapshots"]["kb"]["version"] != c["snapshots"]["kb"]["version"]
+ w.configure_retention("latest")
+ cleanup = await w.collect(dry_run=False, compact=True)
+ assert a["workspace_revision"] in cleanup["retained_revisions"]
+ assert "Alice" in str(await old.call("kb_search", query="owner"))
+ assert "Bob" not in str(await old.call("kb_search", query="owner"))
+ await w.collect(dry_run=False)
+ with pytest.raises(CapabilityError):
+ w.manifest(a["workspace_revision"])
+ async with w.reader() as reader:
+ assert "Bob" in str(await reader.call("kb_search", query="owner"))
+ assert (await reader.call("kb_count"))["chunks"] == 1
+
+
+@pytest.mark.asyncio
+async def test_latest_recent_all_and_protected_revisions(incremental):
+ w, root, _ = incremental
+ versions = []
+ for owner in ("Alice", "Bob", "Carol"):
+ versions.append(await publish(w, root, owner))
+ a, b, c = [m["workspace_revision"] for m in versions]
+ w.configure_retention("recent", 2)
+ preview = await w.collect()
+ assert preview["pruned_revisions"] == [a]
+ assert w.manifest(a)
+ w.protect(a)
+ assert (await w.collect(dry_run=False))["pruned_revisions"] == []
+ w.protect(a, False)
+ await w.collect(dry_run=False)
+ assert w.status(a)["state"] == "PRUNED"
+ w.configure_retention("latest")
+ result = await w.collect(dry_run=False)
+ assert b in result["pruned_revisions"]
+ assert w.manifest(c)
+ assert (await w.collect(dry_run=False))["reclaimed_bytes"] == 0
+
+
+@pytest.mark.asyncio
+async def test_failed_and_stale_publications_leave_previous_head(
+ incremental, monkeypatch
+):
+ w, root, _ = incremental
+ a = await publish(w, root)
+ doc = root / "owner.md"
+ doc.write_text("Bob owns Project")
+ with pytest.raises(Exception):
+ await w.update(
+ str(root),
+ surfaces=["graph"],
+ graph_sources=[
+ {"uri": str(doc), "version": "sha256:" + "0" * 64, "triples": []}
+ ],
+ )
+ assert w.current_revision == a["workspace_revision"]
+ # Publication checks epoch as well as revision, including rollback ABA.
+ b = await publish(w, root, "Bob")
+ w.rollback(a["workspace_revision"], expected_revision=b["workspace_revision"])
+ await publish(w, root, "Carol")
+ async with w.reader() as reader:
+ assert "Carol" in str(await reader.call("graph_neighbors", entity="Project"))
+
+
+def test_new_profiles_default_latest_existing_history_defaults_all(tmp_path):
+ cfg = Settings(_env_file=None, data={"base_dir": tmp_path})
+ w = Workspace(cfg)
+ assert w.retention["mode"] == "latest"
+ w.configure_retention("all")
+ assert Workspace(cfg).retention["mode"] == "all"
+
+
+@pytest.mark.asyncio
+async def test_migration_uses_frozen_sources_and_stored_embeddings(
+ tmp_path, monkeypatch
+):
+ pytest.importorskip("lancedb")
+ from datamind import runtime
+
+ embed = Embed()
+ monkeypatch.setattr(runtime, "build_embedding", lambda *a, **kw: embed)
+ root = (tmp_path / "source").resolve()
+ root.mkdir()
+ cfg = Settings(
+ _env_file=None,
+ data={"base_dir": tmp_path / "runtime"},
+ embedding={"model": "fixture", "dimension": 4},
+ workspace={"retention": "all", "storage_mode": "snapshot", "auto_gc": False},
+ )
+ old = Workspace(cfg)
+ first = await publish(old, root, surfaces=("kb", "graph"))
+ assert embed.calls == 1
+ (root / "owner.md").unlink()
+ cfg.kb.vector_store = "lancedb"
+ cfg.workspace.storage_mode = "incremental"
+ w = Workspace(cfg)
+ result = await w.migrate()
+ assert result["migrated_from"] == first["workspace_revision"]
+ assert result["format_version"] == 2
+ assert embed.calls == 1
+ for revision in (first["workspace_revision"], result["workspace_revision"]):
+ async with w.reader(revision) as reader:
+ assert "Alice" in str(await reader.call("kb_search", query="owner"))
+ assert "Alice" in str(
+ await reader.call("graph_neighbors", entity="Project")
+ )
+ assert (
+ "Alice"
+ in (await reader.call("raw_file_read", path=str(root / "owner.md")))[
+ "text"
+ ]
+ )
+
+
+@pytest.mark.asyncio
+async def test_mcp_operation_pins_across_calls_and_close_releases_history(
+ incremental, monkeypatch
+):
+ from datamind import mcp
+
+ w, root, _ = incremental
+ a = await publish(w, root)
+ monkeypatch.setattr(mcp, "load_settings", lambda profile=None: w.settings)
+ operation = await mcp.execute("datamind_operation_begin", {"mode": "read"})
+ args = {"operation_id": operation["operation_id"], "entity": "Project"}
+ assert "Alice" in str(await mcp.execute("datamind_graph_neighbors", args))
+ b = await publish(w, root, "Bob")
+ w.configure_retention("latest")
+ cleanup = await w.collect(dry_run=False)
+ assert a["workspace_revision"] in cleanup["retained_revisions"]
+ assert "Alice" in str(await mcp.execute("datamind_graph_neighbors", args))
+ with pytest.raises(PermissionError):
+ await mcp.execute(
+ "datamind_workspace_retention",
+ {"operation_id": operation["operation_id"], "mode": "all"},
+ )
+ await mcp.execute(
+ "datamind_operation_close", {"operation_id": operation["operation_id"]}
+ )
+ await w.collect(dry_run=False)
+ assert w.status(a["workspace_revision"])["state"] == "PRUNED"
+ fresh = await mcp.execute("datamind_operation_begin", {"mode": "read"})
+ assert fresh["revision"] == b["workspace_revision"]
+ await mcp.execute(
+ "datamind_operation_close", {"operation_id": fresh["operation_id"]}
+ )
+
+
+@pytest.mark.asyncio
+async def test_expired_operations_fail_instead_of_switching_to_latest(
+ incremental, monkeypatch
+):
+ from datamind import mcp
+
+ w, root, _ = incremental
+ await publish(w, root)
+ monkeypatch.setattr(mcp, "load_settings", lambda profile=None: w.settings)
+ operation = await mcp.execute("datamind_operation_begin", {"mode": "read"})
+ await publish(w, root, "Bob")
+ with w._connect() as db:
+ db.execute("UPDATE leases SET expires=0")
+ w.configure_retention("latest")
+ await w.collect(dry_run=False)
+ with pytest.raises(CapabilityError, match="expired"):
+ await mcp.execute(
+ "datamind_graph_schema", {"operation_id": operation["operation_id"]}
+ )
+ assert await mcp.execute(
+ "datamind_operation_close", {"operation_id": operation["operation_id"]}
+ ) == {"closed": True}
+
+
+@pytest.mark.asyncio
+async def test_failed_native_build_does_not_expose_rows_and_reuses_embeddings(
+ incremental, monkeypatch
+):
+ pytest.importorskip("lancedb")
+ import datamind.workspace as module
+
+ w, root, embed = incremental
+ a = await publish(w, root, surfaces=("kb", "graph"))
+ original = module._atomic_json
+
+ def fail_manifest(path, value, **kwargs):
+ if path.name == "manifest.json":
+ raise OSError("injected before publication")
+ return original(path, value, **kwargs)
+
+ with monkeypatch.context() as patch:
+ patch.setattr(module, "_atomic_json", fail_manifest)
+ with pytest.raises(OSError):
+ await publish(w, root, "Bob", surfaces=("kb", "graph"))
+ assert w.current_revision == a["workspace_revision"]
+ async with w.reader() as reader:
+ assert "Alice" in str(await reader.call("kb_search", query="owner"))
+ assert "Bob" not in str(await reader.call("kb_search", query="owner"))
+ calls = embed.calls
+ b = await publish(Workspace(w.settings), root, "Bob", surfaces=("kb", "graph"))
+ assert embed.calls == calls
+ assert b["metrics"]["sources_reused"] == 1
+ async with w.reader() as reader:
+ assert "Bob" in str(await reader.call("kb_search", query="owner"))
+
+
+@pytest.mark.asyncio
+async def test_rollback_aba_during_build_fails_epoch_cas(incremental, monkeypatch):
+ import datamind.workspace as module
+
+ w, root, _ = incremental
+ a = await publish(w, root)
+ b = await publish(w, root, "Bob")
+ original = module._atomic_json
+
+ def rollback_between_prepare_and_publish(path, value, **kwargs):
+ result = original(path, value, **kwargs)
+ if path.name == "manifest.json":
+ w.rollback(
+ a["workspace_revision"], expected_revision=b["workspace_revision"]
+ )
+ w.rollback(
+ b["workspace_revision"], expected_revision=a["workspace_revision"]
+ )
+ return result
+
+ monkeypatch.setattr(module, "_atomic_json", rollback_between_prepare_and_publish)
+ with pytest.raises(CapabilityError, match="conflict"):
+ await publish(w, root, "Carol")
+ assert w.current_revision == b["workspace_revision"]
+ async with w.reader() as reader:
+ assert "Bob" in str(await reader.call("graph_neighbors", entity="Project"))
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("crash_point", ["build.json", "manifest.json"])
+async def test_process_death_before_publish_is_recovered_without_partial_visibility(
+ incremental,
+ crash_point,
+):
+ import subprocess, sys, textwrap
+
+ w, root, _ = incremental
+ a = await publish(w, root)
+ code = """
+import asyncio,hashlib,os,sys
+from pathlib import Path
+from datamind.config import Settings
+from datamind.workspace import Workspace
+import datamind.workspace as module
+cfg=Settings(_env_file=None,data={'base_dir':sys.argv[1]},workspace={'storage_mode':'incremental','retention':'all','auto_gc':False},embedding={'model':'fixture','dimension':4},kb={'vector_store':'lancedb'})
+w=Workspace(cfg);root=Path(sys.argv[2]);doc=root/'owner.md';doc.write_text('Bob owns Project')
+original=module._atomic_json
+def crash(path,value,**kwargs):
+ if path.name==sys.argv[3]:os._exit(91)
+ original(path,value,**kwargs)
+module._atomic_json=crash
+asyncio.run(w.update(str(root),surfaces=['graph'],graph_sources=[{'uri':str(doc),'version':'sha256:'+hashlib.sha256(doc.read_bytes()).hexdigest(),'triples':[{'subject':'Bob','relation':'owns','object':'Project'}]}]))
+"""
+ result = await asyncio.to_thread(
+ subprocess.run,
+ [
+ sys.executable,
+ "-c",
+ textwrap.dedent(code),
+ str(w.settings.data.base_dir),
+ str(root),
+ crash_point,
+ ],
+ capture_output=True,
+ text=True,
+ timeout=30,
+ )
+ assert result.returncode == 91, result.stderr
+ fresh = Workspace(w.settings)
+ assert fresh.current_revision == a["workspace_revision"]
+ async with fresh.reader() as reader:
+ assert "Alice" in str(await reader.call("graph_neighbors", entity="Project"))
+ building = [
+ r["revision"] for r in fresh.history()["versions"] if r["state"] == "BUILDING"
+ ]
+ assert len(building) == 1
+ await fresh.collect(dry_run=False)
+ assert fresh.status(building[0])["state"] == "PRUNED"
+ b = await publish(fresh, root, "Bob")
+ async with fresh.reader(b["workspace_revision"]) as reader:
+ assert "Bob" in str(await reader.call("graph_neighbors", entity="Project"))
+
+
+@pytest.mark.asyncio
+async def test_graph_parallel_sources_retract_independently_and_cycles_do_not_hide_paths(
+ incremental,
+):
+ from datamind.workspace_graph import put_graph, GraphView
+
+ w, root, _ = incremental
+ first = put_graph(
+ w.root,
+ [
+ {"subject": "A", "relation": "owns", "object": "A", "source": "one"},
+ {"subject": "A", "relation": "owns", "object": "B", "source": "one"},
+ ],
+ )
+ second = put_graph(
+ w.root, [{"subject": "A", "relation": "owns", "object": "B", "source": "two"}]
+ )
+ view = GraphView(w.root, [first, second])
+ assert len(await view.neighbors("A", direction="out")) == 3
+ assert (await view.traverse("A", max_results=1))[0].nodes == ["A", "B"]
+ await view.aclose()
+ view = GraphView(w.root, [second])
+ edges = await view.neighbors("A")
+ assert len(edges) == 1 and edges[0].properties["source"] == "two"
+ await view.aclose()
+
+
+@pytest.mark.asyncio
+async def test_returning_to_old_embedding_index_keeps_old_reader_valid(incremental):
+ pytest.importorskip("lancedb")
+ w, root, _ = incremental
+ a = await publish(w, root, surfaces=("kb", "graph"))
+ async with w.reader(a["workspace_revision"]) as old:
+ w.settings.embedding.model = "fixture-b"
+ await publish(w, root, "Bob", surfaces=("kb", "graph"))
+ w.settings.embedding.model = "fixture"
+ await publish(w, root, "Carol", surfaces=("kb", "graph"))
+ assert "Alice" in str(await old.call("kb_search", query="owner"))
+ assert "Carol" not in str(await old.call("kb_search", query="owner"))
+ async with w.reader() as reader:
+ assert "Carol" in str(await reader.call("kb_search", query="owner"))
+
+
+@pytest.mark.asyncio
+async def test_cancelled_storage_worker_cannot_write_after_releasing_writer_lock(
+ tmp_path,
+):
+ import threading
+ from filelock import AsyncFileLock
+ from datamind.workspace_native import blocking
+
+ entered = threading.Event()
+ finish = threading.Event()
+ events = []
+
+ def write():
+ entered.set()
+ assert finish.wait(5)
+ events.append("finished write")
+
+ async def build():
+ async with AsyncFileLock(str(tmp_path / "writer.lock")):
+ await blocking(write)
+
+ task = asyncio.create_task(build())
+ assert await asyncio.to_thread(entered.wait, 5)
+ task.cancel()
+ await asyncio.sleep(0)
+ assert not task.done()
+
+ async def next_build():
+ async with AsyncFileLock(str(tmp_path / "writer.lock")):
+ events.append("next writer")
+
+ next_task = asyncio.create_task(next_build())
+ finish.set()
+ with pytest.raises(asyncio.CancelledError):
+ await task
+ await asyncio.wait_for(next_task, 5)
+ assert events == ["finished write", "next writer"]
+
+
+@pytest.mark.asyncio
+async def test_open_graph_reader_does_not_block_next_publication(incremental):
+ w, root, _ = incremental
+ a = await publish(w, root)
+ async with w.reader(a["workspace_revision"]) as reader:
+ assert "Alice" in str(await reader.call("graph_neighbors", entity="Project"))
+ await asyncio.wait_for(publish(w, root, "Bob"), 5)
+ assert "Alice" in str(await reader.call("graph_neighbors", entity="Project"))
+ w.configure_retention("latest")
+ await asyncio.wait_for(w.collect(dry_run=False, compact=True), 5)
+ assert "Alice" in str(await reader.call("graph_neighbors", entity="Project"))
+
+
+def test_saved_retention_overrides_config_defaults_after_restart(tmp_path):
+ cfg = Settings(
+ _env_file=None, data={"base_dir": tmp_path}, workspace={"retention": "all"}
+ )
+ w = Workspace(cfg)
+ assert w.retention["mode"] == "all"
+ w.configure_retention("latest")
+ assert Workspace(cfg).retention["mode"] == "latest"
diff --git a/datamind/tests/test_workspace_native_db.py b/datamind/tests/test_workspace_native_db.py
new file mode 100644
index 0000000..95add3b
--- /dev/null
+++ b/datamind/tests/test_workspace_native_db.py
@@ -0,0 +1,269 @@
+"""Real Dolt integration: shared snapshots, pruning and rollback baselines."""
+
+import hashlib
+import shutil
+import pytest
+from datamind.config import Settings
+from datamind.workspace import Workspace
+from datamind.core.errors import CapabilityError
+
+pytestmark = pytest.mark.skipif(
+ shutil.which("dolt") is None, reason="requires real Dolt binary"
+)
+
+
+@pytest.mark.asyncio
+async def test_dolt_snapshots_reuse_unchanged_tables_and_prune_history(tmp_path):
+ root = (tmp_path / "sources").resolve()
+ root.mkdir()
+ sales = root / "sales.csv"
+ sales.write_text("amount\n100\n200\n")
+ other = root / "other.csv"
+ other.write_text("item\nunchanged\n")
+ cfg = Settings(
+ _env_file=None,
+ data={"base_dir": tmp_path / "runtime"},
+ db={"dialect": "dolt"},
+ workspace={"storage_mode": "incremental", "retention": "all", "auto_gc": False},
+ )
+ w = Workspace(cfg)
+ a = await w.update(str(root), surfaces=["db"])
+ b = await w.update(str(root), surfaces=["db"])
+ assert a["snapshots"]["db"] == b["snapshots"]["db"]
+ assert b["metrics"]["db_tables_changed"] == 0
+ assert b["metrics"]["db_tables_reused"] == 2
+ sales.write_text("amount\n500\n")
+ c = await w.update(str(root), surfaces=["db"])
+ assert c["metrics"]["db_tables_changed"] == 1
+ assert c["metrics"]["db_tables_reused"] == 1
+
+ async def total(revision):
+ async with w.reader(revision) as reader:
+ names = (await reader.call("db_list_tables"))["tables"]
+ for name in names:
+ if "amount" in str(await reader.call("db_describe_table", table=name)):
+ table = name
+ break
+ result = await reader.call(
+ "db_query_sql",
+ sql=f"SELECT SUM(CAST(amount AS DOUBLE)) AS total FROM `{table}`",
+ )
+ with pytest.raises(CapabilityError):
+ await reader.call(
+ "db_query_sql", sql=f"SELECT * FROM `dolt/main`.`{table}`"
+ )
+ return result["rows"][0][0]
+
+ assert await total(a["workspace_revision"]) == 300
+ assert await total(c["workspace_revision"]) == 500
+ w.rollback(a["workspace_revision"], expected_revision=c["workspace_revision"])
+ other.write_text("item\nchanged\n")
+ sales.write_text("amount\n100\n200\n")
+ d = await w.update(str(root), surfaces=["db"])
+ assert await total(d["workspace_revision"]) == 300
+ w.configure_retention("latest")
+ cleanup = await w.collect(dry_run=False, compact=True)
+ assert cleanup["providers"]["db"]["physical_gc_pending"] is False
+ assert await total(d["workspace_revision"]) == 300
+ with pytest.raises(CapabilityError):
+ w.manifest(a["workspace_revision"])
+ assert len(list((w.root / "shared" / "dolt").glob(".dolt"))) == 1
+
+
+@pytest.mark.asyncio
+async def test_migrate_all_surfaces_and_publish_new_shared_generation(
+ tmp_path, monkeypatch
+):
+ pytest.importorskip("lancedb")
+ from datamind import runtime
+
+ class Embed:
+ name = "fixture"
+ dimension = 4
+ calls = 0
+
+ async def embed_texts(self, texts):
+ self.calls += len(texts)
+ return [[1.0, 0.0, 0.0, 0.0] for _ in texts]
+
+ async def embed_query(self, text):
+ return [1.0, 0.0, 0.0, 0.0]
+
+ embed = Embed()
+ monkeypatch.setattr(runtime, "build_embedding", lambda *a, **kw: embed)
+ root = (tmp_path / "source").resolve()
+ root.mkdir()
+ doc = root / "owner.md"
+ doc.write_text("Alice owns Project")
+ sales = root / "sales.csv"
+ sales.write_text("amount\n100\n200\n")
+ cfg = Settings(
+ _env_file=None,
+ data={"base_dir": tmp_path / "runtime"},
+ embedding={"model": "fixture", "dimension": 4},
+ workspace={"storage_mode": "snapshot", "retention": "all", "auto_gc": False},
+ )
+ graph = [
+ {
+ "uri": str(doc),
+ "version": "sha256:" + hashlib.sha256(doc.read_bytes()).hexdigest(),
+ "triples": [{"subject": "Alice", "relation": "owns", "object": "Project"}],
+ }
+ ]
+ old = Workspace(cfg)
+ a = await old.update(str(root), graph_sources=graph)
+ old_calls = embed.calls
+ doc.unlink()
+ sales.unlink()
+ cfg.workspace.storage_mode = "incremental"
+ cfg.kb.vector_store = "lancedb"
+ cfg.db.dialect = "dolt"
+ w = Workspace(cfg)
+ b = await w.migrate()
+ assert b["migrated_from"] == a["workspace_revision"]
+ assert embed.calls == old_calls
+
+ async def total(reader):
+ tables = (await reader.call("db_list_tables"))["tables"]
+ assert len(tables) == 1
+ result = await reader.call(
+ "db_query_sql",
+ sql=f"SELECT SUM(CAST(amount AS DOUBLE)) AS total FROM `{tables[0]}`",
+ )
+ return result["rows"][0][0]
+
+ for revision in (a["workspace_revision"], b["workspace_revision"]):
+ async with w.reader(revision) as reader:
+ assert await total(reader) == 300
+ assert "Alice" in str(await reader.call("kb_search", query="owner"))
+ assert "Alice" in str(
+ await reader.call("graph_neighbors", entity="Project")
+ )
+ assert (
+ "Alice" in (await reader.call("raw_file_read", path=str(doc)))["text"]
+ )
+ doc.write_text("Bob owns Project")
+ sales.write_text("amount\n500\n")
+ graph[0].update(
+ version="sha256:" + hashlib.sha256(doc.read_bytes()).hexdigest(),
+ triples=[{"subject": "Bob", "relation": "owns", "object": "Project"}],
+ )
+ async with w.reader(b["workspace_revision"]) as old_reader:
+ await w.update(str(root), graph_sources=graph)
+ w.configure_retention("latest")
+ await w.collect(dry_run=False, compact=True)
+ assert await total(old_reader) == 300
+ assert "Alice" in str(await old_reader.call("kb_search", query="owner"))
+ assert "Alice" in str(
+ await old_reader.call("graph_neighbors", entity="Project")
+ )
+ await w.collect(dry_run=False, compact=True)
+ assert w.status(a["workspace_revision"])["state"] == "PRUNED"
+ async with w.reader() as reader:
+ assert await total(reader) == 500
+ assert "Bob" in str(await reader.call("kb_search", query="owner"))
+ assert "Bob" in str(await reader.call("graph_neighbors", entity="Project"))
+ table = (await reader.call("db_list_tables"))["tables"][0]
+ assert (
+ await reader.call(
+ "db_query_sql",
+ sql=f"WITH a AS (SELECT amount FROM `{table}`) SELECT * FROM a",
+ )
+ )["rows"] == [["500"]]
+ for sql in (
+ f"SELECT * FROM `{table}` AS OF HEAD",
+ "SELECT * FROM mysql.user",
+ "SELECT * FROM dolt_log",
+ ):
+ with pytest.raises(CapabilityError):
+ await reader.call("db_query_sql", sql=sql)
+
+
+@pytest.mark.asyncio
+async def test_interrupted_dolt_child_is_reaped_on_next_operation(tmp_path):
+ import asyncio, subprocess, sys, textwrap
+
+ root = (tmp_path / "source").resolve()
+ root.mkdir()
+ (root / "sales.csv").write_text("amount\n100\n")
+ cfg = Settings(
+ _env_file=None,
+ data={"base_dir": tmp_path / "runtime with spaces"},
+ db={"dialect": "dolt"},
+ workspace={"storage_mode": "incremental", "retention": "all", "auto_gc": False},
+ )
+ w = Workspace(cfg)
+ first = await w.update(str(root), surfaces=["db"])
+ code = """
+import asyncio,os,sys
+from contextlib import asynccontextmanager
+from datamind.config import Settings
+from datamind.workspace import Workspace
+from datamind.workspace_native import DoltRepository
+cfg=Settings(_env_file=None,data={'base_dir':sys.argv[1]},db={'dialect':'dolt'},workspace={'storage_mode':'incremental','retention':'all','auto_gc':False})
+w=Workspace(cfg)
+async def crash():
+ async with DoltRepository(w).session(): os._exit(92)
+asyncio.run(crash())
+"""
+ result = await asyncio.to_thread(
+ subprocess.run,
+ [sys.executable, "-c", textwrap.dedent(code), str(cfg.data.base_dir)],
+ capture_output=True,
+ text=True,
+ timeout=40,
+ )
+ assert result.returncode == 92, result.stderr
+ # The parent is dead, but its private SQL child is still running. A fresh
+ # process/object must recover it without touching other Dolt instances.
+ fresh = Workspace(cfg)
+ async with fresh.reader(first["workspace_revision"]) as reader:
+ table = (await reader.call("db_list_tables"))["tables"][0]
+ result = await reader.call("db_query_sql", sql=f"SELECT amount FROM `{table}`")
+ assert result["rows"] == [["100"]]
+ (root / "sales.csv").write_text("amount\n200\n")
+ updated = await fresh.update(str(root), surfaces=["db"])
+ assert updated["workspace_revision"] != first["workspace_revision"]
+
+
+@pytest.mark.asyncio
+async def test_mcp_native_publication_does_not_open_independent_dolt_store(
+ tmp_path, monkeypatch
+):
+ from datamind import mcp
+
+ root = (tmp_path / "source").resolve()
+ root.mkdir()
+ (root / "sales.csv").write_text("amount\n100\n")
+ cfg = Settings(
+ _env_file=None,
+ data={"base_dir": tmp_path / "runtime"},
+ db={"dialect": "dolt"},
+ hooks={"path_allowlist_extra": [str(root)]},
+ workspace={"storage_mode": "incremental", "retention": "all", "auto_gc": False},
+ )
+ monkeypatch.setattr(mcp, "load_settings", lambda profile=None: cfg)
+ operation = await mcp.execute("datamind_operation_begin", {"mode": "write"})
+ result = await mcp.execute(
+ "datamind_workspace_publish",
+ {
+ "operation_id": operation["operation_id"],
+ "path": str(root),
+ "surfaces": ["db"],
+ },
+ )
+ receipt = result["result"]
+ assert receipt["results"][0]["status"] == "stored", receipt
+ assert receipt["results"][0]["details"]["format_version"] == 2
+ assert not (cfg.data.storage_dir / ".dolt").exists()
+ await mcp.execute(
+ "datamind_operation_close", {"operation_id": operation["operation_id"]}
+ )
+ read = await mcp.execute("datamind_operation_begin", {"mode": "read"})
+ tables = await mcp.execute(
+ "datamind_db_list_tables", {"operation_id": read["operation_id"]}
+ )
+ assert len(tables["result"]["tables"]) == 1
+ await mcp.execute(
+ "datamind_operation_close", {"operation_id": read["operation_id"]}
+ )
diff --git a/datamind/tests/test_workspace_revisions.py b/datamind/tests/test_workspace_revisions.py
index 64b363d..83c8023 100644
--- a/datamind/tests/test_workspace_revisions.py
+++ b/datamind/tests/test_workspace_revisions.py
@@ -39,7 +39,7 @@ def setup_workspace(tmp_path, monkeypatch):
model, embedding = Model(), Embed()
monkeypatch.setattr(options, 'build_embedding', lambda *a, **k: embedding)
settings = Settings(llm={'api_key': 'test', 'model': 'test'},
- data={'base_dir': tmp_path / 'runtime'}, embedding={'dimension': 4})
+ data={'base_dir': tmp_path / 'runtime'}, embedding={'dimension': 4}, workspace={'retention':'all'})
source = tmp_path / 'source'
source.mkdir()
(source / 'a.md').write_text('Alice owns Project')
diff --git a/datamind/workspace.py b/datamind/workspace.py
index 1e2bf69..e129304 100644
--- a/datamind/workspace.py
+++ b/datamind/workspace.py
@@ -2,9 +2,9 @@
The catalog is safe across processes on one host. Builds prepare private indexes;
publication uses compare-and-swap against their starting revision. Readers retain
-one immutable revision across all tool calls. This local snapshot implementation
-supports SQLite, NetworkX, and local Chroma/LanceDB; it is not a remote-store
-snapshot coordinator. Source checkpoints avoid repeated model work after restart.
+one immutable revision across all tool calls. Incremental storage shares source
+objects, source-scoped graph assertions, LanceDB versions and Dolt table trees.
+Legacy local snapshots remain readable and can be migrated explicitly.
"""
from __future__ import annotations
@@ -27,6 +27,7 @@
from datamind.core.logging import bind_context, current_context
from datamind.core.protocols import GraphTriple, GraphEvidence
from datamind.core.recovery import recover_read, recovery_summary
+from datamind.workspace_catalog import WorkspaceCatalog
def _digest(value: bytes) -> str:
@@ -52,7 +53,7 @@ def _atomic_json(path: Path, value: Any, *, overwrite: bool = True) -> None:
temporary.unlink(missing_ok=True)
-class Workspace:
+class Workspace(WorkspaceCatalog):
"""A profile-scoped snapshot catalog. Pass separate settings per tenant."""
def __init__(self, settings, *, builder=None):
@@ -70,6 +71,7 @@ def __init__(self, settings, *, builder=None):
''')
if 'epoch' not in {row[1] for row in conn.execute('PRAGMA table_info(head)')}:
conn.execute('ALTER TABLE head ADD COLUMN epoch INTEGER NOT NULL DEFAULT 0')
+ self._init_catalog()
def _connect(self):
return sqlite3.connect(self._catalog, timeout=30)
@@ -92,13 +94,17 @@ def manifest(self, revision: str) -> dict[str, Any]:
raise CapabilityError('workspace', 'revision was never published')
return json.loads((self.root / 'revisions' / revision / 'manifest.json').read_text())
- def rollback(self, revision: str, *, expected_revision: str) -> dict[str, Any]:
+ def rollback(self, revision: str, *, expected_revision: str | None = None) -> dict[str, Any]:
"""Repoint to a retained complete snapshot without deleting history."""
+ with self._lock():
+ return self._rollback(revision, expected_revision=expected_revision)
+
+ def _rollback(self, revision, *, expected_revision):
self.manifest(revision)
with self._connect() as conn:
conn.execute('BEGIN IMMEDIATE')
head = conn.execute('SELECT revision FROM head WHERE singleton=1').fetchone()[0]
- if head != expected_revision:
+ if expected_revision is not None and head != expected_revision:
raise CapabilityError('workspace', 'publication conflict; current revision changed')
conn.execute('UPDATE head SET revision=?, epoch=epoch+1 WHERE singleton=1', (revision,))
return {'workspace_revision': revision, 'previous_revision': head}
@@ -115,6 +121,10 @@ async def _system(self, settings, surfaces):
from datamind.data_plane import build_data_plane
return await build_data_plane(settings, enable=set(surfaces))
+ async def migrate(self):
+ from datamind.workspace_migration import migrate
+ return await migrate(self)
+
def _inventory(self, source: Path) -> list[dict[str, str]]:
if not source.is_dir():
raise CapabilityError('workspace', 'workspace path must be a directory')
@@ -127,8 +137,11 @@ def _inventory(self, source: Path) -> list[dict[str, str]]:
continue
if not path.resolve().is_relative_to(source):
raise CapabilityError('workspace', f'symlink escapes workspace: {relative}')
+ sha = hashlib.sha256()
+ with path.open('rb') as handle:
+ for block in iter(lambda: handle.read(1024 * 1024), b''): sha.update(block)
inventory.append({'relative_path': relative.as_posix(), 'uri': str(path.resolve()),
- 'version': 'sha256:' + _digest(path.read_bytes())})
+ 'version': 'sha256:' + sha.hexdigest()})
return inventory
async def update(self, path: str, *, surfaces=('kb', 'graph', 'db'), relation_types=None, max_triples_per_file=200, graph_sources=None) -> dict[str, Any]:
@@ -148,10 +161,15 @@ async def update(self, path: str, *, surfaces=('kb', 'graph', 'db'), relation_ty
ctx.extra.update(nested_model_usage=[], embedding_usage=[])
started = time.monotonic()
try:
+ from datamind.workspace_incremental import incremental_enabled, update_incremental
+ update = update_incremental if incremental_enabled(self, surfaces) else Workspace._update
with bind_context(ctx):
- result = await self._update(path, surfaces=surfaces, relation_types=relation_types,
+ result = await update(self, path, surfaces=surfaces, relation_types=relation_types,
max_triples_per_file=max_triples_per_file, metrics=metrics, graph_sources=graph_sources)
metrics["status"] = "published"
+ if self.settings.workspace.auto_gc:
+ try:metrics["retention_cleanup"] = await self.collect(dry_run=False,compact=self.retention['mode']!='all')
+ except Exception as exc:metrics["cleanup_error"] = type(exc).__name__
return {**result, "metrics": metrics}
except BaseException as exc:
metrics.update(status="failed", error_type=type(exc).__name__)
@@ -168,14 +186,15 @@ async def update(self, path: str, *, surfaces=('kb', 'graph', 'db'), relation_ty
if parent is not None:
parent.extra.setdefault("nested_model_usage", []).extend(usage)
parent.extra.setdefault("embedding_usage", []).extend(embed_usage)
- if metrics.get("workspace_revision"):
- report = self._settings_for(metrics["workspace_revision"]).data.base_dir / "build_metrics.json"
- try:
- _atomic_json(report, metrics)
- except OSError:
- # An observability failure must not undo a published head or
- # replace the original build error.
- metrics["metrics_persisted"] = False
+ revision = metrics.get('workspace_revision')
+ if revision:
+ # GC cannot remove a revision between checking it and persisting
+ # metrics; a metrics write must never resurrect a pruned directory.
+ with self._lock():
+ report = self.root / 'revisions' / revision / 'build_metrics.json'
+ if report.parent.is_dir():
+ try: _atomic_json(report, metrics)
+ except OSError: metrics['metrics_persisted'] = False
async def _update(self, path: str, *, surfaces, relation_types, max_triples_per_file, metrics, graph_sources=None) -> dict[str, Any]:
"""Prepare and publish one generation, retaining the previous one on failure.
@@ -331,7 +350,8 @@ async def _update(self, path: str, *, surfaces, relation_types, max_triples_per_
await ingest._ingest_rows(table=table, rows=safe_rows, safe_cols=safe_cols,
if_exists='replace', source_file=item['uri'])
manifest_sources.append({**item, 'source_id': source_id,
- 'snapshot': str(snapshot.relative_to(settings.data.base_dir))})
+ 'snapshot': str(snapshot.relative_to(settings.data.base_dir)),
+ 'checkpoint': str(cache_path.relative_to(self.root))})
metrics['sources_completed'] += 1
metrics['materialization_ms'] += (time.monotonic() - stage_started) * 1000
stage_started = time.monotonic()
@@ -385,22 +405,41 @@ async def _update(self, path: str, *, surfaces, relation_types, max_triples_per_
@asynccontextmanager
async def reader(self, revision: str | None = None, *, read_retries: int = 1, enable=None):
- revision = revision or self.current_revision
+ token,revision=self.acquire_lease(revision)
if revision is None:
raise CapabilityError('workspace', 'no published revision')
- manifest = self.manifest(revision)
- settings = self._settings_for(revision)
- selected = set(manifest['surfaces']) if enable is None else set(enable)
- if not selected <= set(manifest['surfaces']):
- raise CapabilityError('workspace','requested surface is absent from the pinned revision')
- system = await self._system(settings, selected)
+ system=None
+ heartbeat=asyncio.create_task(self.keepalive(token))
try:
- reader = WorkspaceReader(system, manifest, settings.data.base_dir, read_retries=read_retries,
- timeout_s=settings.agent.wall_clock_timeout_s)
- reader._system_settings = self.settings
+ manifest=self.manifest(revision)
+ selected=set(manifest['surfaces']) if enable is None else set(enable)
+ if not selected<=set(manifest['surfaces']):
+ raise CapabilityError('workspace','requested surface is absent from the pinned revision')
+ if manifest.get('format_version')==2:
+ from datamind.workspace_incremental import reader_system
+ system,settings=await reader_system(self,manifest,selected)
+ root=self.root
+ else:
+ settings=self._settings_for(revision)
+ if manifest.get('snapshots',{}).get('kb'):
+ settings.kb.vector_store=manifest['snapshots']['kb']['provider']
+ if manifest.get('snapshots',{}).get('db'):
+ settings.db.dialect=manifest['snapshots']['db']['provider'];settings.db.dsn=None
+ settings.graph.backend='networkx'
+ system=await self._system(settings,selected)
+ root=settings.data.base_dir
+ reader=WorkspaceReader(system,manifest,root,read_retries=read_retries,
+ timeout_s=settings.agent.wall_clock_timeout_s)
+ reader._system_settings=self.settings
+ reader._lease=(self,token)
yield reader
finally:
- await system.aclose()
+ heartbeat.cancel()
+ await asyncio.gather(heartbeat,return_exceptions=True)
+ try:
+ if system is not None: await system.aclose()
+ finally: self.release_lease(token)
+
async def query(self, message: str, *, revision: str | None = None, read_retries: int = 1, **kwargs):
async with self.reader(revision, read_retries=read_retries) as reader:
@@ -441,6 +480,15 @@ async def read_snapshot(path: str, **kwargs):
async def inspect_snapshot(path: str, **kwargs):
candidate = Path(path).expanduser().resolve()
+ if manifest.get('format_version')==2:
+ if candidate!=source_root and not candidate.is_relative_to(source_root):
+ raise CapabilityError('workspace','inspection must belong to the pinned revision')
+ items=[{'path':s['uri'],'relative_path':s['relative_path'],'sha256':s['version'].removeprefix('sha256:')}
+ for s in manifest['sources'] if Path(s['uri']).is_relative_to(candidate) or Path(s['uri'])==candidate]
+ limit=kwargs.get('max_files',2000)
+ return {'path':str(candidate),'files':items[:limit],'count':len(items[:limit]),'truncated':len(items)>limit,
+ 'workspace_revision':self.revision}
+
if candidate.is_relative_to(source_root):
candidate = snapshot_root / candidate.relative_to(source_root)
elif not candidate.is_relative_to(snapshot_root):
@@ -469,6 +517,8 @@ async def retry_read(_handler=original_handler, _name=name, **kwargs):
tools.replace(replace(spec, handler=retry_read))
async def call(self, name: str, **kwargs):
+ if hasattr(self,'_lease'):
+ self._lease[0].renew_lease(self._lease[1])
parent = current_context()
if parent is not None:
parent.extra.setdefault('recovery_events', [])
diff --git a/datamind/workspace_catalog.py b/datamind/workspace_catalog.py
new file mode 100644
index 0000000..7a92660
--- /dev/null
+++ b/datamind/workspace_catalog.py
@@ -0,0 +1,355 @@
+"""Single-host workspace retention and durable reader leases.
+
+All reference changes and GC run under one OS lock. Files are immutable;
+publication is still a compare-and-swap in the catalog, not a multi-store commit.
+"""
+
+from __future__ import annotations
+import json
+import shutil
+import time
+import uuid
+from contextlib import contextmanager
+from filelock import FileLock
+from datamind.core.errors import CapabilityError
+
+
+class WorkspaceCatalog:
+ def _init_catalog(self):
+ with self._connect() as db:
+ db.executescript("""
+ CREATE TABLE IF NOT EXISTS workspace_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
+ CREATE TABLE IF NOT EXISTS leases (
+ id TEXT PRIMARY KEY, revision TEXT NOT NULL, expires REAL NOT NULL, kind TEXT NOT NULL);
+ CREATE INDEX IF NOT EXISTS leases_revision ON leases(revision);
+ CREATE TABLE IF NOT EXISTS protected_versions (revision TEXT PRIMARY KEY);
+ CREATE TABLE IF NOT EXISTS incremental_builds (revision TEXT PRIMARY KEY);
+ """)
+ legacy = db.execute("SELECT COUNT(*) FROM revisions").fetchone()[0] > 0
+ db.execute(
+ "INSERT OR IGNORE INTO workspace_meta VALUES (?,?)",
+ (
+ "retention",
+ json.dumps(
+ {
+ "mode": "all" if legacy else "latest",
+ "keep_last": self.settings.workspace.keep_last,
+ }
+ ),
+ ),
+ )
+
+ def _lock(self):
+ return FileLock(
+ str(self.root / "references.lock"),
+ timeout=self.settings.workspace.lock_timeout_s,
+ )
+
+ @property
+ def retention(self):
+ with self._connect() as db:
+ value = json.loads(
+ db.execute(
+ "SELECT value FROM workspace_meta WHERE key='retention'"
+ ).fetchone()[0]
+ )
+ cfg = self.settings.workspace
+ if not value.get("explicit") and cfg.retention != "auto":
+ return {"mode": cfg.retention, "keep_last": cfg.keep_last}
+ return {k: value[k] for k in ("mode", "keep_last")}
+
+ def configure_retention(self, mode, keep_last=10):
+ if (
+ mode not in {"latest", "recent", "all"}
+ or not isinstance(keep_last, int)
+ or isinstance(keep_last, bool)
+ or keep_last < 1
+ ):
+ raise ValueError(
+ "retention must be latest, recent, or all; keep_last must be positive"
+ )
+ value = {"mode": mode, "keep_last": keep_last}
+ with self._lock(), self._connect() as db:
+ db.execute(
+ "INSERT OR REPLACE INTO workspace_meta VALUES (?,?)",
+ ("retention", json.dumps({**value, "explicit": True})),
+ )
+ # An explicit API policy takes effect on this instance too.
+ self.settings.workspace.retention = "auto"
+ return value
+
+ def storage_usage(self):
+ sizes = {}
+ for name in (
+ "objects",
+ "vectors",
+ "tables",
+ "artifacts",
+ "shared",
+ "revisions",
+ "checkpoints",
+ ):
+ sizes[name] = sum(
+ p.stat().st_size for p in (self.root / name).rglob("*") if p.is_file()
+ )
+ graph = self.root / "graph.sqlite3"
+ sizes["graph"] = graph.stat().st_size if graph.exists() else 0
+ return {
+ "bytes": sum(sizes.values()),
+ "components": sizes,
+ "note": "Allocated file sizes; database free pages and filesystem compression may differ.",
+ }
+
+ async def keepalive(self, token):
+ import asyncio
+
+ while token:
+ await asyncio.sleep(min(30, self.settings.workspace.lease_ttl_s / 3))
+ self.renew_lease(token)
+
+ def history(self):
+ with self._connect() as db:
+ rows = db.execute(
+ "SELECT id,parent,state,error FROM revisions ORDER BY rowid DESC"
+ ).fetchall()
+ return {
+ "current_revision": self.current_revision,
+ "retention": self.retention,
+ "versions": [
+ dict(zip(("revision", "parent", "state", "error"), r)) for r in rows
+ ],
+ }
+
+ def acquire_lease(self, revision=None, *, kind="reader"):
+ with self._lock(), self._connect() as db:
+ db.execute("BEGIN IMMEDIATE")
+ if revision is None:
+ revision = db.execute(
+ "SELECT revision FROM head WHERE singleton=1"
+ ).fetchone()[0]
+ if revision is None:
+ return None, None
+ row = db.execute(
+ "SELECT state FROM revisions WHERE id=?", (revision,)
+ ).fetchone()
+ if row is None or row[0] not in (
+ {"BUILDING"} if kind == "build" else {"PUBLISHED"}
+ ):
+ raise CapabilityError(
+ "workspace", "revision is unavailable or has been pruned"
+ )
+ token = uuid.uuid4().hex
+ db.execute(
+ "INSERT INTO leases VALUES (?,?,?,?)",
+ (
+ token,
+ revision,
+ time.time() + self.settings.workspace.lease_ttl_s,
+ kind,
+ ),
+ )
+ return token, revision
+
+ def renew_lease(self, token):
+ with self._lock(), self._connect() as db:
+ row = db.execute(
+ "SELECT revision,expires FROM leases WHERE id=?", (token,)
+ ).fetchone()
+ if row is None or row[1] <= time.time():
+ raise CapabilityError(
+ "workspace", "workspace lease expired; start a new operation"
+ )
+ db.execute(
+ "UPDATE leases SET expires=? WHERE id=?",
+ (time.time() + self.settings.workspace.lease_ttl_s, token),
+ )
+ return row[0]
+
+ def release_lease(self, token):
+ if token:
+ with self._lock(), self._connect() as db:
+ db.execute("DELETE FROM leases WHERE id=?", (token,))
+
+ @contextmanager
+ def pinned(self, revision=None):
+ token, revision = self.acquire_lease(revision)
+ try:
+ yield revision
+ finally:
+ self.release_lease(token)
+
+ def protect(self, revision, enabled=True):
+ with self._lock():
+ self.manifest(revision)
+ with self._connect() as db:
+ if enabled:
+ db.execute(
+ "INSERT OR IGNORE INTO protected_versions VALUES (?)",
+ (revision,),
+ )
+ else:
+ db.execute(
+ "DELETE FROM protected_versions WHERE revision=?", (revision,)
+ )
+ return {"revision": revision, "protected": enabled}
+
+ async def collect(self, *, dry_run=True, compact=False):
+ """Mark/sweep immutable objects, retaining active readers and builders.
+
+ Provider compaction is separate and must honor all retained bindings.
+ A failed cleanup is retryable; tombstoned versions never become readable.
+ """
+ from filelock import AsyncFileLock
+
+ # Same acquisition order as incremental builds: writer, then references.
+ async with AsyncFileLock(
+ str(self.root / "writer.lock"),
+ timeout=self.settings.workspace.lock_timeout_s,
+ ):
+ with self._lock():
+ now = time.time()
+ with self._connect() as db:
+ rows = db.execute(
+ "SELECT id,state FROM revisions ORDER BY rowid DESC"
+ ).fetchall()
+ leases = db.execute(
+ "SELECT revision FROM leases WHERE expires>?", (now,)
+ ).fetchall()
+ active = {r[0] for r in leases}
+ roots = {
+ r[0]
+ for r in db.execute("SELECT revision FROM protected_versions")
+ }
+ head = db.execute(
+ "SELECT revision FROM head WHERE singleton=1"
+ ).fetchone()[0]
+ if head:
+ roots.add(head)
+ roots |= active
+ published = [r for r, state in rows if state == "PUBLISHED"]
+ policy = self.retention
+ if policy["mode"] == "all":
+ roots.update(published)
+ elif policy["mode"] == "recent":
+ roots.update(published[: policy["keep_last"]])
+ # Old builders have no heartbeat protocol. Never remove them.
+ building = set()
+ native_builds = {
+ r[0]
+ for r in db.execute("SELECT revision FROM incremental_builds")
+ }
+ for r, state in rows:
+ if state != "BUILDING":
+ continue
+ marker = self.root / "revisions" / r / "build.json"
+ if r in native_builds or (
+ marker.exists()
+ and json.loads(marker.read_text()).get("format_version")
+ == 2
+ ):
+ if not dry_run:
+ db.execute(
+ "UPDATE revisions SET state='FAILED',error='interrupted incremental build' WHERE id=?",
+ (r,),
+ )
+ else:
+ building.add(r)
+ if building:
+ return {
+ "dry_run": dry_run,
+ "deferred": "build in progress or requires recovery",
+ "retention": policy,
+ "retained_revisions": sorted(roots),
+ "pruned_revisions": [],
+ "reclaimed_bytes": 0,
+ }
+ candidates = [
+ r
+ for r, state in rows
+ if (
+ state in {"PUBLISHED", "PRUNED", "FAILED"}
+ or (state == "BUILDING" and r not in building)
+ )
+ and r not in roots
+ ]
+ retained = []
+ for r in roots:
+ path = self.root / "revisions" / r / "manifest.json"
+ if path.is_file():
+ retained.append(json.loads(path.read_text()))
+ referenced = set()
+ for manifest in retained:
+ for item in manifest["sources"]:
+ if manifest.get("format_version") == 2:
+ referenced.add(item["snapshot"])
+ referenced.add(item["artifact"])
+ artifact = self.root / item["artifact"]
+ if artifact.exists():
+ value = json.loads(artifact.read_text())
+ referenced.update(
+ c["vector_ref"] for c in value.get("chunks", [])
+ )
+ referenced.update(
+ t["path"] for t in value.get("tables", [])
+ )
+ elif item.get("checkpoint"):
+ referenced.add(item["checkpoint"])
+ files = [
+ p
+ for prefix in (".source-", ".table-", ".migrate-table-")
+ for p in self.root.glob(prefix + "*")
+ if p.is_file()
+ ]
+ for folder in ("objects", "artifacts", "vectors", "tables"):
+ for path in (self.root / folder).glob("*/*"):
+ if (
+ path.is_file()
+ and path.relative_to(self.root).as_posix()
+ not in referenced
+ ):
+ files.append(path)
+ # V1 caches without explicit references remain until migrated.
+ if retained and all(m.get("format_version") == 2 for m in retained):
+ files.extend(
+ p for p in (self.root / "checkpoints").glob("*.json")
+ )
+ size = sum(p.stat().st_size for p in files)
+ directories = [self.root / "revisions" / r for r in candidates]
+ size += sum(
+ p.stat().st_size
+ for d in directories
+ for p in d.rglob("*")
+ if p.is_file()
+ )
+ result = {
+ "dry_run": dry_run,
+ "retention": policy,
+ "retained_revisions": sorted(roots),
+ "pruned_revisions": candidates,
+ "reclaimable_bytes": size,
+ "reclaimed_bytes": 0,
+ }
+ if dry_run:
+ return result
+ db.executemany(
+ "UPDATE revisions SET state='PRUNED' WHERE id=?",
+ [(r,) for r in candidates],
+ )
+ db.execute("DELETE FROM leases WHERE expires<=?", (now,))
+ # Native versions are removed only after catalog invalidation, so
+ # interrupted cleanup cannot leave a readable dangling manifest.
+ from datamind.workspace_native import prune_native
+
+ result["providers"] = await prune_native(self, retained, compact=compact)
+ from datamind.workspace_graph import prune_graph
+
+ from datamind.workspace_native import blocking
+
+ await blocking(prune_graph, self.root, retained, compact=compact)
+ for path in files:
+ path.unlink(missing_ok=True)
+ for directory in directories:
+ if directory.exists():
+ shutil.rmtree(directory)
+ result["reclaimed_bytes"] = size
+ return result
diff --git a/datamind/workspace_graph.py b/datamind/workspace_graph.py
new file mode 100644
index 0000000..49f5a8e
--- /dev/null
+++ b/datamind/workspace_graph.py
@@ -0,0 +1,228 @@
+"""Immutable, source-scoped graph assertions with indexed snapshot reads."""
+
+from __future__ import annotations
+import difflib
+import hashlib
+import json
+import sqlite3
+from collections import Counter
+from contextlib import closing
+from pathlib import Path
+from datamind.core.protocols import Edge, Entity, GraphPath, GraphTriple
+
+
+def connect(root):
+ db = sqlite3.connect(Path(root) / "graph.sqlite3", timeout=30)
+ db.execute("PRAGMA journal_mode=WAL")
+ db.executescript("""
+ CREATE TABLE IF NOT EXISTS artifacts (id TEXT PRIMARY KEY, count INTEGER NOT NULL);
+ CREATE TABLE IF NOT EXISTS edges (
+ artifact TEXT NOT NULL, identity TEXT NOT NULL, src TEXT NOT NULL, dst TEXT NOT NULL,
+ rel TEXT NOT NULL, payload TEXT NOT NULL, PRIMARY KEY(artifact,identity));
+ CREATE INDEX IF NOT EXISTS graph_out ON edges(src,artifact,rel);
+ CREATE INDEX IF NOT EXISTS graph_in ON edges(dst,artifact,rel);
+ CREATE TABLE IF NOT EXISTS nodes (
+ artifact TEXT NOT NULL, id TEXT NOT NULL, type TEXT NOT NULL,
+ PRIMARY KEY(artifact,id));
+ CREATE INDEX IF NOT EXISTS graph_entities ON nodes(id,artifact);
+ """)
+ return db
+
+
+def put_graph(root, triples):
+ values = [GraphTriple(**t).model_dump(mode="json") for t in triples]
+ value = json.dumps(values, sort_keys=True, ensure_ascii=False)
+ key = hashlib.sha256(value.encode()).hexdigest()
+ with closing(connect(root)) as db, db:
+ if not db.execute("SELECT 1 FROM artifacts WHERE id=?", (key,)).fetchone():
+ for t in values:
+ identity = json.dumps(
+ [t["subject"], t["relation"], t["object"], t["source"]],
+ ensure_ascii=False,
+ )
+ db.execute(
+ "INSERT OR REPLACE INTO edges VALUES (?,?,?,?,?,?)",
+ (
+ key,
+ identity,
+ t["subject"],
+ t["object"],
+ t["relation"],
+ json.dumps(t, ensure_ascii=False),
+ ),
+ )
+ for node, typ in (
+ (t["subject"], t["subject_type"]),
+ (t["object"], t["object_type"]),
+ ):
+ db.execute(
+ "INSERT OR REPLACE INTO nodes VALUES (?,?,?)", (key, node, typ)
+ )
+ db.execute("INSERT INTO artifacts VALUES (?,?)", (key, len(values)))
+ return key
+
+
+def prune_graph(root, manifests, *, compact=False):
+ if not (Path(root) / "graph.sqlite3").exists():
+ return
+ keep = {
+ s["graph_artifact"]
+ for m in manifests
+ for s in m["sources"]
+ if s.get("graph_artifact")
+ }
+ with closing(connect(root)) as db, db:
+ for (key,) in db.execute("SELECT id FROM artifacts").fetchall():
+ if key not in keep:
+ db.execute("DELETE FROM edges WHERE artifact=?", (key,))
+ db.execute("DELETE FROM nodes WHERE artifact=?", (key,))
+ db.execute("DELETE FROM artifacts WHERE id=?", (key,))
+ db.commit()
+ if compact:
+ db.execute("VACUUM")
+ db.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+
+
+class GraphView:
+ def __init__(self, root, artifacts):
+ self.db = connect(root)
+ self.db.execute("CREATE TEMP TABLE visible (id TEXT PRIMARY KEY)")
+ self.db.executemany(
+ "INSERT OR IGNORE INTO visible VALUES (?)", [(k,) for k in artifacts]
+ )
+ self.db.commit() # The manifest pins visibility; never hold a SQLite read transaction across calls.
+
+ def _nodes(self, where="", args=()):
+ # One deterministic type per entity; no other revision enters this view.
+ return self.db.execute(
+ "SELECT n.id,MIN(n.type) FROM nodes n JOIN visible v ON n.artifact=v.id "
+ + where
+ + " GROUP BY n.id",
+ args,
+ )
+
+ def stats(self):
+ return {
+ "nodes": self.db.execute(
+ "SELECT COUNT(DISTINCT n.id) FROM nodes n JOIN visible v ON n.artifact=v.id"
+ ).fetchone()[0],
+ "edges": self.db.execute(
+ "SELECT COUNT(*) FROM edges e JOIN visible v ON e.artifact=v.id"
+ ).fetchone()[0],
+ }
+
+ def schema(self):
+ stats = self.stats()
+ types = Counter(typ for _, typ in self._nodes())
+ relations = dict(
+ self.db.execute(
+ "SELECT e.rel,COUNT(*) FROM edges e JOIN visible v ON e.artifact=v.id GROUP BY e.rel ORDER BY e.rel"
+ )
+ )
+ return {
+ **stats,
+ "node_types": dict(types),
+ "relations": relations,
+ "extensible": True,
+ }
+
+ async def search_entities(self, query, *, top_k=5):
+ q = query.lower().strip()
+ if not q:
+ return []
+ scored = []
+ # Iterate, retaining only top-k candidates; do not load the whole graph.
+ for node, typ in self._nodes():
+ if node.lower() == q:
+ score = 1.0
+ elif q in node.lower():
+ score = 0.7
+ else:
+ ratio = difflib.SequenceMatcher(None, q, node.lower()).ratio()
+ if ratio < 0.4:
+ continue
+ score = 0.4 + 0.4 * ratio
+ scored.append(Entity(id=node, label=node, type=typ, score=score))
+ scored.sort(key=lambda e: (-e.score, e.id))
+ del scored[top_k:]
+ return scored
+
+ def _edges(self, entity, *, direction="both", relation_filter=None, limit=None):
+ cond = {"out": "e.src=?", "in": "e.dst=?", "both": "(e.src=? OR e.dst=?)"}[
+ direction
+ ]
+ args = [entity, entity] if direction == "both" else [entity]
+ if relation_filter:
+ cond += " AND e.rel IN (" + ",".join("?" for _ in relation_filter) + ")"
+ args += list(relation_filter)
+ sql = (
+ "SELECT e.payload FROM edges e JOIN visible v ON e.artifact=v.id WHERE "
+ + cond
+ + " ORDER BY e.dst,e.rel,e.identity,e.artifact"
+ )
+ if limit is not None:
+ sql += " LIMIT ?"
+ args.append(limit)
+ rows = self.db.execute(sql, args)
+ for (payload,) in rows:
+ t = json.loads(payload)
+ yield Edge(
+ source=t["subject"],
+ target=t["object"],
+ relation=t["relation"],
+ weight=t["confidence"],
+ properties={
+ **t["properties"],
+ **({"source": t["source"]} if t["source"] else {}),
+ },
+ evidence=t["evidence"],
+ )
+
+ async def neighbors(
+ self, entity, *, direction="both", relation_filter=None, limit=100
+ ):
+ if limit <= 0:
+ return []
+ return list(
+ self._edges(
+ entity,
+ direction=direction,
+ relation_filter=relation_filter,
+ limit=limit,
+ )
+ )
+
+ async def traverse(
+ self, start, *, max_hops=2, relation_filter=None, max_results=100
+ ):
+ if max_hops <= 0 or max_results <= 0:
+ return []
+ frontier = [(start, [], {start})]
+ paths = []
+ for _ in range(max_hops):
+ next_frontier = []
+ for node, prior, seen in frontier:
+ # Bounded BFS retains exact edge sequences, including origins.
+ for edge in self._edges(
+ node, direction="out", relation_filter=relation_filter
+ ):
+ if edge.target in seen:
+ continue
+ edges = prior + [edge]
+ paths.append(
+ GraphPath(
+ nodes=[start] + [e.target for e in edges],
+ edges=edges,
+ score=sum(e.weight for e in edges) / len(edges),
+ )
+ )
+ if len(paths) >= max_results:
+ return sorted(paths, key=lambda p: -p.score)
+ next_frontier.append((edge.target, edges, seen | {edge.target}))
+ frontier = next_frontier
+ if not frontier:
+ break
+ return sorted(paths, key=lambda p: -p.score)
+
+ async def aclose(self):
+ self.db.close()
diff --git a/datamind/workspace_incremental.py b/datamind/workspace_incremental.py
new file mode 100644
index 0000000..5b32b0f
--- /dev/null
+++ b/datamind/workspace_incremental.py
@@ -0,0 +1,803 @@
+"""Shared immutable source objects plus native KB/DB snapshot bindings."""
+
+from __future__ import annotations
+import csv
+import hashlib
+import json
+import os
+import re
+import struct
+import tempfile
+import time
+import uuid
+from contextlib import asynccontextmanager
+from pathlib import Path
+from filelock import AsyncFileLock
+from datamind.core.errors import CapabilityError, ConfigError
+from datamind.core.protocols import GraphTriple, GraphEvidence
+from datamind.core.logging import current_context
+from datamind.capabilities.ingest.formats import (
+ DOCUMENT_EXTS,
+ TABLE_EXTS,
+ extract_document,
+)
+from datamind.capabilities.kb.indexer import _hash, _split_text
+from datamind.workspace_native import blocking, completed
+
+
+def digest(value):
+ return hashlib.sha256(value).hexdigest()
+
+
+def encoded(value):
+ return json.dumps(value, sort_keys=True, ensure_ascii=False, default=str).encode()
+
+
+def incremental_enabled(workspace, surfaces):
+ mode = workspace.settings.workspace.storage_mode
+ if mode == "snapshot":
+ return False
+ if mode == "incremental":
+ return True
+ head = workspace.current_revision
+ if head:
+ return workspace.manifest(head).get("format_version") == 2
+ cfg = workspace.settings
+ return cfg.kb.vector_store == "lancedb" and cfg.db.dialect == "dolt"
+
+
+def store_bytes(root, kind, data, suffix):
+ key = digest(data)
+ path = root / kind / key[:2] / (key + suffix)
+ if path.exists():
+ return path.relative_to(root).as_posix(), False
+ path.parent.mkdir(parents=True, exist_ok=True)
+ fd, temp = tempfile.mkstemp(dir=path.parent, prefix=".writing-")
+ try:
+ with os.fdopen(fd, "wb") as handle:
+ handle.write(data)
+ handle.flush()
+ os.fsync(handle.fileno())
+ os.replace(temp, path)
+ finally:
+ Path(temp).unlink(missing_ok=True)
+ return path.relative_to(root).as_posix(), True
+
+
+def save_artifact(root, key, value):
+ from datamind.workspace import _atomic_json
+
+ path = root / "artifacts" / key[:2] / (key + ".json")
+ _atomic_json(path, value, overwrite=False)
+ return path.relative_to(root).as_posix()
+
+
+def canonical_tables(workspace, item, snapshot):
+ """Stream CSV/TSV and XLSX rows into immutable typed-cell table objects."""
+ source = Path(item["uri"])
+ streams = []
+ closers = []
+ if source.suffix.lower() in {".csv", ".tsv"}:
+ handle = snapshot.open(newline="", encoding="utf-8", errors="replace")
+ closers.append(handle)
+ rows = iter(
+ csv.reader(
+ handle, delimiter="\t" if source.suffix.lower() == ".tsv" else ","
+ )
+ )
+ header = next(rows, None)
+ if header is not None:
+ streams.append((source.stem, header, rows))
+ elif source.suffix.lower() == ".xlsx":
+ import openpyxl
+
+ book = openpyxl.load_workbook(snapshot, read_only=True, data_only=True)
+ closers.append(book)
+ for sheet in book.worksheets:
+ rows = iter(sheet.values)
+ header = next(rows, None)
+ if header is not None:
+ streams.append((sheet.title, header, rows))
+ else:
+ raise ConfigError(
+ "Incremental DB supports CSV, TSV and XLSX; convert legacy XLS first"
+ )
+ output = []
+ try:
+ for sheet, header, rows in streams:
+ used = set()
+ cols = []
+ for i, cell in enumerate(header, 1):
+ value = str(cell).strip() if cell is not None else ""
+ if (
+ not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,63}", value)
+ or value.casefold() in used
+ or value.casefold() == "_source_file"
+ ):
+ value = "col_" + str(i)
+ base = value
+ n = 2
+ while value.casefold() in used:
+ value = base[:58] + "_" + str(n)
+ n += 1
+ cols.append(value)
+ used.add(value.casefold())
+ cols.append("_source_file")
+ fd, temp = tempfile.mkstemp(dir=workspace.root, prefix=".table-")
+ count = 0
+ sha = hashlib.sha256()
+ try:
+ with os.fdopen(fd, "w", encoding="utf-8", newline="") as handle:
+ writer = csv.writer(handle)
+ writer.writerow(cols)
+ for row in rows:
+ values = list(row[: len(header)])
+ values += [""] * (len(header) - len(values))
+ # Preserve existing TEXT import semantics, including NULL.
+ values = [None if x is None else str(x) for x in values] + [
+ item["uri"]
+ ]
+ writer.writerow(
+ [json.dumps(x, ensure_ascii=False) for x in values]
+ )
+ count += 1
+ handle.flush()
+ os.fsync(handle.fileno())
+ with open(temp, "rb") as handle:
+ for block in iter(lambda: handle.read(1024 * 1024), b""):
+ sha.update(block)
+ key = sha.hexdigest()
+ target = workspace.root / "tables" / key[:2] / (key + ".csv")
+ target.parent.mkdir(parents=True, exist_ok=True)
+ if not target.exists():
+ os.replace(temp, target)
+ source_id = "file:" + digest(item["uri"].encode())
+ output.append(
+ {
+ "name": "ws_" + digest((source_id + sheet).encode())[:24],
+ "sheet": sheet,
+ "columns": cols,
+ "rows": count,
+ "hash": key,
+ "path": target.relative_to(workspace.root).as_posix(),
+ }
+ )
+ finally:
+ Path(temp).unlink(missing_ok=True)
+ finally:
+ for closer in closers:
+ closer.close()
+ return output
+
+
+@asynccontextmanager
+async def _writer(workspace):
+ if getattr(workspace, "_writer_held", False):
+ yield
+ else:
+ async with AsyncFileLock(
+ str(workspace.root / "writer.lock"),
+ timeout=workspace.settings.workspace.lock_timeout_s,
+ ):
+ yield
+
+
+async def update_incremental(
+ workspace,
+ path,
+ *,
+ surfaces,
+ relation_types,
+ max_triples_per_file,
+ metrics,
+ graph_sources,
+):
+ from datamind.workspace import _atomic_json
+ from datamind.workspace_graph import put_graph
+ from datamind.workspace_native import DoltRepository
+
+ selected = set(surfaces)
+ cfg = workspace.settings
+ if not selected or not selected <= {"kb", "graph", "db"}:
+ raise ConfigError("Invalid workspace surfaces")
+ if not 1 <= max_triples_per_file <= 10000:
+ raise ConfigError("Invalid graph extraction limit")
+ if "kb" in selected and cfg.kb.vector_store != "lancedb":
+ raise ConfigError(
+ "Incremental Workspace KB requires lancedb; configure kb.vector_store or use snapshot mode"
+ )
+ if "db" in selected and (cfg.db.dialect != "dolt" or cfg.db.dsn):
+ raise ConfigError(
+ "Incremental Workspace DB requires local Dolt without an external DSN"
+ )
+ if "graph" in selected and graph_sources is None:
+ raise ConfigError("Incremental Workspace requires Codex-prepared graph_sources")
+ source = Path(path).expanduser().resolve()
+ started = time.monotonic()
+ inventory = await blocking(workspace._inventory, source)
+ metrics.update(
+ inventory_ms=(time.monotonic() - started) * 1000,
+ sources_total=len(inventory),
+ object_bytes_written=0,
+ objects_reused=0,
+ kb_rows_upserted=0,
+ kb_rows_deleted=0,
+ db_tables_changed=0,
+ db_tables_reused=0,
+ )
+ # Capture the expected parent BEFORE acquiring the writer lock. A queued
+ # stale build must conflict, not silently rebase to another caller's update.
+ with workspace._connect() as db:
+ parent, epoch = db.execute(
+ "SELECT revision,epoch FROM head WHERE singleton=1"
+ ).fetchone()
+ if (
+ hasattr(workspace, "_expected_parent")
+ and (parent, epoch) != workspace._expected_parent
+ ):
+ raise CapabilityError("workspace", "publication conflict during migration")
+ if (
+ parent
+ and workspace.manifest(parent).get("format_version") != 2
+ and not hasattr(workspace, "_migration_prepared")
+ ):
+ raise ConfigError(
+ "Run explicit workspace migration before changing an existing snapshot Workspace to incremental storage"
+ )
+ old = workspace.manifest(parent) if parent else None
+ prior = (
+ {s["uri"]: s for s in old["sources"]}
+ if old and old.get("format_version") == 2
+ else {}
+ )
+ assertions = {}
+ docs = {
+ s["uri"]: s
+ for s in inventory
+ if Path(s["uri"]).suffix.lower() in DOCUMENT_EXTS - TABLE_EXTS
+ }
+ for entry in graph_sources or []:
+ uri = entry["uri"]
+ if (
+ uri not in docs
+ or uri in assertions
+ or entry["version"] != docs[uri]["version"]
+ ):
+ raise ConfigError("Graph source must match inventory and content hash")
+ triples = [GraphTriple(**v) for v in entry["triples"]]
+ if len(triples) > max_triples_per_file:
+ raise ConfigError("Too many graph triples for source")
+ for triple in triples:
+ if relation_types is not None and triple.relation not in relation_types:
+ raise ConfigError("Graph relation is outside vocabulary")
+ triple.source = uri
+ triple.evidence = [
+ GraphEvidence(
+ source_id="file:" + digest(uri.encode()),
+ source_version=entry["version"],
+ locator={"uri": uri},
+ method="extracted",
+ pipeline_version="codex-explicit-v1",
+ )
+ ]
+ assertions[uri] = {
+ "triples": [t.model_dump(mode="json") for t in triples],
+ "truncated": bool(entry.get("truncated")),
+ }
+ if "graph" in selected:
+ for uri, item in docs.items():
+ if uri not in assertions:
+ before = prior.get(uri)
+ if (
+ before is None
+ or before["version"] != item["version"]
+ or not before.get("graph_artifact")
+ ):
+ raise ConfigError(
+ "Supply graph assertions for every new or changed document, including empty lists"
+ )
+ # Unchanged assertions are immutable and reused, not re-extracted.
+ context = current_context()
+ skills = list(context.extra.get("applied_skills", {}).values()) if context else []
+ recipe = {
+ "version": "workspace-v2",
+ "surfaces": sorted(selected),
+ "embedding": cfg.embedding.model_dump(mode="json", exclude={"api_key"}),
+ "chunk_size": cfg.retrieval.chunk_size,
+ "chunk_overlap": cfg.retrieval.chunk_overlap,
+ "relation_types": relation_types,
+ "max_triples_per_file": max_triples_per_file,
+ "applied_skills": skills,
+ }
+ pipeline = digest(encoded(recipe))
+ revision = "revision-" + uuid.uuid4().hex
+ metrics["workspace_revision"] = revision
+ async with _writer(workspace):
+ with workspace._lock(), workspace._connect() as db:
+ if db.execute(
+ "SELECT revision,epoch FROM head WHERE singleton=1"
+ ).fetchone() != (parent, epoch):
+ raise CapabilityError(
+ "workspace", "publication conflict; current revision changed"
+ )
+ db.execute(
+ "INSERT INTO revisions VALUES (?,?,?,NULL)",
+ (revision, parent, "BUILDING"),
+ )
+ db.execute("INSERT INTO incremental_builds VALUES (?)", (revision,))
+ root = workspace.root / "revisions" / revision
+ _atomic_json(root / "build.json", {"format_version": 2, "parent": parent})
+ embedding = None
+ sources = []
+ bindings = {}
+ all_chunks = {}
+ tables = {}
+ try:
+ if "kb" in selected:
+ from datamind import runtime
+
+ embedding = runtime.build_embedding(cfg.embedding, fallback_llm=cfg.llm)
+ for item in inventory:
+ start = time.monotonic()
+ original = Path(
+ getattr(workspace, "_migration_sources", {}).get(
+ item["uri"], item["uri"]
+ )
+ )
+ suffix = Path(item["uri"]).suffix.lower()
+ blob = (
+ workspace.root
+ / "objects"
+ / item["version"][7:9]
+ / (item["version"][7:] + suffix)
+ )
+ if blob.exists():
+ blob_ref = blob.relative_to(workspace.root).as_posix()
+ metrics["objects_reused"] += 1
+ else:
+ # Stream the original into immutable object storage.
+ fd, temp = tempfile.mkstemp(dir=workspace.root, prefix=".source-")
+ sha = hashlib.sha256()
+ try:
+ with os.fdopen(fd, "wb") as out, original.open("rb") as inp:
+ for block in iter(lambda: inp.read(1024 * 1024), b""):
+ out.write(block)
+ sha.update(block)
+ out.flush()
+ os.fsync(out.fileno())
+ if "sha256:" + sha.hexdigest() != item["version"]:
+ raise CapabilityError(
+ "workspace", "source changed during build; retry"
+ )
+ blob.parent.mkdir(parents=True, exist_ok=True)
+ os.replace(temp, blob)
+ blob_ref = blob.relative_to(workspace.root).as_posix()
+ metrics["object_bytes_written"] += blob.stat().st_size
+ finally:
+ Path(temp).unlink(missing_ok=True)
+ metrics["source_copy_ms"] += (time.monotonic() - start) * 1000
+ assertion = assertions.get(item["uri"])
+ graph_ref = (
+ prior.get(item["uri"], {}).get("graph_artifact")
+ if assertion is None
+ else None
+ )
+ key = digest(
+ encoded(
+ [
+ item["uri"],
+ item["version"],
+ pipeline,
+ assertion if assertion is not None else graph_ref,
+ ]
+ )
+ )
+ artifact = workspace.root / "artifacts" / key[:2] / (key + ".json")
+ start = time.monotonic()
+ if artifact.exists():
+ prepared = json.loads(artifact.read_text())
+ metrics["sources_reused"] += 1
+ metrics["checkpoint_read_ms"] += (time.monotonic() - start) * 1000
+ else:
+ # Fast reuse for omitted, unchanged graph assertions.
+ previous = prior.get(item["uri"], {})
+ if (
+ assertion is None
+ and old
+ and old["pipeline_hash"] == pipeline
+ and previous.get("version") == item["version"]
+ ):
+ artifact = workspace.root / previous["artifact"]
+ prepared = json.loads(artifact.read_text())
+ metrics["sources_reused"] += 1
+ elif item["uri"] in getattr(workspace, "_migration_prepared", {}):
+ prepared = dict(workspace._migration_prepared[item["uri"]])
+ if "graph" in selected:
+ prepared["graph_artifact"] = put_graph(
+ workspace.root,
+ assertion["triples"] if assertion else [],
+ )
+ artifact = workspace.root / save_artifact(
+ workspace.root, key, prepared
+ )
+ metrics["sources_reused"] += 1
+ else:
+ prepared = {
+ "chunks": [],
+ "tables": [],
+ "graph_artifact": None,
+ "truncated": False,
+ }
+ if "kb" in selected and suffix in DOCUMENT_EXTS:
+ text_value = (await blocking(extract_document, blob)).text
+ parts = _split_text(
+ text_value,
+ chunk_size=cfg.retrieval.chunk_size,
+ chunk_overlap=cfg.retrieval.chunk_overlap,
+ )
+ vector_recipe = cfg.embedding.model_dump(
+ mode="json", exclude={"api_key"}
+ )
+ missing = []
+ refs = []
+ for part in parts:
+ vector_key = digest(encoded([vector_recipe, part]))
+ ref = (
+ "vectors/"
+ + vector_key[:2]
+ + "/"
+ + vector_key
+ + ".f32"
+ )
+ refs.append(ref)
+ if not (workspace.root / ref).exists() and ref not in {
+ r for _, r in missing
+ }:
+ missing.append((part, ref))
+ if missing:
+ vectors = await embedding.embed_texts(
+ [p for p, _ in missing]
+ )
+ if len(vectors) != len(missing):
+ raise CapabilityError(
+ "kb", "embedding count mismatch"
+ )
+ from datamind.workspace import _atomic_json
+
+ for (_, ref), vector in zip(missing, vectors):
+ if len(vector) != embedding.dimension:
+ raise CapabilityError(
+ "kb", "embedding dimension mismatch"
+ )
+ dest = workspace.root / ref
+ dest.parent.mkdir(parents=True, exist_ok=True)
+ data = struct.pack("<" + "f" * len(vector), *vector)
+ fd, temporary = tempfile.mkstemp(
+ dir=dest.parent, prefix=".vector-"
+ )
+ try:
+ with os.fdopen(fd, "wb") as handle:
+ handle.write(data)
+ handle.flush()
+ os.fsync(handle.fileno())
+ os.replace(temporary, dest)
+ finally:
+ Path(temporary).unlink(missing_ok=True)
+ metrics["embedding_calls"] += 1
+ metrics["embedding_texts"] += len(missing)
+ prepared["chunks"] = [
+ {
+ "id": _hash(part, item["uri"], ordinal=i),
+ "text": part,
+ "vector_ref": refs[i],
+ "metadata": {
+ "source": item["uri"],
+ "source_id": "file:"
+ + digest(item["uri"].encode()),
+ "source_version": item["version"],
+ "_chunk_ordinal": i,
+ },
+ }
+ for i, part in enumerate(parts)
+ ]
+ if "graph" in selected:
+ if assertion is not None:
+ prepared["graph_artifact"] = put_graph(
+ workspace.root, assertion["triples"]
+ )
+ prepared["truncated"] = assertion["truncated"]
+ elif graph_ref:
+ prepared["graph_artifact"] = graph_ref
+ else:
+ prepared["graph_artifact"] = put_graph(
+ workspace.root, []
+ )
+ if "db" in selected and suffix in TABLE_EXTS:
+ prepared["tables"] = await blocking(
+ canonical_tables, workspace, item, blob
+ )
+ metrics["parsed_table_files"] += 1
+ artifact = workspace.root / save_artifact(
+ workspace.root, key, prepared
+ )
+ metrics["sources_computed"] += 1
+ metrics["source_compute_ms"] += (
+ time.monotonic() - start
+ ) * 1000
+ source_entry = {
+ **item,
+ "source_id": "file:" + digest(item["uri"].encode()),
+ "snapshot": blob_ref,
+ "artifact": artifact.relative_to(workspace.root).as_posix(),
+ "graph_artifact": prepared["graph_artifact"],
+ "truncated": prepared["truncated"],
+ }
+ sources.append(source_entry)
+ all_chunks.update((c["id"], c) for c in prepared["chunks"])
+ tables.update((t["name"], t) for t in prepared["tables"])
+ metrics["sources_completed"] += 1
+ start = time.monotonic()
+ if "kb" in selected:
+ from datamind.capabilities.kb.providers.lancedb_store import (
+ LanceDBVectorStore,
+ )
+
+ kb_key = digest(
+ encoded(
+ [
+ cfg.embedding.model_dump(mode="json", exclude={"api_key"}),
+ embedding.dimension,
+ ]
+ )
+ )
+ location = "shared/kb/" + kb_key
+ store = LanceDBVectorStore(
+ persist_dir=str(workspace.root / location),
+ collection_name="workspace",
+ dimension=embedding.dimension,
+ )
+ base = (
+ old.get("snapshots", {}).get("kb")
+ if old and old.get("format_version") == 2
+ else None
+ )
+ previous_chunks = (
+ base.get("chunks", {}) if base and base["path"] == location else {}
+ )
+ if (
+ base
+ and base["path"] == location
+ and store.latest_version != base["version"]
+ ):
+ await blocking(store._collection.restore, base["version"])
+ elif not previous_chunks and store.existing_count:
+ await blocking(store._collection.delete, "true")
+ new_fingerprints = {
+ key: digest(encoded(c)) for key, c in all_chunks.items()
+ }
+ removed = set(previous_chunks) - set(all_chunks)
+ if removed:
+ await completed(store.delete(sorted(removed)))
+ changed = [
+ c
+ for key, c in all_chunks.items()
+ if previous_chunks.get(key) != new_fingerprints[key]
+ ]
+ for offset in range(0, len(changed), 256):
+ batch = changed[offset : offset + 256]
+ vectors = []
+ for chunk in batch:
+ data = (workspace.root / chunk["vector_ref"]).read_bytes()
+ vectors.append(
+ list(struct.unpack("<" + "f" * (len(data) // 4), data))
+ )
+ await completed(
+ store.add(
+ ids=[c["id"] for c in batch],
+ texts=[c["text"] for c in batch],
+ embeddings=vectors,
+ metadatas=[c["metadata"] for c in batch],
+ )
+ )
+ version = store.latest_version
+ tag = "dm_workspace_" + str(version)
+ if tag not in store._collection.tags.list():
+ await blocking(store._collection.tags.create, tag, version)
+ bindings["kb"] = {
+ "provider": "lancedb",
+ "path": location,
+ "version": version,
+ "dimension": embedding.dimension,
+ "embedding_fingerprint": digest(
+ encoded(
+ cfg.embedding.model_dump(mode="json", exclude={"api_key"})
+ )
+ ),
+ "chunks": new_fingerprints,
+ }
+ metrics["kb_rows_upserted"] = len(changed)
+ metrics["kb_rows_deleted"] = len(removed)
+ if "db" in selected:
+ base = (
+ old.get("snapshots", {}).get("db")
+ if old and old.get("format_version") == 2
+ else None
+ )
+ hashes = {n: t["hash"] for n, t in tables.items()}
+ if base and base["tables"] == hashes:
+ bindings["db"] = base
+ else:
+ bindings["db"] = await DoltRepository(workspace).build(
+ base, tables, revision
+ )
+ before = base.get("tables", {}) if base else {}
+ metrics["db_tables_reused"] = sum(
+ before.get(n) == v for n, v in hashes.items()
+ )
+ metrics["db_tables_changed"] = (
+ len(hashes)
+ - metrics["db_tables_reused"]
+ + len(before.keys() - hashes.keys())
+ )
+ if "graph" in selected:
+ bindings["graph"] = {
+ "provider": "sqlite-source-graph",
+ "path": "graph.sqlite3",
+ }
+ metrics["materialization_ms"] += (time.monotonic() - start) * 1000
+ if await blocking(workspace._inventory, source) != inventory:
+ raise CapabilityError(
+ "workspace", "workspace changed during build; retry"
+ )
+ manifest = {
+ "format_version": 2,
+ "workspace_revision": revision,
+ "parent_revision": parent,
+ "parent_epoch": epoch,
+ "source_root": str(source),
+ "surfaces": sorted(selected),
+ "pipeline_hash": pipeline,
+ "sources": sources,
+ "snapshots": bindings,
+ "reused_sources": metrics["sources_reused"],
+ "applied_skills": skills,
+ "truncated_sources": [s["uri"] for s in sources if s["truncated"]],
+ }
+ _atomic_json(root / "manifest.json", manifest)
+ start = time.monotonic()
+ with workspace._lock(), workspace._connect() as db:
+ db.execute("BEGIN IMMEDIATE")
+ if db.execute(
+ "SELECT revision,epoch FROM head WHERE singleton=1"
+ ).fetchone() != (parent, epoch):
+ raise CapabilityError(
+ "workspace",
+ "publication conflict; another build published first",
+ )
+ db.execute(
+ "UPDATE revisions SET state='PUBLISHED' WHERE id=?", (revision,)
+ )
+ db.execute(
+ "UPDATE head SET revision=?,epoch=epoch+1 WHERE singleton=1",
+ (revision,),
+ )
+ metrics["publication_ms"] = (time.monotonic() - start) * 1000
+ return manifest
+ except BaseException as exc:
+ with workspace._connect() as db:
+ db.execute(
+ "UPDATE revisions SET state='FAILED',error=? WHERE id=?",
+ (type(exc).__name__, revision),
+ )
+ raise
+ finally:
+ if embedding and callable(getattr(embedding, "aclose", None)):
+ await embedding.aclose()
+
+
+async def reader_system(workspace, manifest, selected):
+ from datamind.capabilities.graph import GraphService, build_graph_tools
+ from datamind.capabilities.db import build_db_tools
+ from datamind.capabilities.kb import build_kb_tools
+ from datamind.capabilities.kb.service import KBService
+ from datamind.capabilities.kb.providers.lancedb_store import LanceDBVectorStore
+ from datamind.core.registry import retriever_registry
+ from datamind.core.contracts import ToolAccess
+ from datamind.workspace_native import PinnedVectors, PinnedDB
+ from datamind.workspace_graph import GraphView
+ from datamind import runtime
+
+ settings = workspace.settings.model_copy(deep=True)
+ settings.hooks.path_allowlist_extra = [
+ *settings.hooks.path_allowlist_extra,
+ str(workspace.root),
+ manifest["source_root"],
+ ]
+ system = await workspace._system(settings, set())
+ try:
+ if "kb" in selected:
+ binding = manifest["snapshots"]["kb"]
+ # A historical index must use the model that built its embeddings.
+ current = settings.embedding.model_dump(mode="json", exclude={"api_key"})
+ if digest(encoded(current)) != binding["embedding_fingerprint"]:
+ raise ConfigError(
+ "Pinned KB embedding configuration changed; use the original configuration"
+ )
+ embedding = runtime.build_embedding(
+ settings.embedding, fallback_llm=settings.llm
+ )
+ raw = LanceDBVectorStore(
+ persist_dir=str(workspace.root / binding["path"]),
+ collection_name="workspace",
+ dimension=binding["dimension"],
+ )
+ store = PinnedVectors(raw, binding["version"])
+ strategy = settings.retrieval.strategy
+ if strategy not in {"simple", "hybrid"}:
+ raise ConfigError(
+ "Incremental reads support simple and hybrid retrieval"
+ )
+ options = {"vector_store": store, "embedding": embedding}
+ if strategy == "hybrid":
+ options.update(
+ vector_failure_fallback=settings.retrieval.vector_failure_fallback,
+ vector_retries=settings.retrieval.vector_retries,
+ vector_timeout_s=settings.retrieval.vector_timeout_s,
+ )
+ retriever = retriever_registry.create(strategy, **options)
+ kb = KBService(
+ embedding=embedding,
+ vector_store=store,
+ retriever=retriever,
+ data_dir=settings.data.data_dir,
+ retrieval_cfg=settings.retrieval,
+ )
+
+ async def documents():
+ return [
+ {
+ "path": s["uri"],
+ "source_version": s["version"],
+ "size": (workspace.root / s["snapshot"]).stat().st_size,
+ }
+ for s in manifest["sources"]
+ ]
+
+ kb.list_documents = documents
+ system.services.kb = kb
+ system.services.embedding = embedding
+ system.tools.extend(
+ [s for s in build_kb_tools(kb) if s.access != ToolAccess.WRITE]
+ )
+ if "graph" in selected:
+ graph = GraphService(
+ store=GraphView(
+ workspace.root,
+ {
+ s["graph_artifact"]
+ for s in manifest["sources"]
+ if s.get("graph_artifact")
+ },
+ ),
+ data_dir=settings.data.data_dir,
+ storage_dir=workspace.root,
+ )
+ system.services.graph = graph
+ system.tools.extend(
+ [s for s in build_graph_tools(graph) if s.access != ToolAccess.WRITE]
+ )
+ if "db" in selected:
+ db = PinnedDB(workspace, manifest["snapshots"]["db"])
+ system.services.db = db
+ system.tools.extend(
+ [
+ s
+ for s in build_db_tools(db)
+ if s.access != ToolAccess.WRITE
+ and s.name not in {"db_query_nl", "db_query_nl_at_version"}
+ ]
+ )
+ return system, settings
+ except BaseException:
+ await system.aclose()
+ raise
diff --git a/datamind/workspace_migration.py b/datamind/workspace_migration.py
new file mode 100644
index 0000000..6a973d5
--- /dev/null
+++ b/datamind/workspace_migration.py
@@ -0,0 +1,283 @@
+"""Explicit V1 -> V2 migration from frozen evidence, without model calls.
+
+Old revision IDs remain readable using their original format. The migrated head
+is a new revision; retention is never silently changed during migration.
+"""
+
+from __future__ import annotations
+import csv
+import hashlib
+import json
+import os
+import sqlite3
+import struct
+import tempfile
+from pathlib import Path
+from datamind.core.errors import ConfigError, CapabilityError
+from datamind.core.protocols import GraphTriple
+from datamind.capabilities.ingest.formats import DOCUMENT_EXTS, TABLE_EXTS
+from datamind.workspace_incremental import digest, encoded
+
+
+async def migrate(workspace):
+ from filelock import AsyncFileLock
+
+ async with AsyncFileLock(
+ str(workspace.root / "writer.lock"),
+ timeout=workspace.settings.workspace.lock_timeout_s,
+ ):
+ result = await _migrate(workspace)
+ if workspace.settings.workspace.auto_gc and not result.get("unchanged"):
+ try:
+ result["metrics"]["retention_cleanup"] = await workspace.collect(
+ dry_run=False, compact=workspace.retention["mode"] != "all"
+ )
+ except Exception as exc:
+ result["metrics"]["cleanup_error"] = type(exc).__name__
+ return result
+
+
+async def _migrate(workspace):
+ from datamind.workspace import Workspace
+
+ token, revision = workspace.acquire_lease()
+ if revision is None:
+ raise ConfigError("No published Workspace to migrate")
+ try:
+ old = workspace.manifest(revision)
+ with workspace._connect() as db:
+ expected_parent = db.execute(
+ "SELECT revision,epoch FROM head WHERE singleton=1"
+ ).fetchone()
+ if expected_parent[0] != revision:
+ raise CapabilityError("workspace", "publication conflict during migration")
+ if old.get("format_version") == 2:
+ return {"unchanged": True, "workspace_revision": revision}
+ selected = set(old["surfaces"])
+ if "kb" in selected and workspace.settings.kb.vector_store != "lancedb":
+ raise ConfigError("Configure kb.vector_store=lancedb for migration")
+ if "db" in selected and workspace.settings.db.dialect != "dolt":
+ raise ConfigError("Configure db.dialect=dolt for migration")
+ prepared = {
+ s["uri"]: {
+ "chunks": [],
+ "tables": [],
+ "graph_artifact": None,
+ "truncated": s["uri"] in old.get("truncated_sources", []),
+ }
+ for s in old["sources"]
+ }
+ snapshots = {
+ s["uri"]: workspace._settings_for(revision).data.base_dir / s["snapshot"]
+ for s in old["sources"]
+ }
+ # Validate copied source bytes. Never read mutable originals, which may
+ # have changed or been deleted since the old revision was published.
+ for uri, path in snapshots.items():
+ sha = hashlib.sha256()
+ with path.open("rb") as handle:
+ for block in iter(lambda: handle.read(1024 * 1024), b""):
+ sha.update(block)
+ expected = next(s["version"] for s in old["sources"] if s["uri"] == uri)
+ if "sha256:" + sha.hexdigest() != expected:
+ raise CapabilityError("workspace", "legacy evidence checksum mismatch")
+ graph = {uri: [] for uri in prepared}
+ async with workspace.reader(revision) as reader:
+ services = reader._system.services
+ if "kb" in selected:
+ if services.kb._compatibility_error:
+ raise ConfigError(services.kb._compatibility_error)
+ store = services.kb.vector_store
+ if store.name == "chroma":
+ result = store._collection.get(
+ include=["documents", "metadatas", "embeddings"]
+ )
+ values = zip(
+ result["ids"],
+ result["documents"],
+ result["metadatas"],
+ result["embeddings"],
+ )
+ elif store.name == "lancedb":
+ rows = store._collection.search(None).to_list()
+ values = (
+ (
+ r["id"],
+ r["document"],
+ json.loads(r["metadata"]),
+ r["embedding"],
+ )
+ for r in rows
+ )
+ else:
+ raise ConfigError(
+ "Legacy vector provider cannot export stored embeddings"
+ )
+ for key, text, meta, vector in values:
+ uri = meta.get("source")
+ if uri not in prepared:
+ raise ConfigError("Legacy KB source is absent from manifest")
+ vector = list(vector)
+ ref_key = digest(
+ encoded(
+ [
+ workspace.settings.embedding.model_dump(
+ mode="json", exclude={"api_key"}
+ ),
+ text,
+ ]
+ )
+ )
+ ref = "vectors/" + ref_key[:2] + "/" + ref_key + ".f32"
+ path = workspace.root / ref
+ path.parent.mkdir(parents=True, exist_ok=True)
+ if not path.exists():
+ path.write_bytes(struct.pack("<" + "f" * len(vector), *vector))
+ prepared[uri]["chunks"].append(
+ {"id": key, "text": text, "metadata": meta, "vector_ref": ref}
+ )
+ if "graph" in selected:
+ store = services.graph.store
+ for source, target, data in store._g.edges(data=True):
+ edge = store._edge(source, target, data)
+ uri = edge.properties.get("source")
+ if uri not in graph:
+ raise ConfigError("Legacy graph source is absent from manifest")
+ triple = GraphTriple(
+ subject=source,
+ object=target,
+ relation=edge.relation,
+ subject_type=store._g.nodes[source].get("type", "entity"),
+ object_type=store._g.nodes[target].get("type", "entity"),
+ confidence=edge.weight,
+ source=uri,
+ evidence=edge.evidence,
+ properties={
+ k: v for k, v in edge.properties.items() if k != "source"
+ },
+ )
+ graph[uri].append(triple.model_dump(mode="json"))
+ if "db" in selected:
+ dbpath = services.db.engine.url.database
+ connection = sqlite3.connect(
+ "file:" + str(Path(dbpath).resolve()) + "?mode=ro", uri=True
+ )
+ try:
+ for (name,) in connection.execute(
+ "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
+ ).fetchall():
+ quoted = '"' + name.replace('"', '""') + '"'
+ cursor = connection.execute("SELECT * FROM " + quoted)
+ columns = [c[0] for c in cursor.description]
+ if "_source_file" not in columns:
+ raise ConfigError("Legacy table has no source identity")
+ first = cursor.fetchone()
+ if first is None:
+ candidates = []
+ for source in old["sources"]:
+ checkpoint = source.get("checkpoint")
+ if (
+ checkpoint
+ and (workspace.root / checkpoint).is_file()
+ ):
+ value = json.loads(
+ (workspace.root / checkpoint).read_text()
+ )
+ if any(
+ "ws_"
+ + digest((source["source_id"] + t[0]).encode())[
+ :24
+ ]
+ == name
+ for t in value.get("tables", [])
+ ):
+ candidates.append(source["uri"])
+ if len(candidates) != 1:
+ raise ConfigError(
+ "Cannot identify source of empty legacy table; republish it before migration"
+ )
+ uri = candidates[0]
+ else:
+ uri = first[columns.index("_source_file")]
+ if uri not in prepared:
+ raise ConfigError(
+ "Legacy DB source is absent from manifest"
+ )
+ fd, temp = tempfile.mkstemp(
+ dir=workspace.root, prefix=".migrate-table-"
+ )
+ try:
+ count = 0
+ with os.fdopen(
+ fd, "w", encoding="utf-8", newline=""
+ ) as handle:
+ writer = csv.writer(handle)
+ writer.writerow(columns)
+ rows = [first] if first is not None else []
+ while rows:
+ for row in rows:
+ if row[columns.index("_source_file")] != uri:
+ raise ConfigError(
+ "Legacy table mixes source identities"
+ )
+ writer.writerow(
+ [
+ json.dumps(v, ensure_ascii=False)
+ for v in row
+ ]
+ )
+ count += 1
+ rows = cursor.fetchmany(1000)
+ sha = hashlib.sha256()
+ with open(temp, "rb") as handle:
+ for block in iter(
+ lambda: handle.read(1024 * 1024), b""
+ ):
+ sha.update(block)
+ key = sha.hexdigest()
+ dest = workspace.root / "tables" / key[:2] / (key + ".csv")
+ dest.parent.mkdir(parents=True, exist_ok=True)
+ if not dest.exists():
+ os.replace(temp, dest)
+ prepared[uri]["tables"].append(
+ {
+ "name": name,
+ "hash": key,
+ "path": dest.relative_to(workspace.root).as_posix(),
+ "columns": columns,
+ "rows": count,
+ "sheet": name,
+ }
+ )
+ finally:
+ Path(temp).unlink(missing_ok=True)
+ finally:
+ connection.close()
+ worker = Workspace(workspace.settings, builder=workspace._builder)
+ worker.settings.workspace.storage_mode = "incremental"
+ worker.settings.workspace.auto_gc = False
+ worker._writer_held = True
+ worker._expected_parent = expected_parent
+ inventory = [
+ {k: s[k] for k in ("uri", "relative_path", "version")}
+ for s in old["sources"]
+ ]
+ worker._inventory = lambda path: list(inventory)
+ worker._migration_sources = snapshots
+ worker._migration_prepared = prepared
+ assertions = [
+ {
+ "uri": s["uri"],
+ "version": s["version"],
+ "triples": graph[s["uri"]],
+ "truncated": prepared[s["uri"]]["truncated"],
+ }
+ for s in old["sources"]
+ if Path(s["uri"]).suffix.lower() in (DOCUMENT_EXTS - TABLE_EXTS)
+ ]
+ result = await worker.update(
+ old["source_root"], surfaces=selected, graph_sources=assertions
+ )
+ return {**result, "migrated_from": revision}
+ finally:
+ workspace.release_lease(token)
diff --git a/datamind/workspace_native.py b/datamind/workspace_native.py
new file mode 100644
index 0000000..2ad3722
--- /dev/null
+++ b/datamind/workspace_native.py
@@ -0,0 +1,488 @@
+"""Native LanceDB versions and Dolt commits for immutable Workspace bindings.
+
+The shared stores are private to Workspace; direct live imports never write here.
+Dolt snapshot commits have a common empty parent. Table trees are copied by
+reference from the selected base commit; this makes pruned histories collectible
+without keeping an ancestor chain to every old Workspace revision.
+"""
+
+from __future__ import annotations
+import asyncio
+import csv
+import json
+import os
+import signal
+import time
+import socket
+import subprocess
+from contextlib import asynccontextmanager
+from datetime import timedelta
+from filelock import AsyncFileLock
+from sqlalchemy import create_engine, text
+from sqlalchemy.engine import make_url
+from datamind.core.errors import CapabilityError
+from datamind.capabilities.db.providers.dolt import DoltDialect
+
+
+async def completed(awaitable):
+ # Cancellation must not release a store lock while its worker still writes.
+ task = asyncio.ensure_future(awaitable)
+ try:
+ return await asyncio.shield(task)
+ except asyncio.CancelledError:
+ try:
+ await task
+ finally:
+ raise
+
+
+async def blocking(function, *args, **kwargs):
+ return await completed(asyncio.to_thread(function, *args, **kwargs))
+
+
+class PinnedVectors:
+ name = "lancedb"
+
+ def __init__(self, store, version):
+ self.store, self.version = store, version
+ self.dimension = store.dimension
+ self.existing_count = 0
+
+ @property
+ def latest_version(self):
+ return self.version
+
+ async def query(self, embedding, *, top_k=5, where=None, version=None):
+ if version is not None and version != self.version:
+ raise CapabilityError(
+ "kb", "select a Workspace revision instead of another backend version"
+ )
+ return await completed(
+ self.store.query(embedding, top_k=top_k, where=where, version=self.version)
+ )
+
+ async def count(self):
+ def count():
+ table = self.store._client.open_table(self.store._collection_name)
+ table.checkout(self.version)
+ return table.count_rows()
+
+ return await blocking(count)
+
+ async def get_all_texts(self):
+ return await completed(self.store.get_all_texts(version=self.version))
+
+
+class DoltRepository:
+ def __init__(self, workspace):
+ self.workspace = workspace
+ self.path = workspace.root / "shared" / "dolt"
+ self.path.mkdir(parents=True, exist_ok=True)
+
+ def _reap_interrupted_server(self, config):
+ # A killed parent cannot run Popen cleanup. The absolute, private config
+ # path is an ownership marker; never signal unrelated Dolt processes.
+ result = subprocess.run(
+ ["ps", "-ww", "-axo", "pid=,command="],
+ capture_output=True,
+ text=True,
+ check=True,
+ timeout=10,
+ )
+ for line in result.stdout.splitlines():
+ parts = line.strip().split(None, 1)
+ if len(parts) != 2:
+ continue
+ command = parts[1].strip()
+ marker = " sql-server --config " + str(config)
+ if (
+ not command.endswith(marker)
+ or os.path.basename(command[: -len(marker)]) != "dolt"
+ ):
+ continue
+ pid = int(parts[0])
+ try:
+ os.kill(pid, signal.SIGTERM)
+ except ProcessLookupError:
+ continue
+ deadline = time.monotonic() + 5
+ while time.monotonic() < deadline:
+ state = subprocess.run(
+ ["ps", "-p", str(pid), "-o", "stat="],
+ capture_output=True,
+ text=True,
+ timeout=5,
+ ).stdout.strip()
+ if not state or state.startswith("Z"):
+ break
+ time.sleep(0.05)
+ else:
+ # Confirm the ownership marker again before escalation.
+ current = subprocess.run(
+ ["ps", "-ww", "-p", str(pid), "-o", "command="],
+ capture_output=True,
+ text=True,
+ timeout=5,
+ ).stdout.strip()
+ if current == command:
+ try:
+ os.kill(pid, signal.SIGKILL)
+ except ProcessLookupError:
+ pass
+
+ @asynccontextmanager
+ async def session(self, revision=None):
+ async with AsyncFileLock(
+ str(self.workspace.root / "dolt.lock"),
+ timeout=self.workspace.settings.workspace.lock_timeout_s,
+ ):
+ dialect = DoltDialect()
+ engine = None
+ view_engine = None
+ try:
+ config = self.workspace.root / "shared" / "dolt-server.json"
+ await blocking(self._reap_interrupted_server, config)
+ # This local coordinator owns its repository. It never applies
+ # checkout, reset or GC to an application's external database.
+ with socket.socket() as sock:
+ sock.bind(("127.0.0.1", 0))
+ port = sock.getsockname()[1]
+ dsn = f"mysql+pymysql://root:@127.0.0.1:{port}/dolt"
+ from datamind.workspace import _atomic_json
+
+ _atomic_json(config, {"listener": {"host": "127.0.0.1", "port": port}})
+ engine = await blocking(
+ dialect.build_engine,
+ dsn,
+ storage_dir=str(self.path),
+ server_config=config,
+ )
+ if revision:
+ revision = dialect.validate_version(revision)
+ url = make_url(dsn).set(database="dolt/" + revision)
+ view_engine = create_engine(url, pool_pre_ping=True)
+ yield dialect, view_engine or engine
+ finally:
+ if view_engine is not None:
+ await blocking(view_engine.dispose)
+ if engine is not None:
+ await blocking(engine.dispose)
+ await blocking(dialect.aclose)
+
+ async def build(self, base, tables, revision):
+ async with self.session() as (dialect, root_engine):
+
+ def sql(query, params=None):
+ return dialect._run_sql(root_engine, query, params)
+
+ initial = (await blocking(sql, "SELECT DOLT_HASHOF('main')"))[0][0]
+ branch = "dm_build_" + revision.removeprefix("revision-")
+ await blocking(
+ sql,
+ "CALL DOLT_BRANCH(:branch,:base)",
+ {"branch": branch, "base": initial},
+ )
+ url = root_engine.url.set(database="dolt/" + branch)
+ engine = create_engine(url, pool_pre_ping=True)
+ old = base.get("tables", {}) if base else {}
+ try:
+ if base and old:
+ placeholders = ",".join(":t" + str(i) for i in range(len(old)))
+ params = {
+ "commit": base["version"],
+ **{"t" + str(i): name for i, name in enumerate(old)},
+ }
+ await blocking(
+ dialect._run_sql,
+ engine,
+ "CALL DOLT_CHECKOUT(:commit," + placeholders + ")",
+ params,
+ )
+
+ def apply():
+ from itertools import islice
+
+ with engine.begin() as conn:
+ for name in old.keys() - tables.keys():
+ conn.execute(
+ text("DROP TABLE " + dialect._quote_ident(name))
+ )
+ for name, item in tables.items():
+ if old.get(name) == item["hash"]:
+ continue
+ q = dialect._quote_ident
+ if name in old:
+ conn.execute(text("DROP TABLE " + q(name)))
+ cols = item["columns"]
+ conn.execute(
+ text(
+ "CREATE TABLE "
+ + q(name)
+ + " ("
+ + ", ".join(q(c) + " TEXT" for c in cols)
+ + ")"
+ )
+ )
+ statement = text(
+ "INSERT INTO "
+ + q(name)
+ + " ("
+ + ",".join(q(c) for c in cols)
+ + ") VALUES ("
+ + ",".join(":" + c for c in cols)
+ + ")"
+ )
+ with (self.workspace.root / item["path"]).open(
+ newline="", encoding="utf-8"
+ ) as handle:
+ reader = csv.reader(handle)
+ next(reader, None)
+ while batch := list(islice(reader, 1000)):
+ # Canonical JSON cells preserve NULL vs empty
+ # strings without interpreting user values.
+ conn.execute(
+ statement,
+ [
+ {
+ c: json.loads(v)
+ for c, v in zip(cols, row)
+ }
+ for row in batch
+ ],
+ )
+
+ await blocking(apply)
+ commit = await completed(
+ dialect.commit_version(engine, "Workspace " + revision)
+ )
+ if commit is None:
+ commit = (
+ await blocking(
+ dialect._run_sql, engine, "SELECT DOLT_HASHOF('HEAD')"
+ )
+ )[0][0]
+ tag = "dm_" + commit
+ existing = await blocking(
+ dialect._run_sql,
+ engine,
+ "SELECT tag_name FROM dolt_tags WHERE tag_name=:tag",
+ {"tag": tag},
+ )
+ if not existing:
+ await blocking(
+ dialect._run_sql,
+ engine,
+ "CALL DOLT_TAG(:tag,:commit)",
+ {"tag": tag, "commit": commit},
+ )
+ return {
+ "provider": "dolt",
+ "version": commit,
+ "tables": {n: t["hash"] for n, t in tables.items()},
+ "path": "shared/dolt",
+ }
+ finally:
+ await blocking(engine.dispose)
+ # A retained tag, not this staging branch, owns the snapshot.
+ await blocking(
+ sql, "CALL DOLT_BRANCH('-D',:branch)", {"branch": branch}
+ )
+
+
+class PinnedDB:
+ def __init__(self, workspace, binding):
+ self.repository = DoltRepository(workspace)
+ self.binding = binding
+ self.db_cfg = workspace.settings.db
+ self.dialect = DoltDialect()
+
+ def _validate(self, sql):
+ from sqlglot import parse, exp
+ from sqlglot.optimizer.scope import traverse_scope
+ from sqlglot.errors import ParseError
+
+ try:
+ trees = parse(sql, read="mysql")
+ except ParseError as exc:
+ raise CapabilityError("db", "SQL is not a supported pinned SELECT") from exc
+ if len(trees) != 1 or not isinstance(trees[0], exp.Query):
+ raise CapabilityError("db", "Workspace reads require one SELECT")
+ if trees[0].find(exp.Into) or trees[0].find(exp.Lock):
+ raise CapabilityError(
+ "db", "Workspace reads cannot write files or acquire locks"
+ )
+ for function in trees[0].find_all(exp.Anonymous):
+ if function.name.lower().startswith("dolt_") or function.name.lower() in {
+ "load_file",
+ "get_lock",
+ "release_lock",
+ }:
+ raise CapabilityError(
+ "db", "function is outside the pinned Workspace revision"
+ )
+ for scope in traverse_scope(trees[0]):
+ for source in scope.sources.values():
+ if isinstance(source, exp.Table):
+ if (
+ source.db
+ or source.catalog
+ or source.args.get("version")
+ or source.name not in self.binding["tables"]
+ ):
+ raise CapabilityError(
+ "db", "table is outside the pinned Workspace revision"
+ )
+
+ async def list_tables(self):
+ return sorted(self.binding["tables"])
+
+ async def describe(self, table):
+ if table not in self.binding["tables"]:
+ raise CapabilityError(
+ "db", "table is outside the pinned Workspace revision"
+ )
+ async with self.repository.session(self.binding["version"]) as (
+ dialect,
+ engine,
+ ):
+ return await completed(dialect.describe(engine, table))
+
+ async def query_sql(self, sql):
+ self._validate(sql)
+ async with self.repository.session(self.binding["version"]) as (
+ dialect,
+ engine,
+ ):
+ return await dialect.execute_readonly(
+ engine,
+ sql,
+ row_limit=self.db_cfg.row_limit,
+ timeout_s=self.db_cfg.query_timeout_s,
+ )
+
+ def _check_version(self, version):
+ if version != self.binding["version"]:
+ raise CapabilityError(
+ "db", "select a Workspace revision instead of another backend version"
+ )
+
+ async def list_versions(self):
+ return {"versions": [{"commit": self.binding["version"]}], "pinned": True}
+
+ async def list_tables_at_version(self, *, version):
+ self._check_version(version)
+ return await self.list_tables()
+
+ async def describe_at_version(self, table, *, version):
+ self._check_version(version)
+ return await self.describe(table)
+
+ async def query_sql_at_version(self, sql, *, version):
+ self._check_version(version)
+ return await self.query_sql(sql)
+
+ async def aclose(self):
+ pass
+
+
+async def prune_native(workspace, manifests, *, compact=False):
+ groups = {}
+ for manifest in manifests:
+ kb = manifest.get("snapshots", {}).get("kb", {})
+ if manifest.get("format_version") == 2 and kb:
+ groups.setdefault(kb["path"], set()).add(kb["version"])
+ report = {"kb": [], "db": None, "compacted": compact}
+ import shutil
+
+ kbroot = workspace.root / "shared" / "kb"
+ for path in kbroot.glob("*"):
+ relative = path.relative_to(workspace.root).as_posix()
+ if relative not in groups:
+ shutil.rmtree(path)
+ report["kb"].append({"path": relative, "removed": True})
+ continue
+ import lancedb
+
+ table = lancedb.connect(path).open_table("workspace")
+ keep = groups[relative]
+
+ def clean():
+ for version in keep:
+ tag = "dm_workspace_" + str(version)
+ if tag not in table.tags.list():
+ table.tags.create(tag, version)
+ for tag in table.tags.list():
+ if (
+ tag.startswith("dm_workspace_")
+ and int(tag.removeprefix("dm_workspace_")) not in keep
+ ):
+ table.tags.delete(tag)
+ # Tagged retained versions are protected by the storage engine.
+ return (
+ table.to_lance().cleanup_old_versions(
+ older_than=timedelta(0),
+ delete_unverified=False,
+ error_if_tagged_old_versions=False,
+ )
+ if compact
+ else {"physical_gc_pending": True}
+ )
+
+ stats = await blocking(clean)
+ report["kb"].append(
+ {"path": relative, "retained_versions": sorted(keep), "cleanup": str(stats)}
+ )
+ path = workspace.root / "shared" / "dolt"
+ if (path / ".dolt").is_dir():
+ versions = {
+ m["snapshots"]["db"]["version"]
+ for m in manifests
+ if m.get("format_version") == 2 and m.get("snapshots", {}).get("db")
+ }
+ repo = DoltRepository(workspace)
+ async with repo.session() as (dialect, engine):
+ rows = await blocking(
+ dialect._run_sql, engine, "SELECT tag_name FROM dolt_tags"
+ )
+ for (tag,) in rows:
+ if tag.startswith("dm_") and tag.removeprefix("dm_") not in versions:
+ await blocking(
+ dialect._run_sql,
+ engine,
+ "CALL DOLT_TAG('-d',:tag)",
+ {"tag": tag},
+ )
+ rows = await blocking(
+ dialect._run_sql, engine, "SELECT name FROM dolt_branches"
+ )
+ for (name,) in rows:
+ if name.startswith("dm_build_"):
+ await blocking(
+ dialect._run_sql,
+ engine,
+ "CALL DOLT_BRANCH('-D',:name)",
+ {"name": name},
+ )
+ # Ordinary GC now works: retained snapshots have no parent chain to
+ # pruned Workspace commits, only to the common empty initial commit.
+ if compact:
+ async with AsyncFileLock(
+ str(workspace.root / "dolt.lock"),
+ timeout=workspace.settings.workspace.lock_timeout_s,
+ ):
+
+ def gc():
+ subprocess.run(
+ ["dolt", "gc"],
+ cwd=path,
+ check=True,
+ capture_output=True,
+ timeout=300,
+ )
+
+ await blocking(gc)
+ report["db"] = {
+ "retained_versions": sorted(versions),
+ "physical_gc_pending": not compact,
+ }
+ return report
diff --git a/docs/KNOWLEDGE_LIFECYCLE.md b/docs/KNOWLEDGE_LIFECYCLE.md
index 4430ccb..b4139d5 100644
--- a/docs/KNOWLEDGE_LIFECYCLE.md
+++ b/docs/KNOWLEDGE_LIFECYCLE.md
@@ -52,110 +52,23 @@ an in-memory NetworkX store a concurrent multi-process enterprise backend.
## Published workspace API
-```python
-from datamind.config import Settings
-from datamind.workspace import Workspace
-
-workspace = Workspace(Settings())
-published = await workspace.update(
- "/absolute/path/to/source", surfaces=["kb", "graph", "db"],
- relation_types=["owns", "depends_on", "requires", "cites"],
-)
-async with workspace.reader(published["workspace_revision"]) as reader:
- answer = await reader.query("Which projects depend on the changed procedure?")
- # Direct read tools also remain pinned, even if another process publishes.
- edges = await reader.call("graph_neighbors", entity="Project A")
-```
-
-Codex prepares graph assertions externally and calls `workspace_publish` through direct data tools. The old internally generative `workspace_update` is not exposed to the main agent.
-The Codex adapter exposes `datamind_workspace_update` and
-`datamind_workspace_ask`. Legacy `datamind_use_folder` and normal live queries
-retain their unversioned semantics; a workspace update publishes a separate
-snapshot and never silently replaces those live stores.
-
-Every revision contains copied source bytes, private capability storage, a
-source inventory with stable URI-based IDs/content versions, snapshot bindings,
-and a processing fingerprint. The SQLite catalog publishes by comparing the
-current head with the build's starting head. Concurrent stale builds fail
-publication rather than overwriting newer state. A reader owns one revision for
-its lifetime; original source URIs in raw-evidence reads resolve to copied bytes.
-Workspace inspection is also restricted to the pinned source copy.
-
-To return to a retained complete generation:
-
-```python
-workspace.rollback(old_revision, expected_revision=workspace.current_revision)
-```
-
-Completed-source checkpoints include passages, vectors, extracted relationships,
-and parsed tables. A new build after process termination reuses these checkpoints;
-it does not rerun model calls for already completed sources. A failed source is
-retried in full. Index materialization currently writes a new full snapshot: this
-is file-granularity compute reuse, not dependency-aware sub-file maintenance or
-constant-cost storage. Old snapshots/checkpoints are retained; automatic GC and
-exactly-once external writes are not provided.
-
-The shipped coordinator supports one-host concurrent readers/builders using local
-NetworkX, SQLite, and Chroma/LanceDB storage. It rejects external DB DSNs and
-unsupported graph providers before building. Multi-host scheduling, tenant
-identity/authorization, distributed locks, remote snapshot providers, and service
-capacity/load validation remain deployment work. They do not follow from selecting
-the Agent SDK backend. Snapshot readers should be reused per request/session as
-appropriate; this API currently opens capability services when a reader is opened.
-
-## PR #28 integration
-
-LanceDB versioning is opt-in with `pip install 'datamind[versioning]'` and
-`DATAMIND__KB__VECTOR_STORE=lancedb`. CocoIndex additionally requires
-`pip install 'datamind[cocoindex]'` and `DATAMIND__KB__INGEST_MODE=cocoindex`.
-Personal installations keep Chroma/direct ingestion by default. Dolt requires
-`pip install 'datamind[dolt]'` and a separately installed Dolt binary/service.
-
-History records survive process restart; historical LanceDB reads use private
-table handles and do not change the writer's checkout. Re-adding a chunk ID is an
-upsert. External source copies are keyed by origin so editing one source replaces
-its own managed copy rather than accumulating content-hash filenames. Dolt
-historical SQL rewrites physical tables using an AST, covering joins and CTEs
-without changing string literals. Dolt and external MySQL integration tests need
-their services and must not be counted as validated when skipped.
-
-## Ingestion Skills
-
-`workspace-ingest` and `source-update` ship in both the checkout and wheel skill
-catalogs. The same main Codex agent can inspect and read data while preparing an
-explicitly authorized write. Read-only operations reject writes in code. Skills guide
-routing, schemas and receipt inspection; persistence invariants stay in tools.
-A successfully fetched Skill contributes its name, declared version and content
-hash to subsequent write receipts in that store turn. Fetching records procedural
-context, not proof the model followed every instruction. Receipt context does not
-leak into another store turn. Partial directory failures produce failed receipts
-and are not cached as successful imports.
-
-Pinned readers retry only transient I/O/timeouts and retryable external-service
-errors, with a bounded `read_retries` budget (default 1, configurable from 0 to 5).
-Retries use the same revision. After exhaustion the original error is returned to
-the Agent, which can use another available evidence path; no answer is fabricated
-and no latest-version fallback is made. The query response reports the retry
-count. Application-level deadlines and workload evaluations remain the caller's
-responsibility.
-
-Inline graph text and structured assertions without supplied evidence retain
-immutable input records under `triplets/evidence/`. These are excluded from KB
-raw-document indexing. Explicit assertions are labeled as caller statements, not
-independently verified facts. Logical file IDs now derive from canonical URI;
-content hashes remain source versions. The receipt ledger migrates legacy JSON
-state to SQLite, allocates unique revisions across workers, retains durable
-receipts, and keeps the JSONL export for existing tooling. This does not make an
-external append write and its receipt an exactly-once distributed transaction.
-
-## Validation status
-
-Local validation: 368 package tests passed, 86 skipped; 3 Codex-adapter tests
-passed. The SQLite demo and wheel/resource checks passed. Tests use local storage
-and deterministic model/embedding doubles. Coverage includes actual hard process
-exit, source checkpoint reuse, parallel publication, publication/rollback races,
-transient pinned reads, historical table reads, graph evidence replacement, and
-Skill receipt isolation. Dolt/MySQL service tests were skipped when unavailable;
-live SDK/model integrations, enterprise throughput, and research quality/cost
-benchmarks have not been measured. Publication includes a monotonically increasing
-epoch so an intervening publication and rollback cannot let a stale builder win.
+Codex inventories sources using `datamind_workspace_prepare(path)`, reads new or
+changed documents, then submits `datamind_workspace_publish(path, surfaces,
+graph_sources=[{uri, version, triples}, ...])` using a write operation. Each input
+version must match the inventory SHA-256. Include empty triples when no relations
+were found. Snapshot mode requires all documents; incremental mode reuses unchanged
+source assertions. Source deletions disappear from the new inventory automatically.
+
+Python callers use `Workspace.update(..., graph_sources=...)` and
+`async with workspace.reader(revision) as reader`, then `reader.call(...)` for direct
+pinned reads. Query planning stays with the main Codex agent. Legacy internally
+generative `workspace_update`/`workspace_ask` are not MCP tools.
+
+The catalog publishes one manifest with fixed KB, DB and graph bindings. A revision
+and epoch compare-and-swap rejects stale builds, including rollback ABA. All reads
+within an operation stay on that manifest; original source URIs resolve to immutable
+stored bytes. A failed build never changes the published head.
+
+See [Workspace storage](WORKSPACE_STORAGE.md) for source reuse, native version
+bindings, history retention, reader leases, cleanup and explicit legacy migration.
+Direct live imports remain independent; they do not update published revisions.
diff --git a/docs/STABLE_API.md b/docs/STABLE_API.md
index d1aba80..2eedc4d 100644
--- a/docs/STABLE_API.md
+++ b/docs/STABLE_API.md
@@ -24,7 +24,9 @@ async def run():
`DataSettings` and `Settings` name the same typed settings model.
`DataPlane.call(name, arguments, context=None, read_only=False)` validates the
schema, profile and hooks. Business write tools attach receipts. Selecting
-surfaces controls which services initialize. No conversation client is created;
+surfaces controls which services initialize. `build_data_plane(..., enable=set(),
+workspace_enable={"kb", "db", "graph"})` registers lazy Workspace publication
+without opening independent live stores. No conversation client is created;
KB still requires a configured embedding provider. An explicit `text_client`
parameter is reserved for custom capability pipelines and benchmark fixtures.
@@ -63,14 +65,21 @@ documented in [deployment](CODEX_GATEWAY_DEPLOYMENT.md).
## Workspace snapshots
-`Workspace.prepare` inventories sources; `Workspace.publish` builds and atomically
-publishes selected KB/DB/Graph snapshots from prepared, source-hashed inputs.
+`workspace_prepare` and `workspace_publish` are direct/MCP tools. The Python
+API is `Workspace.update(path, surfaces=..., graph_sources=...)`; it validates
+Codex-prepared assertions and atomically publishes the selected KB/DB/Graph revision.
`Workspace.reader(revision=...)` selects immutable historical snapshots;
`enable` may select a subset of that manifest's surfaces. `reader.call` enforces
read-only access. Query hosting selects the current head at request admission;
without a published head, it reads independent live stores. Direct writes do
not silently modify snapshots. Memory/Skills are not one atomic business snapshot.
+Retention and shared incremental storage are documented in [Workspace storage](WORKSPACE_STORAGE.md).
+`Workspace.history()`, `configure_retention(mode, keep_last)`, `protect(revision)`,
+`collect(dry_run=True, compact=False)`, `migrate()` and `rollback(revision, expected_revision=...)`
+are available to Python callers. `migrate` and `collect` are async. MCP readers must
+close operations with `datamind_operation_close`; calls renew durable snapshot leases.
+
## HTTP API
```json
@@ -80,6 +89,11 @@ not silently modify snapshots. Memory/Skills are not one atomic business snapsho
| Route | Contract |
|---|---|
| `GET /api/health` | Codex backend/model, profile, package version, tool count, receipt revision and published workspace revision |
+| `GET /api/workspace` | History and retention; `?usage=true` includes allocated file sizes |
+| `POST /api/workspace/retention` | Set `{mode: latest/recent/all, keep_last: N}`; no immediate deletion |
+| `POST /api/workspace/gc` | `{dry_run: true, compact: false}` defaults to a preview |
+| `POST /api/workspace/rollback` | Select retained `{revision, expected_revision}` |
+| `POST /api/workspace/migrate` | Explicit migration from frozen legacy evidence |
| `GET /api/tools` | Tools with access, surface, description and schema; role fields retain read/write labels for UI compatibility |
| `POST /api/ask` | Same Codex host with read-only access; optional request field `revision` |
| `POST /api/store` | Same Codex host with write permission and checked receipt results |
diff --git a/docs/WORKSPACE_STORAGE.md b/docs/WORKSPACE_STORAGE.md
new file mode 100644
index 0000000..1fcca2b
--- /dev/null
+++ b/docs/WORKSPACE_STORAGE.md
@@ -0,0 +1,166 @@
+# Workspace storage and history (2.1)
+
+A Workspace revision is a small catalog entry and manifest binding all selected
+surfaces to one immutable state. Keeping history does not require another full
+copy of every store. History retention and storage layout are separate choices.
+
+## Enable shared incremental storage
+
+```sh
+pip install -e ".[versioning,dolt]"
+# Install the Dolt 2.5.0 executable separately and put it on PATH.
+export DATAMIND__WORKSPACE__STORAGE_MODE=incremental
+export DATAMIND__KB__VECTOR_STORE=lancedb
+export DATAMIND__DB__DIALECT=dolt
+```
+
+Supported/tested native packages: LanceDB 0.40.x, pylance 9.x, PyArrow >=19,
+SQLAlchemy and PyMySQL (the `dolt` extra). Catalog/graph use Python SQLite;
+cross-process locks use filelock. KB also needs an embedding provider.
+Graph-only incremental workspaces do not require LanceDB or Dolt. Selected KB/DB
+surfaces require those native providers. An external DB DSN is rejected: Workspace
+owns a private local Dolt repository and never resets an application's database.
+
+`storage_mode=auto` preserves the layout of an existing head. For a new Workspace
+it chooses incremental when KB=LanceDB and DB=Dolt are configured, otherwise the
+legacy snapshot layout. `snapshot` retains the legacy SQLite/NetworkX and local
+Chroma/LanceDB implementation. Providers are never switched silently.
+
+## What is stored
+
+| Component | Shared data | What a revision references |
+|---|---|---|
+| Raw evidence | Immutable files keyed by SHA-256 and format | Object path, original URI and source hash |
+| KB | One private LanceDB table per embedding configuration; immutable vector cache | Fixed table version and chunk fingerprints |
+| DB | One private Dolt repository; unchanged table trees copied by reference | Fixed commit and table hashes |
+| Graph | Indexed SQLite assertions grouped by immutable source artifact | Source artifact IDs visible to this revision |
+| Build output | Immutable source artifacts with parsed tables, chunks and provenance | Processing fingerprint and source artifact path |
+
+Unchanged sources do not need copying, embedding, parsing or graph extraction.
+A changed document reuses identical text embeddings and updates its KB rows and
+source graph assertions. A removed source disappears from the next manifest;
+other origins supporting the same graph edge survive. Codex supplies graph
+assertions for new or changed documents, including explicit empty lists.
+
+CSV/TSV/XLSX imports stream canonical rows into table objects. Unchanged tables
+are reused. Changed tables are currently replaced as a unit; Dolt shares matching
+internal data where possible. This release does **not** promise row-level upserts
+or constant update cost for a giant changed CSV. All imported columns preserve
+TEXT semantics (NULL and empty strings remain distinct). Convert legacy XLS first.
+
+Dolt snapshot commits share an empty initial parent rather than forming a chain
+of Workspace snapshots. Table trees are copied from the selected base commit.
+The Workspace catalog preserves logical parentage. Removing a snapshot tag thus
+allows ordinary Dolt GC to reclaim it without retaining all prior commits.
+
+Manifests still contain a complete source/reference inventory, and builds still
+scan/hash source files to detect changes. Metadata and scanning cost therefore grow
+with workspace size; incremental storage does not mean every operation is O(delta).
+Changing embedding configuration creates a separate index; existing readers require
+the original configuration and credentials. Model/API secrets are never saved in
+manifests. This coordinator supports processes on one host with local storage,
+not distributed locking, a remote graph service, or multi-host transactions.
+Dolt SQL sessions are currently serialized and start a local server per operation;
+enterprise high-throughput storage needs a persistent server adapter separately.
+
+## Retention choices
+
+New workspaces default to `latest`. Existing catalogs with revisions default to
+`all`, preserving prior history. An explicitly saved API/CLI/UI policy takes
+precedence over environment defaults and survives restart. Configure through CLI, the UI's Workspace history
+selector, HTTP, MCP, Python or `DATAMIND__WORKSPACE__RETENTION`.
+
+```sh
+datamind workspace-status --usage
+datamind workspace-policy latest # no historical rollback after cleanup
+datamind workspace-policy recent --keep-last 10
+datamind workspace-policy all
+datamind workspace-protect revision-... # retain a particular published version
+```
+
+Setting a policy alone does not delete history. Automatic cleanup runs after
+successful publication (`AUTO_GC=true`); latest/recent also compact native history.
+Active readers and protected versions are retained in addition to policy roots.
+Latest means minimal retained *history*, not literally zero transient storage:
+build staging, active readers and provider housekeeping still take space.
+
+```sh
+datamind workspace-gc # preview only
+datamind workspace-gc --apply # remove unreferenced objects/version bindings
+datamind workspace-gc --apply --compact # also clean Lance history, run Dolt GC, vacuum graph
+```
+
+GC marks unreachable revisions PRUNED before removing files/provider versions;
+interrupted cleanup is retryable. Pruned revisions cannot be read or rolled back.
+Shared data is removed only when no retained revision references it. Cleanup errors
+do not undo a successful publication and are reported in build metrics; retry GC.
+`reclaimed_bytes` counts removed immutable files and revision directories. Native
+provider cleanup reports separately; allocated file size and filesystem free space
+may differ. Non-compacting GC can leave provider history physically allocated.
+
+## Consistency, leases and rollback
+
+Builds serialize under an OS writer lock, prepare shared immutable artifacts and
+native versions, then publish one manifest using revision+epoch CAS in SQLite.
+Failures leave the old head readable. Completed artifacts/embeddings can be reused
+when a failed build restarts. A dead incremental builder is recognized after the
+writer lock is released; GC can discard its unreachable output. The private Dolt server is identified by its absolute configuration path; after a
+parent crash, the next locked operation stops that orphan before reopening the
+repository. Unrelated Dolt processes are left alone. Legacy builds are
+conservatively deferred for explicit recovery instead of guessed dead.
+
+An MCP read operation atomically acquires the head and a durable reader lease.
+Every business tool call reads that revision's bindings, even if another process
+publishes between calls. Calls renew the lease; `datamind_operation_close` releases
+it. Default TTL is 1800 seconds; an expired operation errors and must be restarted,
+never silently switches versions. Python `Workspace.reader` and the Codex host also
+run heartbeats and release leases on exit/cancellation. Crashed readers expire.
+
+```python
+from datamind.config import DataSettings
+from datamind.workspace import Workspace
+
+workspace = Workspace(DataSettings())
+workspace.configure_retention('recent', keep_last=10)
+# graph_sources are prepared externally by the main Codex agent.
+revision = await workspace.update('/absolute/sources', graph_sources=graph_sources)
+async with workspace.reader(revision['workspace_revision']) as reader:
+ print(await reader.call('graph_schema'))
+workspace.rollback(old_revision, expected_revision=workspace.current_revision)
+```
+
+Rollback changes only the catalog head, increments the epoch and preserves fixed
+bindings. The next build uses the rolled-back head as its base; it does not reuse
+an incompatible provider's mutable latest state. Protect rollback targets if a
+latest/recent cleanup might otherwise remove them.
+
+## Migrate an existing Workspace
+
+1. Back up the storage directory and stop legacy snapshot writers.
+2. Configure the native providers above, keeping the original embedding configuration.
+3. Run `datamind workspace-migrate` (or `Workspace.migrate()` / MCP `workspace_migrate`).
+4. Validate pinned reads, choose a history policy, preview GC, then apply if desired.
+
+Migration reads the current revision's frozen evidence, stored embeddings,
+source-attributed graph and SQLite rows; it needs no original files or model calls.
+It publishes a new V2 revision. Existing IDs keep their V1 contents and remain
+readable until retention allows pruning; migration never silently deletes them.
+Only the current head is migrated, not every historical snapshot. Keeping V1
+history therefore keeps its full snapshots. Tables without traceable source
+identity or incompatible embeddings fail explicitly rather than guessing.
+Independent live data is not migrated into a Workspace by this command.
+
+## API entry points
+
+MCP tools (with `datamind_` prefix): `workspace_history`, `workspace_retention`,
+`workspace_gc`, `workspace_rollback`, `workspace_protect`, `workspace_migrate`.
+Mutations require a write operation. GC defaults to `dry_run=true`.
+HTTP exposes history, retention, GC, rollback and migration under `/api/workspace`;
+see [API contracts](STABLE_API.md). These share the existing trusted-host HTTP
+boundary; enterprise deployments must apply their own tenant/auth policy.
+
+The same data layer serves the personal Codex plugin and the Codex SDK host.
+Tests cover native fixed-version reads, table/object reuse, active/expired leases,
+rollback ABA, failed native publication, process death before publication,
+source-scoped graph retraction, migration and provider cleanup. Passing tests
+supports these contracts; it cannot guarantee all backend scales or model answers.
diff --git a/plugins/datamind-context/.codex-plugin/plugin.json b/plugins/datamind-context/.codex-plugin/plugin.json
index fad14e2..aa758a3 100644
--- a/plugins/datamind-context/.codex-plugin/plugin.json
+++ b/plugins/datamind-context/.codex-plugin/plugin.json
@@ -1,6 +1,6 @@
{
"name": "datamind-context",
- "version": "2.0.0",
+ "version": "2.1.0",
"description": "Shared workspace data tools and Skills for Codex.",
"author": {
"name": "DataMind contributors",
diff --git a/plugins/datamind-context/README.md b/plugins/datamind-context/README.md
index 1e7fe56..9a57886 100644
--- a/plugins/datamind-context/README.md
+++ b/plugins/datamind-context/README.md
@@ -34,3 +34,9 @@ The new default tool catalogue replaces ask/store/graph_query/workspace_ask with
main-agent direct calls. CLI, HTTP and UI now use the same main Codex host; old dual-agent Python APIs have been removed in 2.0.
See [architecture and SDK setup](../../README.md).
+
+## Incremental Workspace storage
+
+Local Codex and the Codex SDK host use the same Workspace storage and retention
+APIs. See [storage, setup and migration](../../docs/WORKSPACE_STORAGE.md).
+Close read operations when finished so latest-mode cleanup can reclaim history.
diff --git a/plugins/datamind-context/skills/datamind-context/SKILL.md b/plugins/datamind-context/skills/datamind-context/SKILL.md
index 714bd52..ea9b09f 100644
--- a/plugins/datamind-context/skills/datamind-context/SKILL.md
+++ b/plugins/datamind-context/skills/datamind-context/SKILL.md
@@ -17,7 +17,8 @@ or builds a new published snapshot. Published revisions cannot be mutated.
Fetch the applicable procedure using `datamind_skill_get` with the same ID:
- `workspace-ingest`: choose surfaces, prepare sources, extract graph triples with
- Codex, then publish `{uri, version, triples}` per document. DataMind validates
+ Codex, then publish `{uri, version, triples}` for new or changed documents in incremental
+ mode (all documents in snapshot mode). DataMind validates
hashes, builds KB/DB, and atomically publishes the selected surfaces.
- `source-update`: refresh a snapshot or replace/retract one live source's support.
- `workspace-retrieve`: search passages, inspect tables and generate SQL, discover
@@ -30,5 +31,12 @@ to see newly published data; existing read operations retain their revision.
Skills and Memory are live context, separate from snapshot business data.
KB needs an embedding provider. Direct Graph, SQL, Skills and Memory do not need
-a DataMind conversation-model key. The snapshot coordinator supports local
-SQLite/NetworkX/Chroma/LanceDB; report unsupported providers without switching them.
+a DataMind conversation-model key. Incremental Workspace storage uses local
+LanceDB, Dolt and a source-scoped SQLite graph. Legacy SQLite/NetworkX/Chroma
+snapshots remain supported. Report unsupported providers without switching them.
+
+Close finished operations with `datamind_operation_close`. Calls renew their
+lease; expired operations fail and must be restarted. `datamind_workspace_history`
+shows retained revisions and policy. New workspaces default to latest; existing
+histories remain all. Change retention only when requested. GC previews by
+default; applying it can permanently remove historical rollback targets.
diff --git a/plugins/datamind-context/tests/test_datamind_mcp.py b/plugins/datamind-context/tests/test_datamind_mcp.py
index 995fba6..51b590c 100644
--- a/plugins/datamind-context/tests/test_datamind_mcp.py
+++ b/plugins/datamind-context/tests/test_datamind_mcp.py
@@ -21,7 +21,7 @@ def test_catalogue_exposes_direct_tools_without_nested_agents():
def test_initialize_teaches_codex_main_agent_contract():
result = handle({'jsonrpc':'2.0','id':1,'method':'initialize'})['result']
assert 'main agent' in result['instructions']
- assert result['serverInfo']['version'] == '2.0.0'
+ assert result['serverInfo']['version'] == '2.1.0'
def test_plugin_and_wheel_entrypoints_run_real_stdio_without_llm(tmp_path):
diff --git a/pyproject.toml b/pyproject.toml
index 2d78f67..ebd4e46 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -62,6 +62,7 @@ dependencies = [
# Async primitives
"anyio>=4.2",
+ "filelock>=3.16",
# Workspace document/table extraction
"pypdf>=5.0",
@@ -85,8 +86,8 @@ legacy = [
# Dialect / provider extras — installed on demand, no hard requirement
codex = ["openai-codex>=0.163,<0.164"]
-versioning = ["lancedb>=0.39", "pyarrow>=19"]
-cocoindex = ["lancedb>=0.39", "pyarrow>=19", "cocoindex>=1.0"]
+versioning = ["lancedb>=0.40,<0.41", "pylance>=9,<10", "pyarrow>=19"]
+cocoindex = ["lancedb>=0.40,<0.41", "pylance>=9,<10", "pyarrow>=19", "cocoindex>=1.0"]
dolt = ["pymysql>=1.1", "mysql-connector-python>=9"]
mysql = ["pymysql>=1.1", "cryptography>=42"]
postgres = ["psycopg[binary]>=3.1"]
diff --git a/static/app.html b/static/app.html
index 8fa29a4..5d1ae6c 100644
--- a/static/app.html
+++ b/static/app.html
@@ -469,7 +469,7 @@
.action-btn:hover { border-color: var(--accent); background: var(--surface3); }
.action-btn.small { padding: 3px 8px; font-size: 10.5px; }
-.memory-box input {
+.memory-box input, .memory-box select {
width: 100%;
background: var(--surface2);
border: 1px solid var(--border);
@@ -570,7 +570,7 @@
- DataMind two-agent
+ DataMind Codex
@@ -625,6 +625,21 @@ 配置 刷新
+