S10 Task System：持久化任务并管理依赖

@dataclass
class Task:
    id: str
    subject: str
    description: str
    status: str
    owner: str | None
    blockedBy: list[str]

{
  "id": "task_e5f6a7b8",
  "subject": "实现 API",
  "description": "实现接口并验证数据读写",
  "status": "pending",
  "owner": null,
  "blockedBy": ["task_a1b2c3d4"]
}

def save(self, task: Task) -> None:
        self._path(task.id, create_root=True).write_text(
            json.dumps(asdict(task), indent=2),
            encoding="utf-8",
        )

def load(self, task_id: str) -> Task:
        data = json.loads(self._path(task_id).read_text(encoding="utf-8"))
        task = Task(**data)
        if task.id != task_id:
            raise ValueError(f"Task file ID does not match {task_id}")
        if task.status not in ("pending", "in_progress", "completed"):
            raise ValueError(f"Invalid task status: {task.status}")
        return task

def list(self) -> list[Task]:
        if not self.directory.exists():
            return []
        root = self._root()
        return [self.load(path.stem)
                for path in sorted(root.glob("task_*.json"))]

def load_task(task_id: str) -> Task:
    return TASKS.load(task_id)

def list_tasks() -> list[Task]:
    return TASKS.list()

tool_use(name="create_task", input={...})
  -> execute_tool(block)
  -> TOOL_HANDLERS["create_task"]
  -> run_create_task(subject, description)
  -> create_task(subject, description)
  -> TASKS.create(subject, description)

def create_task(subject: str, description: str = "") -> Task:
    return TASKS.create(subject, description)

def create(self, subject: str, description: str = "") -> Task:
        subject = subject.strip()
        if not subject:
            raise ValueError("Task subject cannot be empty")

        self._root(create=True)
        for _ in range(100):
            task = Task(
                id=f"task_{secrets.token_hex(4)}",
                subject=subject,
                description=description,
                status="pending",
                owner=None,
                blockedBy=[],
            )
            try:
                with self._path(task.id, create_root=True).open(
                    "x", encoding="utf-8"
                ) as handle:
                    json.dump(asdict(task), handle, indent=2)
                return task
            except FileExistsError:
                continue
        raise RuntimeError("Could not allocate a unique task ID")

def run_create_task(subject: str, description: str = "") -> str:
    task = create_task(subject, description)
    print(f"  [create] {task.subject}")
    return f"Created {task.id}: {task.subject}"

def update_task(task_id: str, addBlockedBy: list[str]) -> Task:
    return TASKS.update_dependencies(task_id, addBlockedBy)

def update_dependencies(self, task_id: str,
                            add_blocked_by: list[str]) -> Task:
        if not isinstance(add_blocked_by, list):
            raise ValueError("addBlockedBy must be a list of task IDs")

        task = self.load(task_id)
        if task.status != "pending" or task.owner is not None:
            raise ValueError(
                f"Task {task_id} dependencies can only be updated while "
                "pending and unowned"
            )

        dependencies = list(dict.fromkeys(add_blocked_by))
        for dependency in dependencies:
            if dependency == task_id:
                raise ValueError("Task cannot depend on itself")
            if not self.exists(dependency):
                raise ValueError(f"Dependency not found: {dependency}")
            if dependency not in task.blockedBy and self._depends_on(
                dependency, task_id
            ):
                raise ValueError(
                    f"Dependency cycle detected: {task_id} -> {dependency}"
                )

        task.blockedBy.extend(
            dependency for dependency in dependencies
            if dependency not in task.blockedBy
        )
        self.save(task)
        return task

def _depends_on(self, task_id: str, target_id: str) -> bool:
        """Return whether task_id transitively depends on target_id."""
        pending = [task_id]
        visited = set()
        while pending:
            current = pending.pop()
            if current == target_id:
                return True
            if current in visited:
                continue
            visited.add(current)
            pending.extend(self.load(current).blockedBy)
        return False

def run_update_task(task_id: str, addBlockedBy: list[str]) -> str:
    task = update_task(task_id, addBlockedBy)
    dependencies = ", ".join(task.blockedBy) or "(none)"
    print(f"  [update] {task.subject} blockedBy: {dependencies}")
    return f"Updated {task.id} blockedBy: {dependencies}"

def get_task(task_id: str) -> str:
    return json.dumps(asdict(load_task(task_id)), indent=2)

def can_start(task_id: str) -> bool:
    return not incomplete_dependencies(load_task(task_id))

def incomplete_dependencies(task: Task) -> list[str]:
    incomplete = []
    for dependency in task.blockedBy:
        try:
            if load_task(dependency).status != "completed":
                incomplete.append(dependency)
        except (FileNotFoundError, ValueError):
            incomplete.append(dependency)
    return incomplete

unfinished = [t for t in list_tasks() if t.status != "completed"]
available = [t for t in unfinished
             if t.status == "pending" and can_start(t.id)]

def run_list_tasks() -> str:
    tasks = list_tasks()
    if not tasks:
        return "No tasks. Use create_task to add some."
    lines = []
    for task in tasks:
        marker = {
            "pending": "[ ]",
            "in_progress": "[>]",
            "completed": "[x]",
        }.get(task.status, "[?]")
        dependencies = (
            f" (blockedBy: {', '.join(task.blockedBy)})"
            if task.blockedBy else ""
        )
        owner = f" [{task.owner}]" if task.owner else ""
        lines.append(
            f"{marker} {task.id}: {task.subject} "
            f"[{task.status}]{owner}{dependencies}"
        )
    return "\n".join(lines)

