fix: work unit key and reserve system workspace

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

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,
     ),
 }

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:

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

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}")