Refactor task update methods in TaskRepository

- Consolidated multiple task update methods (status, inputs, dependencies, name, priority, params, schemas) into a single method `update_task` that accepts arbitrary fields.
- Updated related type hints and documentation to reflect changes.
- Modified all references to the old update methods in tests and other parts of the codebase to use the new `update_task` method.
- Removed unused JSON extraction functions from helpers and replaced them with a more efficient JSON finder.
- Updated tests to ensure compatibility with the new task update method and removed redundant tests for the deleted JSON extraction functions.

Author: tercel

docs/api/python.md

@@ -306,13 +306,7 @@ Database operations for tasks.
 - `get_task_by_id(task_id)`: Get task by ID
 - `get_root_task(task)`: Get root task
 - `build_task_tree(task)`: Build task tree from task
-- `update_task_status(task_id, status, ...)`: Update task status and related fields (error, result, progress, timestamps)
-- `update_task_inputs(task_id, inputs)`: Update task inputs
-- `update_task_dependencies(task_id, dependencies)`: Update task dependencies (with validation)
-- `update_task_name(task_id, name)`: Update task name
-- `update_task_priority(task_id, priority)`: Update task priority
-- `update_task_params(task_id, params)`: Update executor parameters
-- `update_task_schemas(task_id, schemas)`: Update validation schemas
+- `update_task(task_id, **kwarg)` Update task fields
 - `delete_task(task_id)`: Physically delete a task from the database
 - `get_all_children_recursive(task_id)`: Recursively get all child tasks (including grandchildren)
 - `find_dependent_tasks(task_id)`: Find all tasks that depend on a given task (reverse dependencies)
@@ -513,7 +507,7 @@ async def my_hook(task):
     repo = get_hook_repository()
     if repo:
         # Modify task fields
-        await repo.update_task_priority(task.id, 10)
+        await repo.update_task(task.id, priority=10)
         # Query other tasks
         pending = await repo.get_tasks_by_status("pending")
 ```

docs/api/quick-reference.md

@@ -113,33 +113,12 @@ task = await task_manager.task_repository.get_task_by_id(task_id)
 
 ```python
 # Update status and related fields
