teams

@tool(context=True)
async def dispatch_team(
    team_id: str,
    mandate: str,
    agents: list[dict[str, Any]],
    tool_context: ToolContext,
    file_scope: list[str] | None = None,
    key_questions: list[str] | None = None,
    artifact_pointers: list[str] | None = None,
    acceptance_criteria: list[str] | None = None,
    authority: dict[str, Any] | None = None,
    budget: dict[str, Any] | None = None,
    input_revisions: list[str] | None = None,
    max_handoffs: int = 10,
    execution_timeout: float | None = None,
    node_timeout: float | None = None,
) -> str:
    """Dispatch a specialist team to investigate one area and persist its findings.

    Each team runs as a native Strands Swarm — a group of agents that hand off to
    one another to complete a mandate. The team's task is automatically augmented
    with a Required-Output block instructing the last agent to write its findings
    and metadata files (paths are supplied in the augmented task). A confined
    ``write_file`` tool is auto-added to every agent.

    TASK CONTRACT — a team is dispatched under a typed authority + budget contract:
        - ``authority`` grants capabilities. A team whose grant has
          ``may_create_blocking_finding`` false CANNOT persist a blocking finding:
          this tool strips any such finding from the team's typed findings and
          returns an error. Enforcement is in this tool's code, not the agents'
          prompts. Omit ``authority`` to permit blocking findings (legacy behavior).
        - ``budget`` bounds resources. A positive ``max_wall_clock_minutes`` sets
          the swarm's execution timeout to ``max_wall_clock_minutes * 60``.
        - ``acceptance_criteria`` are the conditions the team must satisfy; they
          are surfaced in the team task AND checked for coverage by the quality hook.
        - ``input_revisions`` are opaque revision-id strings the team builds on.

    AGENT SPEC FORMAT — each entry in ``agents`` is a dict with:
        - ``name`` (str): unique node name (e.g. ``"analyst"``, ``"reader"``).
        - ``system_prompt`` (str): role and instructions for this agent.
        - ``tools`` (list[str]): tool names for this agent, resolved from the
          coordinator's tool registry. Unknown names are dropped.

    Args:
        team_id: Unique team identifier (e.g. ``"team-structure"``).
        mandate: What to investigate and why (1-3 sentences).
        agents: List of agent spec dicts (see AGENT SPEC FORMAT).
        tool_context: Framework-injected run context (invocation state).
        file_scope: Optional file paths or directories to focus on.
        key_questions: Optional questions the team should answer.
        artifact_pointers: Optional pre-computed artifact paths to consult.
        acceptance_criteria: Optional conditions the team's output must satisfy.
        authority: Optional capability grant as a JSON dict with any of
            ``may_create_blocking_finding``, ``may_write``, ``may_approve``
            (each defaults false). Omit to permit blocking findings (legacy).
        budget: Optional resource envelope as a JSON dict with any of
            ``max_wall_clock_minutes``, ``max_cost_usd``, ``max_retries``.
        input_revisions: Optional opaque revision-id strings the team builds on.
        max_handoffs: Max handoffs (and iterations) within the team.
        execution_timeout: Max seconds for the entire team. Omit for the run default.
        node_timeout: Max seconds per agent turn. Omit for the run default.

    Returns:
        JSON with ``team_id``, ``status``, ``duration_seconds``, ``findings_path``,
        ``metadata_path``, and a result ``summary``. On failure — including a
        rejected forbidden blocking finding — ``status`` is ``"error"`` and
        ``message`` carries the reason.
    """
    state = _require_state(tool_context)
    layout = state["layout"]

    # Parse the task-contract dicts into typed models (AC-2-2). A malformed
    # LLM-supplied dict yields the tool's standard error envelope rather than an
    # escaping ValidationError (AC-2-2/AC-2-3).
    parsed = _parse_contract(layout, team_id=team_id, agent_count=len(agents), authority=authority, budget=budget)
    if isinstance(parsed, str):
        return parsed
    grant, team_budget = parsed

    # A positive wall-clock budget overrides the run-default execution timeout
    # (AC-2-5); an explicit ``execution_timeout`` argument still wins over both.
    effective_exec = _effective_execution_timeout(
        execution_timeout=execution_timeout,
        budget=team_budget,
        default=state["execution_timeout"],
    )
    effective_node = node_timeout if node_timeout is not None else state["node_timeout"]

    agent_names = [a.get("name", "?") for a in agents]
    logger.debug(
        f"dispatch_team({team_id}): execution_timeout={effective_exec}s, "
        f"node_timeout={effective_node}s, agents={agent_names}",
    )

    team_dir = layout.team_dir(team_id)
    team_dir.mkdir(parents=True, exist_ok=True)
    findings_path = str(layout.findings_path(team_id))
    findings_json_path = str(layout.findings_json_path(team_id))
    metadata_path = str(layout.metadata_path(team_id))

    # Persist the task contract next to the team's findings so it survives the run
    # and the quality hook can read acceptance criteria for coverage (AC-2-6).
    _write_team_contract(
        layout,
        team_id,
        authority=grant,
        budget=team_budget,
        acceptance_criteria=acceptance_criteria,
        input_revisions=input_revisions,
    )

    augmented_task = _build_augmented_task(
        mandate=mandate,
        findings_path=findings_path,
        findings_json_path=findings_json_path,
        metadata_path=metadata_path,
        file_scope=file_scope,
        key_questions=key_questions,
        artifact_pointers=artifact_pointers,
        acceptance_criteria=acceptance_criteria,
        input_revisions=input_revisions,
        output_root=str(layout.output_root),
    )

    write_file_tool = make_write_file(layout)
    write_findings_tool = make_write_findings(layout, team_id)
    start = time.monotonic()

    # Build a NEW invocation_state carrying the per-team contract by reference,
    # rather than mutating the coordinator's shared state dict (AC-5-1, AC-5-4).
    # The live AuthorityGrant/Budget objects cross no serialization boundary and
    # carry no session_manager; node tools must treat them read-only (AC-5-3).
    team_state: dict[str, Any] = {
        **state,
        "authority": grant,
        "budget": team_budget,
        "acceptance_criteria": list(acceptance_criteria or []),
        "input_revisions": list(input_revisions or []),
    }

    status, summary, message, error_type = await _run_and_enforce(
        agents,
        state=state,
        layout=layout,
        team_id=team_id,
        grant=grant,
        write_file_tool=write_file_tool,
        write_findings_tool=write_findings_tool,
        augmented_task=augmented_task,
        team_state=team_state,
        max_handoffs=max_handoffs,
        execution_timeout=effective_exec,
        node_timeout=effective_node,
    )
    duration = time.monotonic() - start

    # AC-2-2/AC-2-3: on EVERY path (success, failure, timeout) write today's debug
    # entry AND emit a typed dispatch TeamEvent to the event plane.
    _record_dispatch_completion(
        layout,
        state,
        team_id=team_id,
        status=status,
        duration_seconds=round(duration, 1),
        effective_exec=effective_exec,
        effective_node=effective_node,
        agent_names=agent_names,
        agent_count=len(agents),
        mandate=mandate,
        error_type=error_type,
    )

    response: dict[str, Any] = {
        "team_id": team_id,
        "status": status,
        "duration_seconds": round(duration, 1),
        "findings_path": findings_path,
        "findings_json_path": findings_json_path,
        "metadata_path": metadata_path,
        "summary": summary,
    }
    if message is not None:
        response["message"] = message
    if error_type:
        response["error_type"] = error_type
    return json.dumps(response)

