orchestrata_ultimate.tools.task_tools

Reference › orchestrata_ultimate › tools › task_tools

This module provides the two task management tools of SPEC-069: TaskStopTool and TaskOutputTool, backed by an in-process TaskManager. They are registered through register_task_tools, by agents.tool_registry, and exported from the package __init__. tests/tools/test_task_tools.py covers them.

The module records a spec fork: the base create, get, list and update tasks that SPEC-069 extends do not exist as tools in this codebase, so TaskManager implements those operations itself as a thread-safe store (create, get, list, update, append_output). It also states that the Redis-cluster store, WebSocket streaming and external alerting the spec describes are not built.

What TaskManager does for real:

• Stops a genuine subprocess.Popen with SIGTERM and escalates to SIGKILL, using the graceful timeout and retry settings in TaskStopConfig, with modes in TaskStopMode.
• Authorises a stop by ownership and by critical or protected task status, with an admin override. This is the in-process analogue of a governance rule; wiring to the governance engine is noted as follow-up.
• Generates collision-safe task ids and makes a second concurrent stop a clean error through a per-record flag.
• Applies a per-agent rate limit, keeps an audit log, and purges expired output with purge_expired_output.

TaskStopTool.execute stops by task_id or batch-stops by name with exact or pattern matching through stop_by_pattern; requesting_agent_id is required. TaskOutputTool.execute returns full, tail, line-range or streamed output (OutputMode) or the task status, after a rate-limit check. Both return failed ToolResult values for TaskToolError and other exceptions.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

TaskStatus (enum)

Bases: StrEnum · Line 77

Task lifecycle state (spec.md Terminology "Task State").

TaskStopMode (enum)

Bases: StrEnum · Decorators: traces(spec='069', story='US-069-001', reqs=['REQ-069-024', 'REQ-069-025']) · Line 95

How :class:TaskStopTool terminates a task (spec.md TaskStopMode).

OutputMode (enum)

Bases: StrEnum · Decorators: traces(spec='069', story='US-069-002', reqs=['REQ-069-028', 'REQ-069-029', 'REQ-069-030', 'REQ-069-031']) · Line 108

How :class:TaskOutputTool windows captured output (spec.md OutputMode).

TaskType (enum)

Bases: StrEnum · Decorators: traces(spec='069', story='US-069-002', reqs=['REQ-069-032']) · Line 118

Task category for governance (TaskStatusResult schema).

PatternType (enum)

Bases: StrEnum · Decorators: traces(spec='069', story='US-069-003', reqs=['REQ-069-023']) · Line 129

Name-matching mode for batch stop / list (spec.md TaskStopByNameRequest).

TaskStopConfig (dataclass)

Decorators: dataclass(slots=True, frozen=True) · Line 142

Mirrors spec.md's TASK_STOP_CONFIG.

TaskOutputConfig (dataclass)

Decorators: dataclass(slots=True, frozen=True) · Line 152

Mirrors spec.md's TASK_OUTPUT_CONFIG.

TaskToolErrorCode

