feat(task): add async task tracking for add-resource and add-skill operations (#1763)

* feat: add async task tracking for add-resource/add-skill/write operations

- Return task_id when add-resource/add-skill/write called without --wait
- Add 'ov task status <task_id>' and 'ov task list' CLI commands
- Bridge RequestWaitTracker and TaskTracker via background monitor coroutines
- Format TaskRecord timestamps as ISO 8601 in to_dict()
- Always generate telemetry_id (remove 'not wait' condition)
- Extract _create_write_task helper to eliminate code duplication
- Add unregister_wait_telemetry in _monitor_write_queue for consistency
- Update CLI async prompt from 'ov wait' to 'ov task status <task_id>'
- Add 7 new tests covering async task tracking

* refactor: remove task_id from write operations, keep only add-resource/add-skill

Write operations (write/create_file/write_memory) are primarily called
by internal flows like session commit, which have their own task tracking.
Only add-resource and add-skill are user-facing CLI commands that need
task_id for async progress tracking.

* fix: address PR review feedback

1. Fix async failure path leaking request-scoped tracker/telemetry state
   - Add monitor_started flag to ensure cleanup when monitor coroutine
     hasn't been launched yet
   - Finally block now cleans up if wait or not telemetry_id or not monitor_started

2. Fix TaskRecord timestamp API compatibility break
   - Keep original created_at/updated_at as float (backward compatible)
   - Add new created_at_iso/updated_at_iso fields with ISO 8601 strings

3. Fix ruff format lint failure on resource_service.py

4. Add regression test for async failure cleanup
   - test_add_resource_async_failure_cleans_up_tracker verifies no
     RequestWaitTracker or telemetry registry state leaks when processor
     raises before task/monitor creation

* fix: add missing unregister_wait_telemetry in add_skill finally block

* fix: remove unused imports in test_add_resource_async_failure_cleans_up_tracker

* fix: queue failure shows as completed and business error creates unreachable task

- _monitor_queue_processing: check error_count from build_queue_status,
  mark task as failed when queue processing has errors
- add_resource: skip task creation when process_resource returns
  status=error, preventing unreachable ghost tasks
- Improve test_add_resource_async_failure_cleans_up_tracker: patch
  internal processor instead of add_resource itself to cover finally
  cleanup logic
- Add test_add_skill_async_returns_task_id for add_skill coverage
- Add test_add_resource_business_error_no_task regression test
- Add test_monitor_marks_failed_on_queue_error regression test

* test: move add_skill tests to test_session_task_tracking.py

Move add_skill task tracking tests from test_api_resources.py to
test_session_task_tracking.py where other task tracking tests live,
and add sync no-task-id coverage.

* test: update async task queryable assertion to include failed status

Queue errors now correctly mark task as failed (Bug 1 fix), so the
test assertion must accept 'failed' as a valid terminal status.

* refactor: use explicit return result for business error path

* fix: remove duplicate asyncio imports in test_api_resources.py

* fix: avoid task creation on watch conflict

---------

Co-authored-by: qin-ctx <qinhaojie.exe@bytedance.com>

Author: sponge225

crates/ov_cli/src/client.rs

@@ -576,6 +576,28 @@ impl HttpClient {
         }
     }
 
