refactor(tasks): converge task/work store contract

openviking/server/routers/tasks.py

@@ -104,10 +104,6 @@ async def list_tasks(
     if pagination == "cursor":
         from openviking.service.task_pagination import decode_cursor, encode_cursor, page_scope
 
-        # ROOT retains the legacy system + cached visibility and its own
-        # persisted tasks, including submissions restored after restart.
-        account = SYSTEM_TASK_ACCOUNT_ID if _ctx.role == Role.ROOT else _ctx.account_id
-        user = SYSTEM_TASK_USER_ID if _ctx.role == Role.ROOT else _ctx.user.user_id
         filters = {
             "task_type": task_type,
             "status": status,
@@ -118,17 +114,32 @@ async def list_tasks(
         scope = page_scope(
             account=_ctx.account_id, user=_ctx.user.user_id, role=str(_ctx.role), **filters
         )
-        tasks = await tracker.list_page(
-            account_id=account,
-            user_id=user,
-            limit=limit + 1,
-            before=decode_cursor(cursor, scope),
-            include_cached=_ctx.role == Role.ROOT,
-            additional_owner=(_ctx.account_id, _ctx.user.user_id)
+        list_call = (
+            tracker.list_root_tasks(
+                account_id=_ctx.account_id,
+                user_id=_ctx.user.user_id,
+                limit=limit + 1,
+                before=decode_cursor(cursor, scope),
+                task_type=task_type,
+                status=status,
+                resource_id=resource_id,
+                include_internal=include_internal,
+                q=q,
+            )
             if _ctx.role == Role.ROOT
-            else None,
-            **filters,
+            else tracker.list_tasks(
+                account_id=_ctx.account_id,
+                user_id=_ctx.user.user_id,
+                limit=limit + 1,
+                before=decode_cursor(cursor, scope),
+                task_type=task_type,
+                status=status,
+                resource_id=resource_id,
+                include_internal=include_internal,
+                q=q,
+            )
         )
+        tasks = await list_call
         more = len(tasks) > limit
         items = [t.to_dict() for t in tasks[:limit]]
         return Response(

openviking/service/deletion.py

@@ -319,8 +319,13 @@ async def _process(self, message: dict[str, Any]) -> Optional[str]:
             if scheduler is not None:
                 await run_to_completion(lambda: scheduler.delete_tasks(account_id, user_id))
             await run_to_completion(lambda: self._cancel_tasks(account_id, user_id))
-            if user_id is not None:
-                await run_to_completion(lambda: tracker.delete_user_tasks(account_id, user_id))
+            await run_to_completion(
+                lambda: tracker.delete_tasks(
+                    account_id,
+                    user_id,
+                    allow_active_types={"user_delete"} if user_id is None else (),
+                )
+            )
 
             vectors = self._service.viking_fs.vector_store
             if vectors is not None:
@@ -383,7 +388,6 @@ async def _delete_account_files(self, account_id: str) -> None:
             await agfs.stat(path)
         except Exception as exc:
             if is_not_found_error(exc):
-                get_task_tracker().forget_account_tasks(account_id)
                 return
             raise
         raise RuntimeError("Account directory still exists after deletion")

openviking/service/external_task_service.py

@@ -343,7 +343,7 @@ async def execute(self, task_id: str, account_id: str, user_id: str) -> bool:
             await self._execute_task(task, provider, payload, account_id, user_id)
             return True
         except asyncio.CancelledError:
-            if not tracker.is_cancellation_requested(task_id):
+            if not await tracker.is_cancellation_requested(task_id):
                 raise
             await run_to_completion(lambda: self.cancel_recovered(task_id, account_id, user_id))
             raise

openviking/service/fs_service.py

@@ -739,7 +739,7 @@ async def _wait_for_refresh(self) -> Dict[str, Any]:
                 )
         except (TimeoutError, asyncio.TimeoutError) as exc:
             raise DeadlineExceededError("queue processing", None) from exc
-        return tracker.build_queue_status(
+        return await tracker.build_queue_status(
             context.task_id, work_ids if context.work_id is not None else None
         )
 

openviking/service/memory_compile.py

@@ -192,7 +192,11 @@ async def _run(
         ctx: RequestContext,
     ) -> None:
         tracker = get_task_tracker()
-        tracker.register_running_task(task_id)
+        await tracker.register_running_task(
+            task_id,
+            account_id=ctx.account_id,
+            user_id=ctx.user.user_id,
+        )
         root_token = None
         if bind_root_observability_context is not None:
             root_attrs = create_root_span_attributes(

openviking/service/reindex_executor.py

@@ -685,7 +685,7 @@ async def _run(
                 )
                 self._apply_embedding_wait_status(
                     counters,
-                    get_task_tracker().build_queue_status(task_context.task_id),
+                    await get_task_tracker().build_queue_status(task_context.task_id),
                 )
         finally:
             active_lease = run.lease if run is not None else lease

openviking/service/resource_service.py

@@ -2463,7 +2463,11 @@ async def create_skill_task() -> None:
             await task_tracker.start(
                 task.task_id, account_id=ctx.account_id, user_id=ctx.user.user_id
             )
-            task_tracker.register_running_task(task.task_id)
+            await task_tracker.register_running_task(
+                task.task_id,
+                account_id=ctx.account_id,
+                user_id=ctx.user.user_id,
+            )
             try:
                 with bind_task_context(task.task_id, ctx.account_id, ctx.user.user_id):
                     if isinstance(data, SkillProcessingPreparation):
@@ -2543,7 +2547,11 @@ async def create_skill_task() -> None:
                     "queue.wait.duration_ms",
                     round((time.perf_counter() - wait_start) * 1000, 3),
                 )
-                status = task_tracker.build_queue_status(task.task_id)
+                status = await task_tracker.build_queue_status(
+                    task.task_id,
+                    account_id=ctx.account_id,
+                    user_id=ctx.user.user_id,
+                )
                 result["queue_status"] = status
                 self._raise_queue_status_errors(status)
                 # Cancellation settles queue entries without counting errors.
@@ -2582,7 +2590,11 @@ async def cancel_skill_processing(self, task_id: str, ctx: RequestContext) -> No
             return  # Synchronous preparation failed before the task was created.
         # This includes queued work that must be discarded and active handlers
         # that must finish their cancellation/write cleanup, not just task status.
-        while tracker.has_work(task_id):
+        while await tracker.has_work(
+            task_id,
+            account_id=ctx.account_id,
+            user_id=ctx.user.user_id,
+        ):
             await asyncio.sleep(0.05)
 
     async def build_index(