@tool(context=True)
def read_team_findings(tool_context: ToolContext, team_id: str | None = None) -> str:
    """Read findings from dispatched teams.

    If ``team_id`` is ``None``, lists all team directories with their status,
    metadata summary, and whether typed findings are present — use this to see
    which teams have completed. If ``team_id`` is given, returns that team's typed
    ``findings.json`` (parsed) when it exists and validates, AND the human
    ``findings.md`` narrative alongside it; when the typed sidecar is absent or
    invalid it falls back to the raw ``findings.md`` markdown.

    Args:
        tool_context: Framework-injected run context (invocation state).
        team_id: Specific team to read, or ``None`` to list all teams.

    Returns:
        JSON listing all teams (when ``team_id`` is ``None``), or the findings
        content (when a ``team_id`` is given). For a typed read, ``findings`` is a
        list of finding objects, ``source`` is ``"typed"``, ``evidence_gate`` flags
        any blocking finding lacking evidence, ``findings_path`` points at the
        human ``findings.md``, and ``markdown`` carries its full narrative (mermaid,
        file:line prose, structured sections) when the file exists — this markdown
        is the rich input for bundle synthesis. For a fallback read, ``findings``
        is the markdown string and ``source`` is ``"markdown"``.
    """
    layout = _layout_of(tool_context)
    teams_root = layout.teams_dir()

    if team_id is None:
        if not teams_root.exists():
            return json.dumps({"status": "no_teams", "teams": []})
        teams = []
        for team_dir in sorted(teams_root.iterdir()):
            if not team_dir.is_dir():
                continue
            meta_path = team_dir / layout.metadata_name
            findings_file = team_dir / layout.findings_name
            meta = None
            if meta_path.exists():
                try:
                    meta = json.loads(meta_path.read_text())
                except (json.JSONDecodeError, OSError):
                    pass  # Corrupt/unreadable metadata — skip it.
            typed = _try_load_typed_findings(layout, team_dir.name)
            teams.append(
                {
                    "team_id": team_dir.name,
                    "has_findings": findings_file.exists(),
                    "findings_lines": len(findings_file.read_text().splitlines()) if findings_file.exists() else 0,
                    "has_typed_findings": typed is not None,
                    "typed_finding_count": len(typed) if typed is not None else 0,
                    "metadata": meta,
                },
            )
        return json.dumps({"status": "ok", "teams": teams}, indent=2)

    team_dir = layout.team_dir(team_id)
    if not team_dir.exists():
        return json.dumps({"status": "error", "message": f"Team directory not found: {team_id}"})

    # Prefer the typed sidecar, but ALWAYS also surface the human findings.md so
    # bundle synthesis keeps its rich markdown input (mermaid, file:line prose,
    # structured sections) — the typed findings augment, never replace, it
    # (AC-6-2). Even when findings.md is absent, ``findings_path`` lets the
    # coordinator discover the markdown and read it via ``read_file_bounded``.
    findings_file = layout.findings_path(team_id)
    typed = _try_load_typed_findings(layout, team_id)
    if typed is not None:
        return json.dumps(_typed_read_payload(team_id, typed, findings_file))

    if not findings_file.exists():
        return json.dumps(
            {
                "status": "no_findings",
                "message": f"Team {team_id} has no findings — it may still be running or failed before writing.",
            },
        )

    content = findings_file.read_text()
    return json.dumps(
        {
            "status": "ok",
            "team_id": team_id,
            "source": "markdown",
            "findings": content,
            "line_count": len(content.splitlines()),
        },
    )

