Function build_forked_messages

src/agent/fork_subagent.py:183–253

Build the trailing message pair for a fork child. Mirrors ``buildForkedMessages()`` from ``forkSubagent.ts:107-169``. For prompt-cache sharing, every fork child must produce a byte-identical API request prefix. This function: 1. Clones the parent assistant message (preserving ever

(
    directive: str,
    parent_assistant: AssistantMessage | None,
)

def build_forked_messages(
    directive: str,
    parent_assistant: AssistantMessage | None,
) -> list[Message]:
    """Build the trailing message pair for a fork child.

    Mirrors ``buildForkedMessages()`` from ``forkSubagent.ts:107-169``.

    For prompt-cache sharing, every fork child must produce a byte-identical
    API request prefix. This function:

    1. Clones the parent assistant message (preserving every ``tool_use``
       block and its ID).
    2. For each ``tool_use`` block, emits a ``tool_result`` with the
       constant ``FORK_PLACEHOLDER_RESULT`` text.
    3. Appends a single user message containing all those placeholder
       ``tool_result`` blocks followed by the boilerplate-wrapped
       directive.

    Result: ``[cloned_assistant, user_message]``. Only the final text block
    differs per child, maximizing cache hits.

    If the parent assistant has no ``tool_use`` blocks (or no parent at
    all), we fall back to a single user message carrying the directive.
    """
    if parent_assistant is None:
        return [
            create_user_message(content=[TextBlock(text=build_child_message(directive))])
        ]

    tool_use_blocks = _collect_tool_use_blocks(parent_assistant.content)

    if not tool_use_blocks:
        return [
            create_user_message(content=[TextBlock(text=build_child_message(directive))])
        ]

    # Clone the assistant message — new uuid, deep-copied content list — so
    # we never mutate the parent's message in place.
    cloned_content: list[Any] = []
    for block in parent_assistant.content if isinstance(parent_assistant.content, list) else []:
        cloned_content.append(copy.deepcopy(block))
    cloned_assistant = AssistantMessage(
        content=cloned_content,
        uuid=str(uuid4()),
        timestamp=parent_assistant.timestamp,
        stop_reason=parent_assistant.stop_reason,
        model=parent_assistant.model,
        usage=parent_assistant.usage,
        requestId=parent_assistant.requestId,
    )

    # Build placeholder tool_result blocks for every parent tool_use block.
    tool_result_blocks: list[ToolResultBlock] = []
    for block in tool_use_blocks:
        tool_use_id = _get_attr(block, "id", "") or ""
        tool_result_blocks.append(
            ToolResultBlock(