Function _launch_async_agent

src/tool_system/tools/agent.py:432–651

Launch an agent in the background and return immediately. Chapter-10 layered story: * Chunk B / WI-1.5 — state on ``context.runtime_tasks`` as a typed ``LocalAgentTaskState`` (no more ``context.tasks`` / ``metadata._internal=True`` workaround). * Chunk C

(
        *,
        run_params: RunAgentParams,
        context: ToolContext,
        agent_id: str,
        description: str,
        prompt: str,
        agent_type: str,
        agent_name: str | None = None,
    )

        )

    def _launch_async_agent(
        *,
        run_params: RunAgentParams,
        context: ToolContext,
        agent_id: str,
        description: str,
        prompt: str,
        agent_type: str,
        agent_name: str | None = None,
    ) -> ToolResult:
        """Launch an agent in the background and return immediately.

        Chapter-10 layered story:
        * Chunk B / WI-1.5 — state on ``context.runtime_tasks`` as a
          typed ``LocalAgentTaskState`` (no more ``context.tasks`` /
          ``metadata._internal=True`` workaround).
        * Chunk C / WI-2.2 (gate-zero) — sidechain JSONL transcript
          opened for the lifetime of the agent run; ``output_file`` is
          its absolute path.
        * Chunk C / WI-2.3 — lifecycle goes through the named helpers
          (``register_async_agent`` / ``complete_agent_task`` /
          ``fail_agent_task``) so the registry mutations are atomic
          and consistent across spawn / kill / completion paths.
        * Chunk C / WI-2.4 — token accounting via ``ProgressTracker``;
          ``finalize_agent_tool`` reads its accumulated totals instead
          of reporting ``total_tokens=0``.
        * Chunk F / WI-6.1 — optional ``agent_name`` registers the
          spawn under ``context.agent_name_registry`` so SendMessage
          can resolve ``to: <name>``. Collision-on-running raises;
          collision-on-terminal silently overwrites.
        """
        # Local imports defer the cycle: ``src.tasks.local_agent``
        # reaches back into ``src.task_registry`` which is fine, but
        # importing them at module scope would tangle with
        # ``defaults.py``'s tool-construction order.
        from src.agent.transcript import TranscriptWriter
        from src.tasks.local_agent import (
            LocalAgentTaskState,
            complete_agent_task,
            fail_agent_task,
            register_async_agent,
        )
        from src.tasks.progress import (
            ProgressTracker,
            update_progress_from_message,
        )
        from src.services.swarm.agent_name_registry import (
            AgentNameAlreadyClaimedError,
        )
        from src.utils.task_notification import enqueue_agent_notification

        # WI-6.1 + critic C1 (Phase-7 fix): atomic check-and-claim
        # under the typed registry's RLock. The previous Phase-6
        # implementation had a TOCTOU window between the read and
        # the write; the typed ``claim_or_raise`` closes it. We do
        # the claim BEFORE the runtime_tasks write so a refused
        # spawn doesn't leak a half-constructed agent_id into the
        # runtime registry.