@tool(context=True)
def write_artifact(area: str, content: str, tool_context: ToolContext, is_executive_summary: bool = False) -> str:
    """Write a per-area narrative artifact or the top-level executive summary.

    All artifact paths come from the run's layout. Diagrams must use mermaid
    code-fenced blocks (never ASCII art). The artifact section structure the
    coordinator should follow is described in its system prompt, not here.

    Args:
        area: Artifact area identifier (e.g. ``"auth"``). Ignored when
            ``is_executive_summary`` is True.
        content: The full markdown content to write.
        tool_context: Framework-injected run context (invocation state).
        is_executive_summary: If True, write the executive summary instead of a
            per-area artifact.

    Returns:
        JSON with ``status``, absolute ``path``, ``relative_path``, and ``line_count``.
    """
    layout = _layout_of(tool_context)

    if is_executive_summary:
        file_path = layout.executive_summary_path()
    else:
        layout.artifacts_dir().mkdir(parents=True, exist_ok=True)
        file_path = layout.artifact_path(area)

    file_path.parent.mkdir(parents=True, exist_ok=True)
    try:
        file_path.write_text(content, encoding="utf-8")
    except OSError as e:
        return json.dumps({"status": "error", "message": str(e)})

    line_count = len(content.splitlines())
    try:
        relative_path = str(file_path.relative_to(layout.output_root))
    except ValueError:
        relative_path = file_path.name
    logger.info(f"write_artifact: wrote {file_path.name} ({line_count} lines)")
    return json.dumps(
        {
            "status": "ok",
            "path": str(file_path),
            "relative_path": relative_path,
            "line_count": line_count,
        },
    )

@tool(context=True)
def score_narrative(area: str, tool_context: ToolContext) -> str:
    """Score a per-area artifact's quality using the run's rubric.

    Call after ``write_artifact`` to evaluate quality. If ``needs_revision`` is
    True (total below the rubric's revision threshold), call ``enrich_artifact``
    and rewrite via ``write_artifact``.

    Args:
        area: The artifact area to score (e.g. ``"auth"``).
        tool_context: Framework-injected run context (invocation state).

    Returns:
        JSON with ``area``, per-dimension ``scores`` and ``counts``, the ``total``
        average, the ``needs_revision`` flag, and improvement ``suggestions``.
    """
    state = _require_state(tool_context)
    layout = state["layout"]
    rubric = state["rubric"]

    artifact_path = layout.artifact_path(area)
    if not artifact_path.exists():
        return json.dumps({"status": "error", "message": f"Artifact not found: {artifact_path}"})

    report = rubric.score(artifact_path.read_text())
    return json.dumps(
        {
            "status": "ok",
            "area": area,
            "scores": report.scores,
            "counts": report.counts,
            "total": report.total,
            "needs_revision": report.needs_revision,
            "suggestions": report.suggestions,
        },
    )

@tool(context=True)
def enrich_artifact(area: str, feedback: str, tool_context: ToolContext) -> str:
    """Read an existing artifact and prepare enrichment context.

    Call when ``score_narrative`` reports ``needs_revision``. Returns the current
    artifact content plus the feedback so you can rewrite it with improvements via
    ``write_artifact``.

    Args:
        area: The artifact area to enrich (e.g. ``"auth"``).
        feedback: Specific feedback from ``score_narrative`` about what to improve.
        tool_context: Framework-injected run context (invocation state).

    Returns:
        JSON with the current artifact ``current_content``, its ``line_count``, the
        ``feedback``, and a rewrite ``instruction``.
    """
    layout = _layout_of(tool_context)
    artifact_path = layout.artifact_path(area)
    if not artifact_path.exists():
        return json.dumps({"status": "error", "message": f"Artifact not found: {artifact_path}"})

    content = artifact_path.read_text()
    return json.dumps(
        {
            "status": "ok",
            "area": area,
            "current_content": content,
            "line_count": len(content.splitlines()),
            "feedback": feedback,
            "instruction": (
                "Rewrite this artifact incorporating the feedback above. "
                "Use write_artifact to save the improved version. "
                "Preserve all existing accurate information while adding depth."
            ),
        },
    )