feat(resources): add resource watch scheduling and status tracking (volcengine#709)

* feat(resource): add watch interval support for resource monitoring

implement resource watch functionality that allows automatic monitoring and re-processing of resources at specified intervals. key features include:
- add watch_interval parameter to resource APIs
- create watch scheduler service for task execution
- handle conflict detection for active watch tasks
- provide watch status query capability
- include comprehensive tests and examples

the watch feature enables periodic automatic updates of resources without manual intervention, improving data freshness for frequently changing content

* feat(resources): add watch status tracking and improve resource processing

- Implement get_watch_status API for tracking resource watch status
- Add immediate persistence for first-time resource additions
- Improve file change detection with size comparison
- Refactor watch scheduler with better concurrency control
- Add test coverage for watch status and resource processing
- Remove unused watch manager references and clean up code

* refactor: improve code style and fix minor issues

- Simplify logging by removing redundant data copying
- Fix syntax errors in docstrings and string literals
- Add new fields to EmbeddingMsg class
- Improve line wrapping and formatting
- Update watch task storage URIs to use hidden files

* refactor(embedding_msg): simplify EmbeddingMsg constructor by removing unused fields

Remove media_uri, media_mime_type and id parameters as they are not used in the implementation

* feat(watch): add backup task recovery and simplify permission check

Add test case for recovering tasks from backup storage when primary is missing
Remove require_owner parameter from _check_permission as it's redundant with the existing role-based checks

* feat(resources): add watch_interval support for resource updates

Add watch_interval parameter to enable periodic resource updates. When target is specified, watch_interval > 0 creates/updates a watch task, while <= 0 disables it. Also simplify resource moving logic in ResourceProcessor by using direct mv operation.

* refactor(watch): remove deprecated get_watch_status functionality

remove get_watch_status method and related tests, update examples to use direct task access
update watch manager to use ConflictError for URI conflicts and include original_role in tasks
add validation for watch_interval requiring target URI

* fix(resource_service): validate watch interval before processing resource

Move watch interval validation earlier in the flow to fail fast when 'to' parameter is missing

* refactor(resource_processor): remove redundant temp_uri assignment

Author: myysy

crates/ov_cli/src/client.rs

@@ -480,6 +480,7 @@ impl HttpClient {
         include: Option<String>,
         exclude: Option<String>,
         directly_upload_media: bool,
+        watch_interval: f64,
     ) -> Result<serde_json::Value> {
         let path_obj = Path::new(path);
 
@@ -501,6 +502,7 @@ impl HttpClient {
                     "include": include,
                     "exclude": exclude,
                     "directly_upload_media": directly_upload_media,
+                    "watch_interval": watch_interval,
                 });
 
                 self.post("/api/v1/resources", &body).await
@@ -520,6 +522,7 @@ impl HttpClient {
                     "include": include,
                     "exclude": exclude,
                     "directly_upload_media": directly_upload_media,
+                    "watch_interval": watch_interval,
                 });
 
                 self.post("/api/v1/resources", &body).await
@@ -537,6 +540,7 @@ impl HttpClient {
                     "include": include,
                     "exclude": exclude,
                     "directly_upload_media": directly_upload_media,
+                    "watch_interval": watch_interval,
                 });
 
                 self.post("/api/v1/resources", &body).await
@@ -555,6 +559,7 @@ impl HttpClient {
                 "include": include,
                 "exclude": exclude,
                 "directly_upload_media": directly_upload_media,
+                "watch_interval": watch_interval,
             });
 
             self.post("/api/v1/resources", &body).await

crates/ov_cli/src/commands/resources.rs