-await task_repository.update_task_status(
+await task_repository.update_task(
     task_id,
     status="completed",
     result={"data": "result"},
     progress=1.0
 )
-
-# Update inputs
-await task_repository.update_task_inputs(task_id, {"key": "new_value"})
-
-# Update name
-await task_repository.update_task_name(task_id, "New Task Name")
-
-# Update priority
-await task_repository.update_task_priority(task_id, 2)
-
-# Update params
-await task_repository.update_task_params(task_id, {"executor_id": "new_executor"})
-
-# Update schemas
-await task_repository.update_task_schemas(task_id, {"input_schema": {...}})
-
-# Update dependencies (only for pending tasks, with validation)
-await task_repository.update_task_dependencies(
-    task_id,
-    [{"id": "dep-task-id", "required": True}]
-)
 ```
 
 **Critical Field Validation:**
@@ -389,8 +368,7 @@ async def modify_task_with_db(task):
     repo = get_hook_repository()
     if repo:
         # Update task fields
-        await repo.update_task_name(task.id, "New Name")
-        await repo.update_task_priority(task.id, 10)
+        await repo.update_task(task.id, name="New Name")
 
         # Query other tasks
         pending_tasks = await repo.get_tasks_by_status("pending")

docs/architecture/task-tree-lifecycle.md

@@ -184,7 +184,7 @@ async def execute_task_tree(
    task = await self.task_repository.get_task_by_id(task_id)
 
    # Update to in_progress
-   await self.task_repository.update_task_status(
+   await self.task_repository.update_task(
        task_id=task_id,
        status="in_progress",
        started_at=datetime.now(timezone.utc)
@@ -197,7 +197,7 @@ async def execute_task_tree(
    resolved_inputs = await resolve_task_dependencies(task, self.task_repository)
 
    if resolved_inputs != (task.inputs or {}):
-       await self.task_repository.update_task_inputs(task_id, resolved_inputs)
+       await self.task_repository.update_task(task_id, inputs=resolved_inputs)
    ```
 
 3. **Pre-Hook Execution** (lines 761-794)
@@ -210,7 +210,7 @@ async def execute_task_tree(
 
    # Auto-persist if inputs changed
    if inputs_after_pre_hooks != inputs_before_pre_hooks:
-       await self.task_repository.update_task_inputs(task_id, inputs_to_save)
+       await self.task_repository.update_task(task_id, inputs=inputs_to_save)
        task = await self.task_repository.get_task_by_id(task_id)  # Refresh
    ```
 
@@ -223,7 +223,7 @@ async def execute_task_tree(
 5. **Status Update and Cleanup** (lines 832-856)
    ```python
    # Update task status
-   await self.task_repository.update_task_status(
+   await self.task_repository.update_task(
        task_id=task_id,
        status="completed",
        progress=1.0,
@@ -267,11 +267,11 @@ TaskExecutor.execute_task_tree (session created)
 │   ├── on_tree_started hooks
 │   ├── _execute_task_tree_recursive
 │   │   ├── _execute_single_task (task 1)
-│   │   │   ├── update_task_status
+│   │   │   ├── update_task
 │   │   │   ├── resolve_task_dependencies
 │   │   │   ├── pre-hooks (can modify task.inputs)
 │   │   │   ├── execute task
-│   │   │   ├── update_task_status
+│   │   │   ├── update_task
 │   │   │   └── post-hooks
 │   │   ├── _execute_single_task (task 2)
 │   │   └── ...
@@ -285,12 +285,6 @@ TaskExecutor.execute_task_tree (session created)
 
 **Per-Operation Commits**:
 - Each `TaskRepository` method commits its own transaction
-- Example in `update_task_status()`:
-  ```python
-  await self.db.commit()
-  flag_modified(task, "result")  # For JSON fields
-  await self.db.refresh(task)    # Ensure fresh data
-  ```
 
 **Benefits**:
 - No cascading rollbacks across the entire task tree
@@ -439,7 +433,7 @@ async def update_metadata(task, inputs, result):
     repo = get_hook_repository()
 
     # Explicit repository call required for non-inputs fields
-    await repo.update_task_params(
+    await repo.update_task(
         task_id=task.id,
         params={"processed_at": datetime.now().isoformat()}
     )
@@ -458,7 +452,7 @@ async def aggregate_results(root_task, status):
 
     # Aggregate and update
     total_tokens = sum(t.result.get("token_usage", 0) for t in all_tasks if t.result)
-    await repo.update_task_result(
+    await repo.update_task(
         task_id=root_task.id,
         result={"total_tokens": total_tokens}
     )
@@ -474,7 +468,7 @@ try:
     task_result = await self._execute_task_with_schemas(task, final_inputs)
 except Exception as e:
     # Update task status to failed
-    await self.task_repository.update_task_status(
+    await self.task_repository.update_task(
         task_id=task_id,
         status="failed",
         error=str(e),

docs/development/design/cli-design.md

@@ -239,7 +239,7 @@ Return: {
 ```
 1. CLI/API: tasks cancel task-123
    ↓
-2. TaskRepository.update_task_status(task_id, status="cancelled")
+2. TaskRepository.update_task(task_id, status="cancelled")
    ↓
 3. TaskManager._execute_single_task() checks status at multiple points:
    - Before starting execution

docs/development/extending.md

@@ -261,17 +261,11 @@ async def modify_task_fields(task):
         return  # Not in hook context
 
     # Modify task fields explicitly
-    await repo.update_task_name(task.id, "Modified Name")
-    await repo.update_task_priority(task.id, 10)
+    await repo.update_task(task.id, name="Modified Name", priority=10)
 
     # Query other tasks
     pending_tasks = await repo.get_tasks_by_status("pending")
     print(f"Found {len(pending_tasks)} pending tasks")
-    
-    # Modify dependency tasks
-    if task.dependencies:
-        dep_id = task.dependencies[0]["id"]
-        await repo.update_task_priority(dep_id, 100)
 ```
 
 **Key Points:**
@@ -292,11 +286,7 @@ Hook database access is managed through a context that spans the entire task tre
 For detailed lifecycle information, see [Task Tree Execution Lifecycle](../architecture/task-tree-lifecycle.md).
 
 **Available Hook Repository Methods:**
-- `update_task_name(task_id, name)` - Update task name
-- `update_task_priority(task_id, priority)` - Update task priority
-- `update_task_status(task_id, status)` - Update task status
-- `update_task_params(task_id, params)` - Update task params
-- `update_task_inputs(task_id, inputs)` - Update task inputs (usually not needed, direct modification is auto-saved)
+- `update_task(task_id, **kwarg)` - Update task (usually not needed, direct modification is auto-saved)
 - `get_task_by_id(task_id)` - Query task by ID
 - `get_tasks_by_status(status)` - Query tasks by status
 - And all other TaskRepository methods...

docs/guides/best-practices.md

@@ -384,7 +384,7 @@ async def adjust_priority_by_load(task):
 
     # Adjust priority if system is overloaded
     if pending_count > 100:
-        await repo.update_task_priority(task.id, task.priority + 1)
+        await repo.update_task(task.id, priority=task.priority + 1)
 ```
 
 **Remember:**

pyproject.toml

@@ -135,6 +135,7 @@ dev = [
     "mypy>=1.5.0",
     "build>=1.0.0",
     "twine>=4.0.0",
+    "jsonfinder>=0.4.0",
     "pre-commit>=3.0.0",
 ]
 

src/apflow/api/routes/tasks.py

@@ -1288,7 +1288,7 @@ async def handle_task_update(self, params: dict, request: Request, request_id: s
                 # Update status and related fields if provided
                 status = params.get("status")
                 if status is not None:
-                    await task_repository.update_task_status(
+                    await task_repository.update_task(
                         task_id=task_id,
                         status=status,
                         error=params.get("error"),
@@ -1300,44 +1300,44 @@ async def handle_task_update(self, params: dict, request: Request, request_id: s
                 else:
                     # Update individual status-related fields if status is not provided
                     if "error" in params:
-                        await task_repository.update_task_status(
+                        await task_repository.update_task(
                             task_id=task_id, status=task.status, error=params.get("error")
                         )
                     if "result" in params:
-                        await task_repository.update_task_status(
+                        await task_repository.update_task(
                             task_id=task_id, status=task.status, result=params.get("result")
                         )
                     if "progress" in params:
-                        await task_repository.update_task_status(
+                        await task_repository.update_task(
                             task_id=task_id, status=task.status, progress=params.get("progress")
                         )
                     if "started_at" in params:
-                        await task_repository.update_task_status(
+                        await task_repository.update_task(
                             task_id=task_id, status=task.status, started_at=params.get("started_at")
                         )
                     if "completed_at" in params:
-                        await task_repository.update_task_status(
+                        await task_repository.update_task(
                             task_id=task_id, status=task.status, completed_at=params.get("completed_at")
                         )
 
                 # Update other fields
                 if "inputs" in params:
-                    await task_repository.update_task_inputs(task_id, params["inputs"])
+                    await task_repository.update_task(task_id, inputs=params["inputs"])
 
                 if "dependencies" in params:
-                    await task_repository.update_task_dependencies(task_id, params["dependencies"])
+                    await task_repository.update_task(task_id, dependencies=params["dependencies"])
 
                 if "name" in params:
-                    await task_repository.update_task_name(task_id, params["name"])
+                    await task_repository.update_task(task_id, name=params["name"])
 
                 if "priority" in params:
-                    await task_repository.update_task_priority(task_id, params["priority"])
+                    await task_repository.update_task(task_id, priority=params["priority"])
 
                 if "params" in params:
-                    await task_repository.update_task_params(task_id, params["params"])
+                    await task_repository.update_task(task_id, params=params["params"])
 
                 if "schemas" in params:
-                    await task_repository.update_task_schemas(task_id, params["schemas"])
+                    await task_repository.update_task(task_id, schemas=params["schemas"])
 
                 # Refresh task to get updated values
                 updated_task = await task_repository.get_task_by_id(task_id)

src/apflow/cli/commands/tasks.py

@@ -694,7 +694,7 @@ def create(
     Args:
         file: JSON file containing task(s) definition
         stdin: Read from stdin instead of file
-        
+
     """
     try:
         import sys
@@ -841,7 +841,7 @@ async def update_task():
 
             # Update status-related fields if status is provided
             if "status" in update_params:
-                await task_repository.update_task_status(
+                await task_repository.update_task(
                     task_id=task_id,
                     status=update_params["status"],
                     error=update_params.get("error"),
@@ -851,35 +851,35 @@ async def update_task():
             else:
                 # Update individual fields
                 if "error" in update_params:
-                    await task_repository.update_task_status(
+                    await task_repository.update_task(
                         task_id=task_id,
                         status=task.status,
                         error=update_params["error"]
                     )
                 if "result" in update_params:
-                    await task_repository.update_task_status(
+                    await task_repository.update_task(
                         task_id=task_id,
                         status=task.status,
                         result=update_params["result"]
                     )
                 if "progress" in update_params:
-                    await task_repository.update_task_status(
+                    await task_repository.update_task(
                         task_id=task_id,
                         status=task.status,
                         progress=update_params["progress"]
                     )
 
             # Update other fields
             if "name" in update_params:
-                await task_repository.update_task_name(task_id, update_params["name"])
+                await task_repository.update_task(task_id, name=update_params["name"])
             if "priority" in update_params:
-                await task_repository.update_task_priority(task_id, update_params["priority"])
+                await task_repository.update_task(task_id, priority=update_params["priority"])
             if "inputs" in update_params:
-                await task_repository.update_task_inputs(task_id, update_params["inputs"])
+                await task_repository.update_task(task_id, inputs=update_params["inputs"])
             if "params" in update_params:
-                await task_repository.update_task_params(task_id, update_params["params"])
+                await task_repository.update_task(task_id, params=update_params["params"])
             if "schemas" in update_params:
-                await task_repository.update_task_schemas(task_id, update_params["schemas"])
+                await task_repository.update_task(task_id, schemas=update_params["schemas"])
 
             # Get updated task
             updated_task = await task_repository.get_task_by_id(task_id)