def run_get_task(task_id: str) -> str:
    return get_task(task_id)

def claim_task(task_id: str, owner: str = "agent") -> str:
    task = load_task(task_id)
    if task.status != "pending":
        return f"Task {task_id} is {task.status}, cannot claim"
    dependencies = incomplete_dependencies(task)
    if dependencies:
        return f"Blocked by: {dependencies}"
    task.owner = owner
    task.status = "in_progress"
    TASKS.save(task)
    print(f"  [claim] {task.subject} -> in_progress (owner: {owner})")
    return f"Claimed {task.id} ({task.subject})"

def complete_task(task_id: str, owner: str = "agent") -> str:
    task = load_task(task_id)
    if task.status != "in_progress":
        return f"Task {task_id} is {task.status}, cannot complete"
    if task.owner != owner:
        return f"Task {task_id} is owned by {task.owner}, not {owner}"
    ready_before = {
        candidate.id
        for candidate in list_tasks()
        if candidate.status == "pending"
        and candidate.blockedBy
        and can_start(candidate.id)
    }
    task.status = "completed"
    TASKS.save(task)
    unblocked = [candidate.subject for candidate in list_tasks()
                 if candidate.status == "pending"
                 and candidate.blockedBy
                 and candidate.id not in ready_before
                 and can_start(candidate.id)]
    print(f"  [complete] {task.subject}")
    message = f"Completed {task.id} ({task.subject})"
    if unblocked:
        message += f"\nUnblocked: {', '.join(unblocked)}"
        print(f"  [unblocked] {', '.join(unblocked)}")
    return message

schema = create_task("建立数据库结构", "完成表结构与迁移验证")
api = create_task("实现 API", "实现接口并验证数据读写")
tests = create_task("补充测试", "验证 API 的成功与失败路径")
docs = create_task("编写文档", "记录数据结构与使用方式")
deploy = create_task("准备部署", "检查测试结果与交付资料")

update_task(api.id, [schema.id])
update_task(tests.id, [api.id])
update_task(docs.id, [schema.id])
update_task(deploy.id, [tests.id, docs.id])

claim_task(schema.id)
# Execute and verify the schema work here.
complete_task(schema.id)
claim_task(api.id)

{'name': 'create_task',
 'description': 'Create a task and return its runtime-generated ID.',
 'input_schema': {'type': 'object',
                  'properties': {'subject': {'type': 'string'},
                                 'description': {'type': 'string'}},
                  'required': ['subject'],
                  'additionalProperties': False}}

{
    "name": "update_task",
    "input_schema": {
        "type": "object",
        "properties": {
            "task_id": {"type": "string", "pattern": "^task_[0-9a-f]{8}$"},
            "addBlockedBy": {
                "type": "array",
                "items": {"type": "string", "pattern": "^task_[0-9a-f]{8}$"},
                "minItems": 1,
            },
        },
        "required": ["task_id", "addBlockedBy"],
        "additionalProperties": False,
    },
}

def run_claim_task(task_id: str) -> str:
    return claim_task(task_id, owner="agent")

def run_complete_task(task_id: str) -> str:
    return complete_task(task_id, owner="agent")

TOOLS = [*BASE_TOOLS, *TASK_TOOLS]
TOOL_HANDLERS = {
    **BASE_HANDLERS,
    "create_task": run_create_task,
    "update_task": run_update_task,
    "list_tasks": run_list_tasks,
    "get_task": run_get_task,
    "claim_task": run_claim_task,
    "complete_task": run_complete_task,
}

def execute_tool(block) -> str:
    blocked = trigger_hooks("PreToolUse", block)
    if blocked:
        return str(blocked)

    handler = TOOL_HANDLERS.get(block.name)
    try:
        output = handler(**block.input) if handler else f"Unknown: {block.name}"
    except Exception as error:
        output = f"Error: {error}"

    trigger_hooks("PostToolUse", block, output)
    return str(output)

def agent_loop(messages: list):
    while True:
        response = client.messages.create(
            model=MODEL,
            system=SYSTEM,
            messages=messages,
            tools=TOOLS,
            max_tokens=8000,
        )
        messages.append({"role": "assistant", "content": response.content})

        tool_calls = [
            block for block in response.content if block.type == "tool_use"
        ]
        if not tool_calls:
            force = trigger_hooks("Stop", messages)
            if force:
                messages.append({"role": "user", "content": force})
                continue
            return

        results = []
        for block in tool_calls:
            output = execute_tool(block)
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": output,
            })
        messages.append({"role": "user", "content": results})

messages = [user(query)]
while True:
    response = call_model(system=SYSTEM, tools=TOOLS, messages=messages)
    messages.append(assistant(response.content))
    calls = get_tool_use_blocks(response.content)
    if not calls:
        follow_up = trigger_stop_hooks(messages)
        if follow_up:
            messages.append(user(follow_up))
            continue
        break

    results = []
    for call in calls:
        # Existing entry: PreToolUse -> selected handler -> PostToolUse.
        # A task handler additionally uses TaskStore and state checks.
        output = execute_tool(call)
        results.append(tool_result(call.id, output))
    messages.append(user(results))