@@ -16,6 +16,7 @@ pub async fn add_resource(
     include: Option<String>,
     exclude: Option<String>,
     directly_upload_media: bool,
+    watch_interval: f64,
     format: OutputFormat,
     compact: bool,
 ) -> Result<()> {
@@ -33,6 +34,7 @@ pub async fn add_resource(
             include,
             exclude,
             directly_upload_media,
+            watch_interval,
         )
         .await?;
     output_success(&result, format, compact);

crates/ov_cli/src/main.rs

@@ -96,6 +96,9 @@ enum Commands {
         /// Do not directly upload media files
         #[arg(long = "no-directly-upload-media", default_value_t = false)]
         no_directly_upload_media: bool,
+        /// Watch interval in minutes for automatic resource monitoring (0 = no monitoring)
+        #[arg(long, default_value = "0")]
+        watch_interval: f64,
     },
     /// Add a skill into OpenViking
     AddSkill {
@@ -525,6 +528,7 @@ async fn main() {
             include,
             exclude,
             no_directly_upload_media,
+            watch_interval,
         } => {
             handle_add_resource(
                 path,
@@ -539,6 +543,7 @@ async fn main() {
                 include,
                 exclude,
                 no_directly_upload_media,
+                watch_interval,
                 ctx,
             )
             .await
@@ -655,6 +660,7 @@ async fn handle_add_resource(
     include: Option<String>,
     exclude: Option<String>,
     no_directly_upload_media: bool,
+    watch_interval: f64,
     ctx: CliContext,
 ) -> Result<()> {
     let is_url = path.starts_with("http://") 
@@ -722,6 +728,7 @@ async fn handle_add_resource(
         include,
         exclude,
         directly_upload_media,
+        watch_interval,
         ctx.output_format,
         ctx.compact,
     ).await

docs/en/api/02-resources.md

@@ -45,6 +45,7 @@ Add a resource to the knowledge base.
 | instruction | str | No | "" | Special processing instructions |
 | wait | bool | No | False | Wait for semantic processing to complete |
 | timeout | float | No | None | Timeout in seconds (only used when wait=True) |
+| watch_interval | float | No | 0 | Watch interval (minutes). >0 enables/updates watch; <=0 disables watch. Only takes effect when target is provided |
 
 **Python SDK (Embedded / HTTP)**
 
@@ -168,6 +169,48 @@ curl -X POST http://localhost:1933/api/v1/system/wait \
 openviking add-resource ./documents/guide.md --wait
 ```
 
+**Example: Watch for Updates (watch_interval)**
+
+`watch_interval` is in minutes and periodically triggers re-processing for the specified target URI:
+
+- `watch_interval > 0`: create (or reactivate and update) a watch task for the `target`
+- `watch_interval <= 0`: disable (deactivate) the watch task for the `target`
+- watch tasks are only managed when `target` / CLI `--to` is provided
+
+If there is already an active watch task for the same `target`, submitting another request with `watch_interval > 0` returns a conflict error. Disable it first (`watch_interval = 0`) and then set a new interval.
+
+**Python SDK (Embedded / HTTP)**
+
+```python
+client.add_resource(
+    "./documents/guide.md",
+    target="viking://resources/documents/guide.md",
+    watch_interval=60,
+)
+```
+
+**HTTP API**
+
+```bash
+curl -X POST http://localhost:1933/api/v1/resources \
+  -H "Content-Type: application/json" \
+  -H "X-API-Key: your-key" \
+  -d '{
+    "path": "./documents/guide.md",
+    "target": "viking://resources/documents/guide.md",
+    "watch_interval": 60
+  }'
+```
+
+**CLI**
+
+```bash
+openviking add-resource ./documents/guide.md --to viking://resources/documents/guide.md --watch-interval 60
+
+# Disable watch
+openviking add-resource ./documents/guide.md --to viking://resources/documents/guide.md --watch-interval 0
+```
+
 ---
 
 ### export_ovpack()

docs/zh/api/02-resources.md

@@ -45,6 +45,7 @@ Input -> Parser -> TreeBuilder -> AGFS -> SemanticQueue -> Vector Index
 | instruction | str | 否 | "" | 特殊处理指令 |
 | wait | bool | 否 | False | 等待语义处理完成 |
 | timeout | float | 否 | None | 超时时间（秒），仅在 wait=True 时生效 |
+| watch_interval | float | 否 | 0 | 定时更新间隔（分钟）。>0 开启/更新定时任务；<=0 关闭（停用）定时任务。仅在指定 target 时生效 |
 
 **Python SDK (Embedded / HTTP)**
 
@@ -168,6 +169,48 @@ curl -X POST http://localhost:1933/api/v1/system/wait \
 openviking add-resource ./documents/guide.md --wait
 ```
 
+**示例：开启定时更新（watch_interval）**
+
+`watch_interval` 的单位为分钟，用于对指定的目标 URI 定期触发更新处理：
+
+- `watch_interval > 0`：创建（或重新激活并更新）该 `target` 的定时任务
+- `watch_interval <= 0`：关闭（停用）该 `target` 的定时任务
+- 只有在指定 `target` / CLI `--to` 时才会创建定时任务
+
+如果同一个 `target` 已存在激活中的定时任务，再次以 `watch_interval > 0` 提交会返回冲突错误；需要先将 `watch_interval` 设为 `0`（取消/停用）后再重新设置新的间隔。
+
+**Python SDK (Embedded / HTTP)**
+
+```python
+client.add_resource(
+    "./documents/guide.md",
+    target="viking://resources/documents/guide.md",
+    watch_interval=60,
+)
+```
+
+**HTTP API**
+
+```bash
+curl -X POST http://localhost:1933/api/v1/resources \
+  -H "Content-Type: application/json" \
+  -H "X-API-Key: your-key" \
+  -d '{
+    "path": "./documents/guide.md",
+    "target": "viking://resources/documents/guide.md",
+    "watch_interval": 60
+  }'
+```
+
+**CLI**
+
+```bash
+openviking add-resource ./documents/guide.md --to viking://resources/documents/guide.md --watch-interval 60
+
+# 取消监控
+openviking add-resource ./documents/guide.md --to viking://resources/documents/guide.md --watch-interval 0
+```
+
 ---
 
 ### export_ovpack()

examples/watch_resource_example.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
+# SPDX-License-Identifier: Apache-2.0
+"""
+Resource Watch Feature Example
+
+This example demonstrates how to use the resource watch feature in OpenViking.
+The watch feature allows you to automatically re-process resources at specified
+intervals.
+
+Key features:
+- Create resources with watch enabled
+- Update watch intervals (cancel then re-create)
+- Cancel watch tasks
+- Handle conflict errors
+"""
+
+import asyncio
+from pathlib import Path
+
+from openviking import AsyncOpenViking
+from openviking_cli.exceptions import ConflictError
+
+
+async def example_basic_watch():
+    client = AsyncOpenViking(path="./data_watch_example")
+    await client.initialize()
+
+    try:
+        test_file = Path("./test_resource.md")
+        test_file.write_text(
+            """# Test Resource
+
+## Content
+This is a test resource for watch functionality.
+
+## Version
+Version: 1.0
+"""
+        )
+
+        to_uri = "viking://resources/watched_resource"
+
+        print("\nAdding resource with watch_interval=60.0 minutes...")
+        result = await client.add_resource(
+            path=str(test_file),
+            to=to_uri,
+            reason="Example: monitoring a document",
+            instruction="Check for updates and re-index",
+            watch_interval=60.0,
+        )
+
+        print("Resource added successfully!")
+        print(f"  Root URI: {result['root_uri']}")
+    finally:
+        await client.close()
+
+
+async def example_update_watch_interval():
+    client = AsyncOpenViking(path="./data_watch_example")
+    await client.initialize()
+
+    try:
+        test_file = Path("./test_resource.md")
+        to_uri = "viking://resources/watched_resource"
+
+        print("\nUpdating watch interval by canceling then re-creating...")
+        await client.add_resource(
+            path=str(test_file),
+            to=to_uri,
+            watch_interval=0,
+        )
+        await client.add_resource(
+            path=str(test_file),
+            to=to_uri,
+            reason="Updated: more frequent monitoring",
+            watch_interval=120.0,
+        )
+        print("Watch task updated successfully!")
+    finally:
+        await client.close()
+
+
+async def example_cancel_watch():
+    client = AsyncOpenViking(path="./data_watch_example")
+    await client.initialize()
+
+    try:
+        test_file = Path("./test_resource.md")
+        to_uri = "viking://resources/watched_resource"
+
+        print("\nCancelling watch by setting interval to 0...")
+        await client.add_resource(
+            path=str(test_file),
+            to=to_uri,
+            watch_interval=0,
+        )
+        print("Watch task cancelled successfully!")
+    finally:
+        await client.close()
+
+
+async def example_handle_conflict():
+    client = AsyncOpenViking(path="./data_watch_example")
+    await client.initialize()
+
+    try:
+        test_file = Path("./test_resource.md")
+        to_uri = "viking://resources/conflict_example"
+
+        print("\nCreating first watch task...")
+        await client.add_resource(
+            path=str(test_file),
+            to=to_uri,
+            watch_interval=30.0,
+        )
+        print("  First watch task created successfully")
+
+        print("\nAttempting to create second watch task for same URI...")
+        try:
+            await client.add_resource(
+                path=str(test_file),
+                to=to_uri,
+                watch_interval=60.0,
+            )
+            print("  ERROR: This should not happen!")
+        except ConflictError as e:
+            print("  ConflictError caught as expected!")
+            print(f"  Error message: {e}")
+    finally:
+        await client.close()
+
+
+async def main():
+    print("\n" + "=" * 60)
+    print("OpenViking Resource Watch Examples")
+    print("=" * 60)
+
+    await example_basic_watch()
+    await example_update_watch_interval()
+    await example_cancel_watch()
+    await example_handle_conflict()
+
+    print("\n" + "=" * 60)
+    print("All examples completed!")
+    print("=" * 60)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+

openviking/async_client.py

@@ -190,6 +190,7 @@ async def add_resource(
         timeout: float = None,
         build_index: bool = True,
         summarize: bool = False,
+        watch_interval: float = 0,
         telemetry: TelemetryRequest = False,
         **kwargs,
     ) -> Dict[str, Any]:
@@ -223,9 +224,14 @@ async def add_resource(
             build_index=build_index,
             summarize=summarize,
             telemetry=telemetry,
+            watch_interval=watch_interval,
             **kwargs,
         )
 
+    @property
+    def _service(self):
+        return self._client.service
+
     async def wait_processed(self, timeout: float = None) -> Dict[str, Any]:
         """Wait for all queued processing to complete."""
         await self._ensure_initialized()