From 8180c28c162b95a1eb1cd83687adea7f51d92575 Mon Sep 17 00:00:00 2001
From: Rajat Ahuja <rahuja445@gmail.com>
Date: Tue, 13 Jan 2026 16:16:58 -0500
Subject: [PATCH] fix: work unit key and reserve system workspace

---
 config.toml.example            |  2 +-
 src/reconciler/scheduler.py    |  4 ++--
 src/reconciler/sync_vectors.py |  1 +
 src/schemas.py                 |  7 +++++++
 src/utils/work_unit.py         | 19 +++++++++++++++++++
 5 files changed, 30 insertions(+), 3 deletions(-)

diff --git a/config.toml.example b/config.toml.example
index 12d7717b..9332b85f 100644
--- a/config.toml.example
+++ b/config.toml.example
@@ -184,4 +184,4 @@ DIMENSIONS = 1536
 # TURBOPUFFER_API_KEY = "your-turbopuffer-api-key"
 # TURBOPUFFER_REGION = "us-east-1"
 LANCEDB_PATH = "./lancedb_data"
-RECONCILIATION_INTERVAL_SECONDS=
+RECONCILIATION_INTERVAL_SECONDS = 300
diff --git a/src/reconciler/scheduler.py b/src/reconciler/scheduler.py
index 0fdc622d..2b1eee0e 100644
--- a/src/reconciler/scheduler.py
+++ b/src/reconciler/scheduler.py
@@ -45,12 +45,12 @@ QUEUE_CLEANUP_INTERVAL_SECONDS = 12 * 3600  # 12 hours
 RECONCILER_TASKS: dict[str, ReconcilerTask] = {
     "sync_vectors": ReconcilerTask(
         name="sync_vectors",
-        work_unit_key="global:sync_vectors",
+        work_unit_key="reconciler:sync_vectors",
         interval_seconds=settings.VECTOR_STORE.RECONCILIATION_INTERVAL_SECONDS,
     ),
     "cleanup_queue": ReconcilerTask(
         name="cleanup_queue",
-        work_unit_key="global:cleanup_queue",
+        work_unit_key="reconciler:cleanup_queue",
         interval_seconds=QUEUE_CLEANUP_INTERVAL_SECONDS,
     ),
 }
diff --git a/src/reconciler/sync_vectors.py b/src/reconciler/sync_vectors.py
index 032069b9..bcd4d8ce 100644
--- a/src/reconciler/sync_vectors.py
+++ b/src/reconciler/sync_vectors.py
@@ -555,6 +555,7 @@ async def run_vector_reconciliation_cycle() -> ReconciliationMetrics:
             )
             if cleaned:
                 metrics.documents_cleaned += cleaned
+                await db.commit()
                 did_work = True
 
             if not did_work:
diff --git a/src/schemas.py b/src/schemas.py
index c747d3bc..857f8c17 100644
--- a/src/schemas.py
+++ b/src/schemas.py
@@ -206,6 +206,13 @@ class WorkspaceCreate(WorkspaceBase):
 
     model_config = ConfigDict(populate_by_name=True)  # pyright: ignore
 
+    @field_validator("name")
+    @classmethod
+    def validate_not_reserved(cls, v: str) -> str:
+        if v == "__system__":
+            raise ValueError("Workspace name is reserved.")
+        return v
+
 
 class WorkspaceGet(WorkspaceBase):
     filters: dict[str, Any] | None = None
diff --git a/src/utils/work_unit.py b/src/utils/work_unit.py
index da255274..3726cb2f 100644
--- a/src/utils/work_unit.py
+++ b/src/utils/work_unit.py
@@ -64,6 +64,12 @@ def construct_work_unit_key(
             )
         return f"deletion:{workspace_name}:{deletion_type}:{resource_id}"
 
+    if task_type == "reconciler":
+        reconciler_type = payload.get("reconciler_type")
+        if not reconciler_type:
+            raise ValueError("reconciler_type is required for reconciler tasks")
+        return f"reconciler:{reconciler_type}"
+
     raise ValueError(f"Invalid task type: {task_type}")
 
 
@@ -136,4 +142,17 @@ def parse_work_unit_key(work_unit_key: str) -> ParsedWorkUnit:
             observed=None,
         )
 
+    if task_type == "reconciler":
+        if len(parts) != 2:
+            raise ValueError(
+                f"Invalid work_unit_key format for task_type {task_type}: {work_unit_key}"
+            )
+        return ParsedWorkUnit(
+            task_type=task_type,
+            workspace_name="__system__",
+            session_name=None,
+            observer=None,
+            observed=None,
+        )
+
     raise ValueError(f"Invalid task type in work_unit_key: {task_type}")