Decorators: traces(spec='069', story='US-069-001', reqs=['REQ-069-034', 'REQ-069-035', 'REQ-069-036', 'REQ-069-037', 'REQ-069-038',… · Line 182

Error codes for the task-management tools (spec.md Error Conditions).

TaskToolError

Bases: Exception · Line 208

Raised by :class:TaskManager on any spec.md ERR-00N condition.

TaskStatusResult (dataclass)

Decorators: traces(spec='069', story='US-069-002', reqs=['REQ-069-032']), dataclass(slots=True, frozen=True) · Line 292

Mirrors spec.md's TaskStatusResult JSON Schema (FR-011).

TaskOutput (dataclass)

Decorators: traces(spec='069', story='US-069-002', reqs=['REQ-069-028', 'REQ-069-033']), dataclass(slots=True, frozen=True) · Line 312

Mirrors spec.md's TaskOutput JSON Schema (FR-007/FR-012).

StopOutcome (dataclass)

Decorators: traces(spec='069', story='US-069-001', reqs=['REQ-069-022']), dataclass(slots=True, frozen=True) · Line 328

Result of a single :meth:TaskManager.stop call (TaskStopByIdResponse.data).

BatchStopOutcome (dataclass)

Decorators: traces(spec='069', story='US-069-003', reqs=['REQ-069-027']), dataclass(slots=True, frozen=True) · Line 340

Result of :meth:TaskManager.stop_by_pattern (TaskStopByNameResponse.data).

_OutputLine (dataclass)

Decorators: dataclass(slots=True) · Line 354

_TaskRecord (dataclass)

Decorators: traces(spec='069', story='US-069-001', reqs=['REQ-069-016', 'REQ-069-017']), dataclass(slots=True) · Line 363

TaskManager

Line 392

Thread-safe in-process task store backing create/get/list/update/stop/output.

TaskStopTool

Bases: BaseTool · Decorators: traces(spec='069', story='US-069-001', reqs=['REQ-069-022', 'REQ-069-023']) · Line 1041

The registrable BaseTool for stopping tasks (FR-001..006).

TaskOutputTool

Bases: BaseTool · Decorators: traces(spec='069', story='US-069-002', reqs=['REQ-069-028', 'REQ-069-030', 'REQ-069-031', 'REQ-069-032', 'REQ-069-033']) · Line 1208

The registrable BaseTool for output retrieval + status (FR-007..012).

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/tools/task_tools.py

"""Task management tools — TaskStopTool & TaskOutputTool (SPEC-069).

**Spec fork (read before extending):** SPEC-069's "Extends" metadata names
065-task-management-create/066-...-update/067-...-get/068-...-list as the
base CRUD tools this suite builds on. Those spec numbers do not describe
task-management tools in this codebase — ``specs/065-lsp-tool``,
``066-powershell-tool``, ``067-remote-trigger-tool``, and
``068-schedule-cron-tool`` are unrelated tools, and no TaskCreate/TaskUpdate/
TaskGet/TaskList ``BaseTool`` exists anywhere in the tree. Per the build
brief, this module does not fabricate specs/tools for that missing chain.
Instead :class:`TaskManager` is the in-process, thread-safe backing store
that implements the base create/get/list/update operations SPEC-069's two
tools need, and :class:`TaskStopTool` / :class:`TaskOutputTool` are the two
``BaseTool`` classes this spec actually defines and registers.

SPEC-069 also describes infrastructure this codebase does not have wired up
(a Redis-cluster-backed store with TLS + at-rest encryption, a WebSocket
transport for live streaming, Prometheus/PagerDuty export). Those pieces are
NOT fabricated here; every such gap is called out inline and in
``specs/069-task-management-tools/checklists/requirements.md``. What IS real
and load-bearing in this module:

    - Real OS process termination (SIGTERM -> escalate to SIGKILL) against a
      genuine :class:`subprocess.Popen`, with the exact graceful-timeout /
      retry-attempt / retry-delay knobs spec.md's ``TASK_STOP_CONFIG`` names.
    - Real ownership/critical/protected-task authorization (the in-process
      analogue of SEC-REQ-003's Prolog ``may_stop_task/2`` gate — wiring to
      the real governance engine is a follow-up integration, not invented
      here).
    - Real, deterministic collision-safe task-ID generation
      (``secrets.token_urlsafe``, retried under a lock) and idempotent stop
      handling (a per-record ``stopping`` flag makes a second concurrent
      stop call a clean no-op error instead of a double signal).
    - Real in-process rate limiting and output-retention purging — the
      process-local analogues of SEC-REQ-012/ERR-013 that don't require
      Redis to implement correctly.

Architecture (composition over inheritance, one ABC/Protocol boundary, per
``plans/build-factory-architecture.md`` — mirrors ``tools/search.py``):
    - :class:`TaskStopTool` / :class:`TaskOutputTool` are the two
      :class:`~orchestrata_ultimate.tools.base.BaseTool` (Command) instances
      callers/LLMs see. Both compose the same :class:`TaskManager`.
    - :class:`TaskManager` owns all task state (a ``dict[str, _TaskRecord]``
      behind a single :class:`threading.RLock`), process signalling, output
      windowing, rate limiting, and the in-process audit log.
"""

from __future__ import annotations

import json
import re
import secrets
import subprocess
import threading
import time
from collections import deque
from collections.abc import Callable
from dataclasses import asdict, dataclass, field
from datetime import UTC, datetime, timedelta
from enum import StrEnum
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.tracing.tracer import TraceConfig, Tracer
from orchestrata_ultimate.tools.base import (
    BaseTool,
    ToolCategory,
    ToolMetadata,
    ToolPriority,
    ToolResult,
)
from orchestrata_ultimate.tools.registry import ToolRegistry

# ── enums (spec.md Type Definitions + Terminology) ──────────────────────


class TaskStatus(StrEnum):
    """Task lifecycle state (spec.md Terminology "Task State")."""

    PENDING = "pending"
    RUNNING = "running"
    STOPPED = "stopped"
    COMPLETED = "completed"
    FAILED = "failed"
    TIMEOUT = "timeout"


#: Terminal states a task can never leave (idempotency guard for stop/cancel).
_TERMINAL_STATUSES = frozenset(
    {TaskStatus.STOPPED, TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.TIMEOUT}
)


@traces(spec="069", story="US-069-001", reqs=["REQ-069-024", "REQ-069-025"])
class TaskStopMode(StrEnum):
    """How :class:`TaskStopTool` terminates a task (spec.md TaskStopMode)."""

    GRACEFUL = "graceful"
    FORCE = "force"
    CANCEL = "cancel"


@traces(
    spec="069",
    story="US-069-002",
    reqs=["REQ-069-028", "REQ-069-029", "REQ-069-030", "REQ-069-031"],
)
class OutputMode(StrEnum):
    """How :class:`TaskOutputTool` windows captured output (spec.md OutputMode)."""

    FULL = "full"
    TAIL = "tail"
    LINES = "lines"
    STREAM = "stream"


@traces(spec="069", story="US-069-002", reqs=["REQ-069-032"])
class TaskType(StrEnum):
    """Task category for governance (TaskStatusResult schema)."""

    BUILD = "build"
    TEST = "test"
    DEPLOY = "deploy"
    PROCESS = "process"
    CUSTOM = "custom"


@traces(spec="069", story="US-069-003", reqs=["REQ-069-023"])
class PatternType(StrEnum):
    """Name-matching mode for batch stop / list (spec.md TaskStopByNameRequest)."""

    EXACT = "exact"
    REGEX = "regex"
    PREFIX = "prefix"
    SUFFIX = "suffix"


# ── configuration (spec.md Technical Design > Configuration) ────────────


@dataclass(slots=True, frozen=True)
class TaskStopConfig:
    """Mirrors spec.md's ``TASK_STOP_CONFIG``."""

    graceful_timeout: float = 10.0
    max_concurrent_stops: int = 10
    retry_attempts: int = 3
    retry_delay: float = 1.0


@dataclass(slots=True, frozen=True)
class TaskOutputConfig:
    """Mirrors spec.md's ``TASK_OUTPUT_CONFIG``."""

    max_output_bytes: int = 100 * 1024 * 1024
    retention_days: int = 7
    tail_lines_default: int = 100
    rate_limit_per_minute: int = 50


# ── errors (spec.md Error Conditions) ────────────────────────────────────


@traces(
    spec="069",
    story="US-069-001",
    reqs=[
        "REQ-069-034",
        "REQ-069-035",
        "REQ-069-036",
        "REQ-069-037",
        "REQ-069-038",
        "REQ-069-039",
        "REQ-069-040",
        "REQ-069-041",
        "REQ-069-042",
        "REQ-069-044",
        "REQ-069-046",
        "REQ-069-048",
    ],
)
class TaskToolErrorCode:
    """Error codes for the task-management tools (spec.md Error Conditions).

    ERR-010 (STREAM_FAILED), ERR-012 (STORAGE_CORRUPTION), and ERR-014
    (REDIS_ERROR) have no code here: this MVP has no WebSocket transport, no
    persistent/corruptible storage, and no Redis dependency (see the module
    docstring's fork note), so those conditions cannot occur.
    """

    TASK_NOT_FOUND = "TASK_NOT_FOUND"
    TASK_NOT_RUNNING = "TASK_NOT_RUNNING"
    PERMISSION_DENIED = "PERMISSION_DENIED"
    TERMINATION_FAILED = "TERMINATION_FAILED"
    CRITICAL_TASK = "CRITICAL_TASK"
    PROTECTED_TASK = "PROTECTED_TASK"
    OUTPUT_NOT_AVAILABLE = "OUTPUT_NOT_AVAILABLE"
    INVALID_LINE_RANGE = "INVALID_LINE_RANGE"
    OUTPUT_TOO_LARGE = "OUTPUT_TOO_LARGE"
    TASK_ID_COLLISION = "TASK_ID_COLLISION"
    RATE_LIMIT_EXCEEDED = "RATE_LIMIT_EXCEEDED"
    #: Not a spec.md ERR-0NN code: covers malformed-input rejections
    #: (SEC-REQ-006/007/008) that the spec's error table never assigned a
    #: dedicated code to.
    INVALID_INPUT = "INVALID_INPUT"


class TaskToolError(Exception):
    """Raised by :class:`TaskManager` on any spec.md ERR-00N condition."""

    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code
        self.message = message

    def to_result(self, tool_name: str) -> ToolResult:
        """Render this error as a failed :class:`ToolResult` for ``tool_name``."""
        return ToolResult(
            tool_name=tool_name,
            success=False,
            output="",
            error=f"{self.code}: {self.message}",
        )


# ── input validation (SEC-REQ-006/007/008/009) ──────────────────────────

_TASK_ID_PATTERN = re.compile(r"^task_[a-zA-Z0-9_-]+$")
_CONTROL_CHARS = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f]")
_MAX_STRING_LEN = 255
_MAX_ID_ATTEMPTS = 5


@traces(spec="069", story="US-069-001", reqs=["REQ-069-006"])
def _validate_task_id(task_id: Any) -> None:
    """Raise ``INVALID_INPUT`` unless ``task_id`` matches ``task_[a-zA-Z0-9_-]+``."""
    if not isinstance(task_id, str) or not _TASK_ID_PATTERN.match(task_id):
        raise TaskToolError(
            TaskToolErrorCode.INVALID_INPUT,
            f"task_id must match pattern {_TASK_ID_PATTERN.pattern!r}",
        )


@traces(spec="069", story="US-069-001", reqs=["REQ-069-007"])
def _sanitize_string(value: str) -> str:
    """Strip control characters and clamp to :data:`_MAX_STRING_LEN`."""
    return _CONTROL_CHARS.sub("", value).strip()[:_MAX_STRING_LEN]


@traces(spec="069", story="US-069-001", reqs=["REQ-069-007", "REQ-069-008"])
def _validate_name(name: Any) -> None:
    """Raise ``INVALID_INPUT`` unless ``name`` is a non-blank, bounded string."""
    if not isinstance(name, str) or not name.strip():
        raise TaskToolError(
            TaskToolErrorCode.INVALID_INPUT, "name must be a non-blank string"
        )
    if len(name) > _MAX_STRING_LEN:
        raise TaskToolError(
            TaskToolErrorCode.INVALID_INPUT,
            f"name exceeds max length {_MAX_STRING_LEN}",
        )


@traces(spec="069", story="US-069-003", reqs=["REQ-069-023"])
def _name_matches(candidate: str, pattern: str, pattern_type: PatternType) -> bool:
    """Whether ``candidate`` matches ``pattern`` under ``pattern_type`` (FR-002).

    ``prefix``/``suffix`` tolerate a trailing/leading ``*`` in ``pattern``
    (spec.md's own "Batch Stop by Pattern" example passes ``name="test_*"``
    together with ``pattern="prefix"`` — a shell-glob habit layered on top of
    the explicit pattern-type selector) by stripping it before comparing.
    """
    if pattern_type == PatternType.EXACT:
        return candidate == pattern
    if pattern_type == PatternType.PREFIX:
        return candidate.startswith(pattern[:-1] if pattern.endswith("*") else pattern)
    if pattern_type == PatternType.SUFFIX:
        return candidate.endswith(pattern[1:] if pattern.startswith("*") else pattern)
    if pattern_type == PatternType.REGEX:
        try:
            return re.search(pattern, candidate) is not None
        except re.error:
            return False
    return False


# ── public schema dataclasses (spec.md Data Models) ─────────────────────


@traces(spec="069", story="US-069-002", reqs=["REQ-069-032"])
@dataclass(slots=True, frozen=True)
class TaskStatusResult:
    """Mirrors spec.md's ``TaskStatusResult`` JSON Schema (FR-011)."""

    task_id: str
    name: str
    status: str
    started_at: str
    output_lines: int
    stopped_at: str | None = None
    progress: dict[str, Any] | None = None
    pid: int | None = None
    exit_code: int | None = None
    memory_mb: float | None = None
    cpu_percent: float | None = None
    agent_id: str = ""
    task_type: str = "custom"


@traces(spec="069", story="US-069-002", reqs=["REQ-069-028", "REQ-069-033"])
@dataclass(slots=True, frozen=True)
class TaskOutput:
    """Mirrors spec.md's ``TaskOutput`` JSON Schema (FR-007/FR-012)."""

    task_id: str
    stdout: str
    stderr: str
    combined: str
    timestamp: str
    line_count: int
    byte_count: int
    tail_lines: int | None = None
    line_range: dict[str, int] | None = None


@traces(spec="069", story="US-069-001", reqs=["REQ-069-022"])
@dataclass(slots=True, frozen=True)
class StopOutcome:
    """Result of a single :meth:`TaskManager.stop` call (TaskStopByIdResponse.data)."""

    task_id: str
    stop_mode: str
    signal_sent: str | None
    wait_time_ms: int
    exit_code: int | None


@traces(spec="069", story="US-069-003", reqs=["REQ-069-027"])
@dataclass(slots=True, frozen=True)
class BatchStopOutcome:
    """Result of :meth:`TaskManager.stop_by_pattern` (TaskStopByNameResponse.data)."""

    name_pattern: str
    tasks_found: int
    tasks_stopped: int
    failures: int
    task_ids: tuple[str, ...]


# ── internal records ─────────────────────────────────────────────────────


@dataclass(slots=True)
class _OutputLine:
    seq: int
    stream: str
    text: str
    ts: datetime


@traces(spec="069", story="US-069-001", reqs=["REQ-069-016", "REQ-069-017"])
@dataclass(slots=True)
class _TaskRecord:
    task_id: str
    name: str
    status: TaskStatus
    agent_id: str
    task_type: TaskType
    started_at: datetime
    stopped_at: datetime | None = None
    pid: int | None = None
    process: subprocess.Popen[bytes] | None = None
    exit_code: int | None = None
    protected: bool = False
    critical: bool = False
    progress_current: int = 0
    progress_total: int = 0
    progress_message: str | None = None
    memory_mb_override: float | None = None
    cpu_percent_override: float | None = None
    lines: list[_OutputLine] = field(default_factory=list)
    version: int = 0
    #: True while a stop() call is in flight for this record — the
    #: idempotency guard for SEC-REQ-016/017 (a concurrent second stop call
    #: sees this and fails cleanly instead of double-signalling).
    stopping: bool = False


# ── the manager (in-process backing store for create/get/list/update/stop/output) ──


class TaskManager:
    """Thread-safe in-process task store backing create/get/list/update/stop/output.

    See the module docstring's fork note: this replaces the Redis-cluster
    store spec.md assumes. All mutation goes through a single
    :class:`threading.RLock` (``self._lock``); process signalling/waiting
    (the only genuinely slow step) is deliberately performed OUTSIDE the
    lock so concurrent reads/other-task stops are never blocked by one
    task's graceful-timeout wait.
    """

    def __init__(
        self,
        stop_config: TaskStopConfig | None = None,
        output_config: TaskOutputConfig | None = None,
        id_factory: Callable[[], str] | None = None,
    ) -> None:
        self._tasks: dict[str, _TaskRecord] = {}
        self._lock = threading.RLock()
        self._config_stop = stop_config or TaskStopConfig()
        self._config_output = output_config or TaskOutputConfig()
        self._id_factory = id_factory or self._default_id_factory
        self._audit_log: list[dict[str, Any]] = []
        self._rate_buckets: dict[str, deque[float]] = {}

    @staticmethod
    @traces(spec="069", reqs=["REQ-069-014"])
    def _default_id_factory() -> str:
        """Cryptographically random task ID (SEC-REQ-014 / SEC-001)."""
        return f"task_{secrets.token_urlsafe(16)}"

    # ── audit (SEC-REQ-018/019/020/021) ─────────────────────────────

    @traces(spec="069", reqs=["REQ-069-018", "REQ-069-019", "REQ-069-020"])
    def _audit(
        self, action: str, task_id: str, agent_id: str, success: bool, reason: str
    ) -> None:
        """Append an immutable audit entry (RLock-safe, callable while locked)."""
        entry = {
            "action": action,
            "task_id": task_id,
            "agent_id": agent_id,
            "success": success,
            "reason": reason,
            "timestamp": datetime.now(UTC).isoformat(),
        }
        with self._lock:
            self._audit_log.append(entry)

    def audit_log(self) -> tuple[dict[str, Any], ...]:
        """A read-only snapshot of the audit trail (SEC-REQ-021)."""
        with self._lock:
            return tuple(self._audit_log)

    # ── create / get / list / update (the base tool-suite operations) ──

    @traces(
        spec="069",
        story="US-069-001",
        reqs=["REQ-069-014", "REQ-069-015", "REQ-069-044"],
    )
    def create(
        self,
        name: str,
        agent_id: str,
        task_type: TaskType | str = TaskType.CUSTOM,
        *,
        protected: bool = False,
        critical: bool = False,
        process: subprocess.Popen[bytes] | None = None,
        pid: int | None = None,
        status: TaskStatus = TaskStatus.RUNNING,
    ) -> _TaskRecord:
        """Create and register a new task; returns the live record.

        Generates a collision-free ID via ``id_factory`` under the manager
        lock (SEC-REQ-015 atomic creation); raises ``TASK_ID_COLLISION``
        (ERR-011) if :data:`_MAX_ID_ATTEMPTS` consecutive attempts all
        collide with an already-registered ID.
        """
        _validate_name(name)
        with self._lock:
            task_id = ""
            for _ in range(_MAX_ID_ATTEMPTS):
                candidate = self._id_factory()
                if _TASK_ID_PATTERN.match(candidate) and candidate not in self._tasks:
                    task_id = candidate
                    break
            else:
                raise TaskToolError(
                    TaskToolErrorCode.TASK_ID_COLLISION,
                    "Unable to generate a unique task ID after "
                    f"{_MAX_ID_ATTEMPTS} attempts",
                )
            record = _TaskRecord(
                task_id=task_id,
                name=_sanitize_string(name),
                status=status,
                agent_id=agent_id,
                task_type=TaskType(task_type),
                started_at=datetime.now(UTC),
                pid=pid if pid is not None else (process.pid if process else None),
                process=process,
                protected=protected,
                critical=critical,
            )
            self._tasks[task_id] = record
            return record

    def get(self, task_id: str) -> _TaskRecord | None:
        """Return the task record for ``task_id``, or ``None``."""
        with self._lock:
            return self._tasks.get(task_id)

    def list(
        self,
        name: str | None = None,
        pattern_type: PatternType = PatternType.EXACT,
        status: TaskStatus | None = None,
    ) -> list[_TaskRecord]:
        """List tasks, optionally filtered by name pattern and/or status."""
        with self._lock:
            records = list(self._tasks.values())
        if name is not None:
            records = [r for r in records if _name_matches(r.name, name, pattern_type)]
        if status is not None:
            records = [r for r in records if r.status == status]
        return records

    def update(
        self,
        task_id: str,
        *,
        status: TaskStatus | None = None,
        progress_current: int | None = None,
        progress_total: int | None = None,
        progress_message: str | None = None,
        memory_mb: float | None = None,
        cpu_percent: float | None = None,
    ) -> _TaskRecord:
        """Mutate a task's status/progress/reported metrics under the lock."""
        with self._lock:
            record = self._tasks.get(task_id)
            if record is None:
                raise TaskToolError(
                    TaskToolErrorCode.TASK_NOT_FOUND, f"Task {task_id} not found"
                )
            if status is not None:
                record.status = status
            if progress_current is not None:
                record.progress_current = progress_current
            if progress_total is not None:
                record.progress_total = progress_total
            if progress_message is not None:
                record.progress_message = progress_message
            if memory_mb is not None:
                record.memory_mb_override = memory_mb
            if cpu_percent is not None:
                record.cpu_percent_override = cpu_percent
            record.version += 1
            return record

    def append_output(self, task_id: str, stream: str, text: str) -> None:
        """Append one output line (``stream`` is ``"stdout"`` or ``"stderr"``)."""
        with self._lock:
            record = self._tasks.get(task_id)
            if record is None:
                raise TaskToolError(
                    TaskToolErrorCode.TASK_NOT_FOUND, f"Task {task_id} not found"
                )
            seq = len(record.lines)
            record.lines.append(
                _OutputLine(seq=seq, stream=stream, text=text, ts=datetime.now(UTC))
            )

    # ── stop (FR-001..006) ──────────────────────────────────────────

    @traces(
        spec="069",
        story="US-069-001",
        reqs=[
            "REQ-069-001",
            "REQ-069-002",
            "REQ-069-003",
            "REQ-069-004",
            "REQ-069-005",
            "REQ-069-016",
            "REQ-069-017",
            "REQ-069-018",
            "REQ-069-019",
            "REQ-069-020",
            "REQ-069-022",
            "REQ-069-024",
            "REQ-069-025",
            "REQ-069-026",
            "REQ-069-034",
            "REQ-069-035",
            "REQ-069-036",
            "REQ-069-037",
            "REQ-069-038",
            "REQ-069-039",
            "REQ-069-048",
        ],
    )
    def stop(
        self,
        task_id: str,
        mode: TaskStopMode,
        requesting_agent_id: str,
        *,
        is_admin: bool = False,
    ) -> StopOutcome:
        """Stop a single task (FR-001/003/004/005).

        Raises:
            TaskToolError: ``TASK_NOT_FOUND`` (ERR-001), ``TASK_NOT_RUNNING``
                (ERR-002 — also covers the idempotent-retry / terminal-state
                case, SEC-REQ-017), ``PERMISSION_DENIED`` (ERR-003,
                SEC-REQ-002/013), ``CRITICAL_TASK`` (ERR-005, SEC-REQ-005),
                ``PROTECTED_TASK`` (ERR-006, SEC-REQ-004), or
                ``TERMINATION_FAILED`` (ERR-004) if the process still hasn't
                exited after the configured SIGKILL retries.
        """
        with self._lock:
            record = self._tasks.get(task_id)
            if record is None:
                self._audit("stop", task_id, requesting_agent_id, False, "not found")
                raise TaskToolError(
                    TaskToolErrorCode.TASK_NOT_FOUND, f"Task {task_id} not found"
                )
            not_cancelable = record.status in _TERMINAL_STATUSES
            not_running = (
                mode != TaskStopMode.CANCEL and record.status != TaskStatus.RUNNING
            )
            if record.stopping or not_cancelable or not_running:
                self._audit("stop", task_id, requesting_agent_id, False, "not running")
                raise TaskToolError(
                    TaskToolErrorCode.TASK_NOT_RUNNING,
                    f"Task {task_id} is not in running state",
                )
            if not is_admin and requesting_agent_id != record.agent_id:
                self._audit(
                    "stop", task_id, requesting_agent_id, False, "permission denied"
                )
                raise TaskToolError(
                    TaskToolErrorCode.PERMISSION_DENIED,
                    f"Agent does not have permission to stop task {task_id}",
                )
            if record.critical and not is_admin:
                self._audit(
                    "stop", task_id, requesting_agent_id, False, "critical task"
                )
                raise TaskToolError(
                    TaskToolErrorCode.CRITICAL_TASK,
                    f"Cannot stop critical task {task_id}",
                )
            if record.protected and not is_admin:
                self._audit(
                    "stop", task_id, requesting_agent_id, False, "protected task"
                )
                raise TaskToolError(
                    TaskToolErrorCode.PROTECTED_TASK,
                    f"Task {task_id} is protected from termination",
                )
            record.stopping = True
            process = record.process

        start = time.monotonic()
        signal_sent: str | None = None
        exit_code: int | None = None
        try:
            if process is not None:
                signal_sent, exit_code = self._signal_and_wait(process, mode)
        except TaskToolError:
            with self._lock:
                record.stopping = False
            self._audit(
                "stop", task_id, requesting_agent_id, False, "termination failed"
            )
            raise
        wait_ms = int((time.monotonic() - start) * 1000)
        if exit_code is not None and exit_code < 0:
            exit_code = 128 - exit_code

        with self._lock:
            record.status = TaskStatus.STOPPED
            record.stopped_at = datetime.now(UTC)
            record.exit_code = exit_code
            record.stopping = False
            record.version += 1
        self._audit("stop", task_id, requesting_agent_id, True, f"mode={mode.value}")
        return StopOutcome(
            task_id=task_id,
            stop_mode=mode.value,
            signal_sent=signal_sent,
            wait_time_ms=wait_ms,
            exit_code=exit_code,
        )

    @traces(
        spec="069",
        story="US-069-001",
        reqs=["REQ-069-024", "REQ-069-025", "REQ-069-037"],
    )
    def _signal_and_wait(
        self, process: subprocess.Popen[bytes], mode: TaskStopMode
    ) -> tuple[str | None, int | None]:
        """Send the signal ``mode`` implies and wait for the process to exit.

        GRACEFUL (FR-003, the default): SIGTERM, wait up to
        ``graceful_timeout``, escalate to SIGKILL if still alive. FORCE
        (FR-004): SIGKILL immediately. CANCEL: no signal if the process
        already exited; otherwise the same SIGKILL path as FORCE.
        """
        if mode == TaskStopMode.FORCE:
            process.kill()
            return "SIGKILL", self._await_exit(process)
        if mode == TaskStopMode.CANCEL:
            if process.poll() is not None:
                return None, process.returncode
            process.kill()
            return "SIGKILL", self._await_exit(process)
        process.terminate()
        try:
            exit_code = process.wait(timeout=self._config_stop.graceful_timeout)
            return "SIGTERM", exit_code
        except subprocess.TimeoutExpired:
            process.kill()
            return "SIGTERM+SIGKILL", self._await_exit(process)

    @traces(spec="069", reqs=["REQ-069-037"])
    def _await_exit(self, process: subprocess.Popen[bytes]) -> int:
        """Wait for ``process`` to exit, retrying per ``TaskStopConfig``.

        Raises ``TERMINATION_FAILED`` (ERR-004) if the process is still
        alive after every retry — reachable deterministically with
        ``TaskStopConfig(retry_attempts=0)``, which skips waiting entirely
        and checks ``poll()`` immediately after the signal is sent.
        """
        for _ in range(max(self._config_stop.retry_attempts, 0)):
            try:
                return process.wait(timeout=self._config_stop.retry_delay)
            except subprocess.TimeoutExpired:
                continue
        if process.poll() is None:
            raise TaskToolError(
                TaskToolErrorCode.TERMINATION_FAILED,
                f"Failed to terminate process PID {process.pid}",
            )
        return process.returncode

    @traces(spec="069", story="US-069-003", reqs=["REQ-069-023", "REQ-069-027"])
    def stop_by_pattern(
        self,
        name: str,
        pattern_type: PatternType,
        force: bool,
        requesting_agent_id: str,
        *,
        is_admin: bool = False,
    ) -> BatchStopOutcome:
        """Stop every running task whose name matches ``name`` (FR-002/006).

        Protected/critical tasks are excluded from ``tasks_found`` unless
        ``is_admin`` (US-003 acceptance criteria); a per-task failure during
        stop is counted in ``failures`` rather than aborting the batch.
        """
        mode = TaskStopMode.FORCE if force else TaskStopMode.GRACEFUL
        with self._lock:
            candidates = [
                r
                for r in self._tasks.values()
                if r.status == TaskStatus.RUNNING
                and _name_matches(r.name, name, pattern_type)
            ]
        eligible = [
            r
            for r in candidates
            if (is_admin or not r.protected) and (is_admin or not r.critical)
        ]
        stopped_ids: list[str] = []
        failures = 0
        for record in eligible:
            try:
                self.stop(record.task_id, mode, requesting_agent_id, is_admin=is_admin)
                stopped_ids.append(record.task_id)
            except TaskToolError:
                failures += 1
        return BatchStopOutcome(
            name_pattern=name,
            tasks_found=len(eligible),
            tasks_stopped=len(stopped_ids),
            failures=failures,
            task_ids=tuple(stopped_ids),
        )

    # ── output / status (FR-007..012) ───────────────────────────────

    @traces(spec="069", story="US-069-004", reqs=["REQ-069-046"])
    def check_rate_limit(self, agent_id: str) -> None:
        """Raise ``RATE_LIMIT_EXCEEDED`` (ERR-013) past the per-agent quota.

        A process-local sliding window (SEC-REQ isn't distributed — see the
        module docstring's fork note: production would need this backed by
        Redis for multi-process fairness).
        """
        now = time.monotonic()
        with self._lock:
            bucket = self._rate_buckets.setdefault(agent_id, deque())
            while bucket and now - bucket[0] > 60.0:
                bucket.popleft()
            if len(bucket) >= self._config_output.rate_limit_per_minute:
                raise TaskToolError(
                    TaskToolErrorCode.RATE_LIMIT_EXCEEDED,
                    "Too many output requests "
                    f"({self._config_output.rate_limit_per_minute}/minute)",
                )
            bucket.append(now)

    @traces(
        spec="069",
        story="US-069-002",
        reqs=[
            "REQ-069-028",
            "REQ-069-030",
            "REQ-069-031",
            "REQ-069-033",
            "REQ-069-040",
            "REQ-069-041",
            "REQ-069-042",
        ],
    )
    def output_of(
        self,
        task_id: str,
        mode: OutputMode,
        *,
        tail: int | None = None,
        lines: tuple[int, int] | None = None,
        stream_position: int = 0,
        combined_only: bool = False,
        max_output_bytes: int | None = None,
    ) -> tuple[TaskOutput, int | None]:
        """Retrieve windowed output for ``task_id`` (FR-007/009/010/012).

        Returns ``(output, new_stream_position)`` — ``new_stream_position``
        is only meaningful for ``mode=OutputMode.STREAM`` (``None`` otherwise).

        Raises:
            TaskToolError: ``TASK_NOT_FOUND`` (ERR-001), ``OUTPUT_NOT_AVAILABLE``
                (ERR-007 — no output captured yet), ``INVALID_LINE_RANGE``
                (ERR-008), or ``OUTPUT_TOO_LARGE`` (ERR-009, full mode only).
        """
        with self._lock:
            record = self._tasks.get(task_id)
            if record is None:
                raise TaskToolError(
                    TaskToolErrorCode.TASK_NOT_FOUND, f"Task {task_id} not found"
                )
            all_lines = list(record.lines)

        total = len(all_lines)
        if total == 0:
            raise TaskToolError(
                TaskToolErrorCode.OUTPUT_NOT_AVAILABLE,
                f"Output not found for task {task_id}",
            )
        total_bytes = sum(len(line.text.encode()) for line in all_lines)
        new_position: int | None = None

        if mode == OutputMode.FULL:
            limit = (
                max_output_bytes
                if max_output_bytes is not None
                else self._config_output.max_output_bytes
            )
            if total_bytes > limit:
                raise TaskToolError(
                    TaskToolErrorCode.OUTPUT_TOO_LARGE,
                    f"Output size {total_bytes} exceeds limit {limit}",
                )
            window = all_lines
        elif mode == OutputMode.TAIL:
            n = tail if tail is not None else self._config_output.tail_lines_default
            if n < 1:
                raise TaskToolError(
                    TaskToolErrorCode.INVALID_INPUT, "tail must be >= 1"
                )
            window = all_lines[-n:]
        elif mode == OutputMode.LINES:
            if lines is None or len(lines) != 2:
                raise TaskToolError(
                    TaskToolErrorCode.INVALID_INPUT, "lines must be a (start, end) pair"
                )
            start, end = lines
            if start < 1 or end < start or end > total:
                raise TaskToolError(
                    TaskToolErrorCode.INVALID_LINE_RANGE,
                    f"Line range ({start}, {end}) exceeds output length ({total})",
                )
            window = all_lines[start - 1 : end]
        else:  # STREAM
            pos = max(stream_position, 0)
            window = all_lines[pos:]
            new_position = total

        stdout_text = "\n".join(line.text for line in window if line.stream == "stdout")
        stderr_text = "\n".join(line.text for line in window if line.stream == "stderr")
        combined_text = "\n".join(line.text for line in window)
        if combined_only:
            stdout_text = ""
            stderr_text = ""

        output = TaskOutput(
            task_id=task_id,
            stdout=stdout_text,
            stderr=stderr_text,
            combined=combined_text,
            timestamp=datetime.now(UTC).isoformat(),
            line_count=total,
            byte_count=total_bytes,
            tail_lines=(
                (tail or self._config_output.tail_lines_default)
                if mode == OutputMode.TAIL
                else None
            ),
            line_range=(
                {"start": lines[0], "end": lines[1]}
                if lines and mode == OutputMode.LINES
                else None
            ),
        )
        return output, new_position

    @traces(spec="069", story="US-069-002", reqs=["REQ-069-032"])
    def status_of(self, task_id: str) -> TaskStatusResult:
        """Comprehensive status + performance metrics for ``task_id`` (FR-011).

        ``memory_mb``/``cpu_percent`` are sampled live via ``psutil`` when a
        real process is attached and still running; otherwise they fall back
        to any value reported through :meth:`update`, or ``None``.
        """
        with self._lock:
            record = self._tasks.get(task_id)
            if record is None:
                raise TaskToolError(
                    TaskToolErrorCode.TASK_NOT_FOUND, f"Task {task_id} not found"
                )
            memory_mb = record.memory_mb_override
            cpu_percent = record.cpu_percent_override
            pid = record.pid
            if record.process is not None and record.process.poll() is None:
                import psutil

                try:
                    proc = psutil.Process(record.process.pid)
                    memory_mb = proc.memory_info().rss / (1024 * 1024)
                    cpu_percent = proc.cpu_percent(interval=0.0)
                except psutil.Error:
                    pass
            elif record.status != TaskStatus.RUNNING:
                pid = None
            progress = None
            if record.progress_total > 0:
                percent = (record.progress_current / record.progress_total) * 100
                progress = {
                    "current": record.progress_current,
                    "total": record.progress_total,
                    "percent": max(0.0, min(100.0, percent)),
                    "message": record.progress_message,
                }
            return TaskStatusResult(
                task_id=record.task_id,
                name=record.name,
                status=record.status.value,
                started_at=record.started_at.isoformat(),
                stopped_at=record.stopped_at.isoformat() if record.stopped_at else None,
                progress=progress,
                pid=pid,
                exit_code=record.exit_code,
                output_lines=len(record.lines),
                memory_mb=memory_mb,
                cpu_percent=cpu_percent,
                agent_id=record.agent_id,
                task_type=record.task_type.value,
            )

    @traces(spec="069", story="US-069-002", reqs=["REQ-069-012"])
    def purge_expired_output(self, retention: timedelta | None = None) -> int:
        """Drop captured output older than ``retention`` (default 7 days).

        SEC-REQ-012 without a Redis TTL: purge is triggered explicitly
        (there is no background scheduler in this MVP) but the retention
        policy itself — output MUST NOT outlive the configured window — is
        real. Returns the number of lines purged.
        """
        cutoff = datetime.now(UTC) - (
            retention or timedelta(days=self._config_output.retention_days)
        )
        purged = 0
        with self._lock:
            for record in self._tasks.values():
                if record.started_at < cutoff and record.lines:
                    purged += len(record.lines)
                    record.lines.clear()
        return purged


# ── TaskStopTool ──────────────────────────────────────────────────────────

_TASK_STOP_PARAMETERS: dict[str, Any] = {
    "type": "object",
    "properties": {
        "task_id": {
            "type": "string",
            "description": "Task ID to stop, matching ^task_[a-zA-Z0-9_-]+$.",
        },
        "name": {
            "type": "string",
            "description": "Name (or pattern) to batch-stop matching tasks by.",
        },
        "pattern": {
            "type": "string",
            "enum": ["exact", "regex", "prefix", "suffix"],
            "description": "How 'name' is matched (default: exact).",
        },
        "force": {
            "type": "boolean",
            "description": "SIGKILL immediately instead of graceful SIGTERM.",
        },
        "mode": {
            "type": "string",
            "enum": ["graceful", "force", "cancel"],
            "description": "Explicit stop mode; overrides 'force' when given.",
        },
        "requesting_agent_id": {
            "type": "string",
            "description": "Agent performing the stop (SEC-REQ-001/002).",
        },
        "is_admin": {
            "type": "boolean",
            "description": "Whether the requesting agent has admin privileges.",
        },
    },
    "required": ["requesting_agent_id"],
}


@traces(spec="069", story="US-069-001", reqs=["REQ-069-022", "REQ-069-023"])
class TaskStopTool(BaseTool):
    """The registrable ``BaseTool`` for stopping tasks (FR-001..006)."""

    def __init__(
        self,
        manager: TaskManager,
        metadata: ToolMetadata,
        tracer: Tracer | None = None,
    ) -> None:
        self._manager = manager
        self._metadata = metadata
        self._tracer = tracer if tracer is not None else Tracer.default()

    @property
    def metadata(self) -> ToolMetadata:
        return self._metadata

    @traces(spec="069", story="US-069-001", reqs=["REQ-069-022", "REQ-069-023"])
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        """Stop by ``task_id`` or batch-stop by ``name`` pattern.

        Never raises: any :class:`TaskToolError` (or unforeseen exception)
        becomes a failed :class:`ToolResult`.
        """
        with self._tracer.span("TaskStopTool.execute"):
            try:
                requesting_agent_id = arguments.get("requesting_agent_id")
                if (
                    not isinstance(requesting_agent_id, str)
                    or not requesting_agent_id.strip()
                ):
                    raise TaskToolError(
                        TaskToolErrorCode.INVALID_INPUT,
                        "requesting_agent_id is required",
                    )
                is_admin = bool(arguments.get("is_admin", False))
                force = bool(arguments.get("force", False))
                mode_raw = arguments.get("mode")
                mode = (
                    TaskStopMode(mode_raw)
                    if mode_raw is not None
                    else (TaskStopMode.FORCE if force else TaskStopMode.GRACEFUL)
                )

                task_id = arguments.get("task_id")
                name = arguments.get("name")
                if task_id:
                    _validate_task_id(task_id)
                    outcome = self._manager.stop(
                        task_id, mode, requesting_agent_id, is_admin=is_admin
                    )
                    data = asdict(outcome)
                    message = (
                        f"Task {task_id} cancelled"
                        if outcome.stop_mode == TaskStopMode.CANCEL.value
                        else f"Task {task_id} stopped {outcome.stop_mode}ly"
                    )
                elif name:
                    pattern_type = PatternType(arguments.get("pattern", "exact"))
                    batch = self._manager.stop_by_pattern(
                        name,
                        pattern_type,
                        force,
                        requesting_agent_id,
                        is_admin=is_admin,
                    )
                    data = {
                        "name_pattern": batch.name_pattern,
                        "tasks_found": batch.tasks_found,
                        "tasks_stopped": batch.tasks_stopped,
                        "failures": batch.failures,
                        "task_ids": list(batch.task_ids),
                    }
                    message = f"Stopped {batch.tasks_stopped} tasks"
                else:
                    raise TaskToolError(
                        TaskToolErrorCode.INVALID_INPUT,
                        "either task_id or name is required",
                    )

                payload = {
                    "success": True,
                    "message": message,
                    "data": data,
                    "timestamp": datetime.now(UTC).isoformat(),
                }
                return ToolResult(
                    tool_name=self.metadata.name,
                    success=True,
                    output=json.dumps(payload),
                )
            except TaskToolError as exc:
                return exc.to_result(self.metadata.name)
            except Exception as exc:  # noqa: BLE001 - never propagate
                return ToolResult(
                    tool_name=self.metadata.name,
                    success=False,
                    output="",
                    error=str(exc),
                )

    @classmethod
    def create(
        cls, manager: TaskManager, trace_config: TraceConfig | None = None
    ) -> TaskStopTool:
        """Build a :class:`TaskStopTool` over the given :class:`TaskManager`."""
        tracer = Tracer(trace_config) if trace_config is not None else Tracer.default()
        metadata = ToolMetadata(
            name="task_stop",
            description=(
                "Stop a running task by ID or batch-stop tasks matching a name "
                "pattern, gracefully (SIGTERM->SIGKILL) or forced (SIGKILL)."
            ),
            category=ToolCategory.TASK,
            parameters=_TASK_STOP_PARAMETERS,
            priority=ToolPriority.P2,
            tags=("task", "process", "lifecycle"),
        )
        return cls(manager=manager, metadata=metadata, tracer=tracer)


# ── TaskOutputTool ────────────────────────────────────────────────────────

_TASK_OUTPUT_PARAMETERS: dict[str, Any] = {
    "type": "object",
    "properties": {
        "task_id": {"type": "string", "description": "Task ID to read."},
        "request_type": {
            "type": "string",
            "enum": ["output", "status"],
            "description": "'output' (default) retrieves stdout/stderr; 'status' returns metrics.",
        },
        "mode": {
            "type": "string",
            "enum": ["full", "tail", "lines", "stream"],
            "description": "Output windowing mode (request_type='output' only).",
        },
        "tail": {"type": "integer", "description": "Line count for mode='tail'."},
        "lines": {
            "type": "array",
            "items": {"type": "integer"},
            "minItems": 2,
            "maxItems": 2,
            "description": "[start, end] 1-indexed inclusive range for mode='lines'.",
        },
        "combined": {
            "type": "boolean",
            "description": "Only populate the merged 'combined' field, not stdout/stderr.",
        },
        "stream_position": {
            "type": "integer",
            "description": "Cursor for mode='stream' incremental polling.",
        },
        "requesting_agent_id": {
            "type": "string",
            "description": "Agent making the request (used for rate limiting).",
        },
    },
    "required": ["task_id"],
}


@traces(
    spec="069",
    story="US-069-002",
    reqs=["REQ-069-028", "REQ-069-030", "REQ-069-031", "REQ-069-032", "REQ-069-033"],
)
class TaskOutputTool(BaseTool):
    """The registrable ``BaseTool`` for output retrieval + status (FR-007..012).

    ``mode="stream"`` (FR-008) implements the incremental-chunk primitive
    (poll with a ``stream_position`` cursor) but NOT a live WebSocket push
    transport — see the module docstring's fork note; no gateway wiring was
    built as part of this spec.
    """

    def __init__(
        self,
        manager: TaskManager,
        metadata: ToolMetadata,
        tracer: Tracer | None = None,
    ) -> None:
        self._manager = manager
        self._metadata = metadata
        self._tracer = tracer if tracer is not None else Tracer.default()

    @property
    def metadata(self) -> ToolMetadata:
        return self._metadata

    @traces(
        spec="069",
        story="US-069-002",
        reqs=[
            "REQ-069-028",
            "REQ-069-030",
            "REQ-069-031",
            "REQ-069-032",
            "REQ-069-033",
        ],
    )
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        """Retrieve output (full/tail/lines/stream) or status for a task.

        Never raises: any :class:`TaskToolError` (or unforeseen exception)
        becomes a failed :class:`ToolResult`.
        """
        with self._tracer.span("TaskOutputTool.execute"):
            try:
                task_id = arguments.get("task_id")
                if not isinstance(task_id, str) or not task_id.strip():
                    raise TaskToolError(
                        TaskToolErrorCode.INVALID_INPUT, "task_id is required"
                    )
                _validate_task_id(task_id)
                agent_id = arguments.get("requesting_agent_id") or "anonymous"
                self._manager.check_rate_limit(agent_id)

                request_type = arguments.get("request_type", "output")
                if request_type == "status":
                    result = self._manager.status_of(task_id)
                    payload = {
                        "success": True,
                        "message": f"Status for {task_id}",
                        "data": asdict(result),
                        "timestamp": datetime.now(UTC).isoformat(),
                    }
                    return ToolResult(
                        tool_name=self.metadata.name,
                        success=True,
                        output=json.dumps(payload),
                    )

                mode = OutputMode(arguments.get("mode", "full"))
                tail_raw = arguments.get("tail")
                tail = int(tail_raw) if tail_raw is not None else None
                lines_raw = arguments.get("lines")
                lines = (int(lines_raw[0]), int(lines_raw[1])) if lines_raw else None
                combined_only = bool(arguments.get("combined", False))
                stream_position = int(arguments.get("stream_position", 0))

                output, new_position = self._manager.output_of(
                    task_id,
                    mode,
                    tail=tail,
                    lines=lines,
                    stream_position=stream_position,
                    combined_only=combined_only,
                )
                data = asdict(output)
                if new_position is not None:
                    data["stream_position"] = new_position
                payload = {
                    "success": True,
                    "message": f"Retrieved {mode.value} output for {task_id}",
                    "data": data,
                    "timestamp": datetime.now(UTC).isoformat(),
                }
                return ToolResult(
                    tool_name=self.metadata.name,
                    success=True,
                    output=json.dumps(payload),
                )
            except TaskToolError as exc:
                return exc.to_result(self.metadata.name)
            except (TypeError, ValueError) as exc:
                return ToolResult(
                    tool_name=self.metadata.name,
                    success=False,
                    output="",
                    error=f"{TaskToolErrorCode.INVALID_INPUT}: {exc}",
                )
            except Exception as exc:  # noqa: BLE001 - never propagate
                return ToolResult(
                    tool_name=self.metadata.name,
                    success=False,
                    output="",
                    error=str(exc),
                )

    @classmethod
    def create(
        cls, manager: TaskManager, trace_config: TraceConfig | None = None
    ) -> TaskOutputTool:
        """Build a :class:`TaskOutputTool` over the given :class:`TaskManager`."""
        tracer = Tracer(trace_config) if trace_config is not None else Tracer.default()
        metadata = ToolMetadata(
            name="task_output",
            description=(
                "Retrieve a task's captured stdout/stderr (full, tail, line "
                "range, or incremental stream) or its status + performance metrics."
            ),
            category=ToolCategory.TASK,
            parameters=_TASK_OUTPUT_PARAMETERS,
            priority=ToolPriority.P2,
            tags=("task", "process", "output"),
        )
        return cls(manager=manager, metadata=metadata, tracer=tracer)


# ── wiring ────────────────────────────────────────────────────────────────


def register_task_tools(
    registry: ToolRegistry,
    manager: TaskManager | None = None,
    trace_config: TraceConfig | None = None,
) -> tuple[TaskStopTool, TaskOutputTool]:
    """Build :class:`TaskStopTool` + :class:`TaskOutputTool` sharing one
    :class:`TaskManager`, and register both into ``registry``.
    """
    manager = manager if manager is not None else TaskManager()
    stop_tool = TaskStopTool.create(manager, trace_config)
    output_tool = TaskOutputTool.create(manager, trace_config)
    registry.register(stop_tool, replace=True)
    registry.register(output_tool, replace=True)
    return stop_tool, output_tool