Function register_async_agent

src/tasks/local_agent.py:110–160

Register a brand-new background agent on the runtime registry. Idempotent enough to be safe under spawn-then-resume races; if an entry already exists at ``agent_id`` it is replaced (the resume path explicitly wants this so the new event-loop wiring takes over). Returns the just

(
    *,
    agent_id: str,
    description: str,
    prompt: str,
    agent_type: str,
    selected_agent: Any = None,
    model: str | None = None,
    tool_use_id: str | None = None,
    registry: "RuntimeTaskRegistry",
)

def register_async_agent(
    *,
    agent_id: str,
    description: str,
    prompt: str,
    agent_type: str,
    selected_agent: Any = None,
    model: str | None = None,
    tool_use_id: str | None = None,
    registry: "RuntimeTaskRegistry",
) -> LocalAgentTaskState:
    """Register a brand-new background agent on the runtime registry.

    Idempotent enough to be safe under spawn-then-resume races; if an
    entry already exists at ``agent_id`` it is replaced (the resume
    path explicitly wants this so the new event-loop wiring takes
    over).

    Returns the just-registered state so callers can hold a reference
    without re-fetching.

    The transcript path is computed (and the parent dir created) before
    registration so any consumer reading ``state.output_file`` can rely
    on the path being valid filesystem-side. ``output_file`` carries
    the JSONL transcript per WI-2.2 — Phase 3 / WI-3.1 cites it in the
    notification XML, Phase 7 / WI-7.4 reads it for auto-resume.
    """
    # Local import — transcript depends on a CSPRNG-validated agent_id,
    # registered task system imports back through here, defer to keep
    # the cycle untangled.
    from src.agent.transcript import get_agent_transcript_path

    output_file = get_agent_transcript_path(agent_id)

    state = LocalAgentTaskState(
        id=agent_id,
        type="local_agent",
        status="running",
        description=description,
        start_time=time.time(),
        output_file=output_file,
        agent_id=agent_id,
        agent_type=agent_type,
        prompt=prompt,
        selected_agent=selected_agent,
        model=model,
        tool_use_id=tool_use_id,
        is_backgrounded=True,
    )
    registry.upsert(state)
    return state


def queue_pending_message(