+    // ============ Task Methods ============
+
+    pub async fn get_task(&self, task_id: &str) -> Result<serde_json::Value> {
+        let path = format!("/api/v1/tasks/{}", task_id);
+        self.get(&path, &[]).await
+    }
+
+    pub async fn list_tasks(
+        &self,
+        task_type: Option<&str>,
+        status: Option<&str>,
+    ) -> Result<serde_json::Value> {
+        let mut params: Vec<(String, String)> = Vec::new();
+        if let Some(t) = task_type {
+            params.push(("task_type".to_string(), t.to_string()));
+        }
+        if let Some(s) = status {
+            params.push(("status".to_string(), s.to_string()));
+        }
+        self.get("/api/v1/tasks", &params).await
+    }
+
     // ============ Relation Methods ============
 
     pub async fn relations(&self, uri: &str) -> Result<serde_json::Value> {

crates/ov_cli/src/commands/mod.rs

@@ -11,3 +11,4 @@ pub mod resources;
 pub mod search;
 pub mod session;
 pub mod system;
+pub mod task;

crates/ov_cli/src/commands/resources.rs

@@ -44,10 +44,9 @@ pub async fn add_resource(
         )
         .await?;
 
-    // Print helpful message for async processing
     if !wait && matches!(format, OutputFormat::Table) {
         eprintln!("Note: Resource is being processed in the background.");
-        eprintln!("Use 'ov wait' to wait for completion, or 'ov observer queue' to check status.");
+        eprintln!("Use 'ov task status <task_id>' to check progress, or 'ov task list' to see all tasks.");
     }
 
     output_success(&result, format, compact);
@@ -64,10 +63,9 @@ pub async fn add_skill(
 ) -> Result<()> {
     let result = client.add_skill(data, wait, timeout).await?;
 
-    // Print helpful message for async processing
     if !wait && matches!(format, OutputFormat::Table) {
         eprintln!("Note: Skill is being processed in the background.");
-        eprintln!("Use 'ov wait' to wait for completion, or 'ov observer queue' to check status.");
+        eprintln!("Use 'ov task status <task_id>' to check progress, or 'ov task list' to see all tasks.");
     }
 
     output_success(&result, format, compact);

crates/ov_cli/src/commands/task.rs

@@ -0,0 +1,26 @@
+use crate::client::HttpClient;
+use crate::error::Result;
+use crate::output::{OutputFormat, output_success};
+
+pub async fn status(
+    client: &HttpClient,
+    task_id: &str,
+    output_format: OutputFormat,
+    compact: bool,
+) -> Result<()> {
+    let result = client.get_task(task_id).await?;
+    output_success(&result, output_format, compact);
+    Ok(())
+}
+
+pub async fn list(
+    client: &HttpClient,
+    task_type: Option<&str>,
+    status: Option<&str>,
+    output_format: OutputFormat,
+    compact: bool,
+) -> Result<()> {
+    let result = client.list_tasks(task_type, status).await?;
+    output_success(&result, output_format, compact);
+    Ok(())
+}

crates/ov_cli/src/main.rs

@@ -543,6 +543,11 @@ enum Commands {
         #[arg(long)]
         timeout: Option<f64>,
     },
+    /// [Status] Track async resource processing tasks
+    Task {
+        #[command(subcommand)]
+        action: TaskCommands,
+    },
     /// [Status] All OpenViking Server components status
     Status,
     /// [Status] Observe OpenViking Server components status
@@ -596,6 +601,24 @@ impl Commands {
     }
 }
 
+#[derive(Subcommand)]
+enum TaskCommands {
+    /// Show status of a specific task
+    Status {
+        /// Task ID returned by add-resource/add-skill
+        task_id: String,
+    },
+    /// List all tracked tasks
+    List {
+        /// Filter by task type (e.g. add_resource, add_skill, session_commit, reindex)
+        #[arg(long)]
+        task_type: Option<String>,
+        /// Filter by status (pending, running, completed, failed)
+        #[arg(long)]
+        status: Option<String>,
+    },
+}
+
 #[derive(Subcommand)]
 enum SystemCommands {
     /// Wait for queued async processing to complete
@@ -1041,6 +1064,23 @@ async fn main() {
             let client = ctx.get_client();
             commands::system::wait(&client, timeout, ctx.output_format, ctx.compact).await
         }
+        Commands::Task { action } => match action {
+            TaskCommands::Status { task_id } => {
+                let client = ctx.get_client();
+                commands::task::status(&client, &task_id, ctx.output_format, ctx.compact).await
+            }
+            TaskCommands::List { task_type, status } => {
+                let client = ctx.get_client();
+                commands::task::list(
+                    &client,
+                    task_type.as_deref(),
+                    status.as_deref(),
+                    ctx.output_format,
+                    ctx.compact,
+                )
+                .await
+            }
+        },
         Commands::Status => {
             let client = ctx.get_client();
             commands::observer::system(&client, ctx.output_format, ctx.compact).await

openviking/service/resource_service.py

@@ -6,6 +6,7 @@
 Provides resource management operations: add_resource, add_skill, wait_processed.
 """
 
+import asyncio
 import json
 import time
 from typing import TYPE_CHECKING, Any, Dict, List, Optional
@@ -161,7 +162,8 @@ async def add_resource(
         telemetry = get_current_telemetry()
         telemetry_id = register_wait_telemetry(wait)
         request_wait_tracker = get_request_wait_tracker()
-        if wait and telemetry_id:
+        monitor_started = False
+        if telemetry_id:
             request_wait_tracker.register_request(telemetry_id)
         watch_manager = self._get_watch_manager()
         watch_enabled = bool(
@@ -212,7 +214,9 @@ async def add_resource(
                 **kwargs,
             )
 
-            if wait:
+            if result.get("status") == "error":
+                return result
+            elif wait:
                 wait_start = time.perf_counter()
                 try:
                     with telemetry.measure("resource.wait"):
@@ -283,6 +287,24 @@ async def add_resource(
                             logger.warning(
                                 f"[ResourceService] Failed to cancel watch task for {to}: {e}"
                             )
+            if not wait:
+                from openviking.service.task_tracker import get_task_tracker
+
+                task_tracker = get_task_tracker()
+                root_uri = result.get("root_uri", "")
+                task = task_tracker.create(
+                    "add_resource",
+                    resource_id=root_uri,
+                    owner_account_id=ctx.account_id,
+                    owner_user_id=ctx.user.user_id,
+                )
+                result["task_id"] = task.task_id
+                if telemetry_id:
+                    monitor_started = True
+                    asyncio.create_task(self._monitor_queue_processing(task.task_id, telemetry_id))
+                else:
+                    task_tracker.start(task.task_id)
+                    task_tracker.complete(task.task_id, {"root_uri": root_uri})
             return result
         except Exception as exc:
             telemetry.set_error(
@@ -296,7 +318,28 @@ async def add_resource(
                 "resource.request.duration_ms",
                 round((time.perf_counter() - request_start) * 1000, 3),
             )
-            get_request_wait_tracker().cleanup(telemetry_id)
+            if wait or not telemetry_id or not monitor_started:
+                get_request_wait_tracker().cleanup(telemetry_id)
+                unregister_wait_telemetry(telemetry_id)
+
+    async def _monitor_queue_processing(self, task_id: str, telemetry_id: str) -> None:
+        from openviking.service.task_tracker import get_task_tracker
+
+        task_tracker = get_task_tracker()
+        request_wait_tracker = get_request_wait_tracker()
+        task_tracker.start(task_id)
+        try:
+            await request_wait_tracker.wait_for_request(telemetry_id)
+            status = request_wait_tracker.build_queue_status(telemetry_id)
+            errors = sum(int(group.get("error_count", 0) or 0) for group in status.values())
+            if errors:
+                task_tracker.fail(task_id, f"queue processing failed: {status}")
+            else:
+                task_tracker.complete(task_id, {"queue_status": status})
+        except Exception as exc:
+            task_tracker.fail(task_id, str(exc))
+        finally:
+            request_wait_tracker.cleanup(telemetry_id)
             unregister_wait_telemetry(telemetry_id)
 
     async def _handle_watch_task_creation(
@@ -434,7 +477,8 @@ async def add_skill(
         self._ensure_initialized()
         telemetry_id = get_current_telemetry().telemetry_id
         request_wait_tracker = get_request_wait_tracker()
-        if wait and telemetry_id:
+        monitor_started = False
+        if telemetry_id:
             request_wait_tracker.register_request(telemetry_id)
 
         try:
@@ -468,10 +512,28 @@ async def add_skill(
                     round((time.perf_counter() - wait_start) * 1000, 3),
                 )
                 result["queue_status"] = status
+            else:
+                from openviking.service.task_tracker import get_task_tracker
+
+                task_tracker = get_task_tracker()
+                task = task_tracker.create(
+                    "add_skill",
+                    owner_account_id=ctx.account_id,
+                    owner_user_id=ctx.user.user_id,
+                )
+                result["task_id"] = task.task_id
+                if telemetry_id:
+                    monitor_started = True
+                    asyncio.create_task(self._monitor_queue_processing(task.task_id, telemetry_id))
+                else:
+                    task_tracker.start(task.task_id)
+                    task_tracker.complete(task.task_id, {})
 
             return result
         finally:
-            request_wait_tracker.cleanup(telemetry_id)
+            if wait or not telemetry_id or not monitor_started:
+                request_wait_tracker.cleanup(telemetry_id)
+                unregister_wait_telemetry(telemetry_id)
 
     async def build_index(
         self, resource_uris: List[str], ctx: RequestContext, **kwargs

openviking/service/task_tracker.py

@@ -20,6 +20,7 @@
 import time
 from copy import deepcopy
 from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
 from enum import Enum
 from typing import Any, Dict, List, Optional
 from uuid import uuid4
@@ -57,6 +58,8 @@ def to_dict(self) -> Dict[str, Any]:
         """Serialize for JSON response."""
         d = asdict(self)
         d["status"] = self.status.value
+        d["created_at_iso"] = datetime.fromtimestamp(self.created_at, tz=timezone.utc).isoformat()
+        d["updated_at_iso"] = datetime.fromtimestamp(self.updated_at, tz=timezone.utc).isoformat()
         d.pop("owner_account_id", None)
         d.pop("owner_user_id", None)
         return d

openviking/telemetry/resource_summary.py

@@ -41,7 +41,7 @@ def _consume_semantic_dag_stats(telemetry_id: str, root_uri: str | None):
 def register_wait_telemetry(wait: bool) -> str:
     """Register current telemetry collector for async queue consumers when needed."""
     handle = get_current_telemetry()
-    if not wait or not handle.telemetry_id:
+    if not handle.telemetry_id:
         return ""
     if handle.enabled:
         register_telemetry(handle)