"""Derived Cowork graph and trace snapshots for UI/API surfaces."""

from __future__ import annotations

import math
import hashlib
import mimetypes
from collections import Counter, defaultdict
from dataclasses import asdict
from pathlib import Path
from typing import Any

from tinybot.cowork.swarm import build_swarm_parallel_metrics, build_swarm_scheduler_queues
from tinybot.cowork.types import CoworkAgentStep, CoworkEvent, CoworkSession, CoworkStepSummary, now_iso


def _compact(value: Any, limit: int = 180) -> str:
    text = " ".join(str(value or "").split())
    if len(text) <= limit:
        return text
    if limit <= 3:
        return text[:limit]
    return text[: max(0, limit - 3)].rstrip() + "..."


def _status_tone(status: str) -> str:
    value = str(status or "").lower()
    if value in {"completed", "done", "replied"}:
        return "completed"
    if value in {"failed", "blocked", "expired"}:
        return "failed"
    if value in {"working", "in_progress", "active"}:
        return "active"
    if value in {"waiting", "queued", "delivered", "read", "pending", "paused"}:
        return "pending"
    return "idle"


def _add_edge(edges: list[dict[str, Any]], source: str, target: str, kind: str, **extra: Any) -> None:
    if not source or not target or source == target:
        return
    key = (source, target, kind)
    if any((edge.get("from"), edge.get("to"), edge.get("kind")) == key for edge in edges):
        return
    payload = {"from": source, "to": target, "source": source, "target": target, "kind": kind}
    payload.update(extra)
    edges.append(payload)


def build_cowork_graph(
    session: CoworkSession,
    *,
    node_limit: int = 160,
    edge_limit: int = 260,
) -> dict[str, Any]:
    """Return a stable graph projection of a Cowork session.

    The shape intentionally includes both ``from/to`` and ``source/target`` so
    current WebUI code and teacher-inspired graph components can consume it.
    """

    nodes: list[dict[str, Any]] = []
    edges: list[dict[str, Any]] = []
    node_rank: dict[str, int] = {}

    def add_node(node: dict[str, Any], *, rank: int = 100) -> None:
        node.setdefault("label", node.get("title", node.get("id", "")))
        node.setdefault("title", node.get("label", node.get("id", "")))
        node.setdefault("detail", "")
        node.setdefault("status", "idle")
        node.setdefault("tone", _status_tone(str(node.get("status", ""))))
        nodes.append(node)
        node_rank[node["id"]] = rank

    def agent_priority(agent: Any) -> tuple[int, int, int, int, int]:
        waiting = sum(
            1
            for record in session.mailbox.values()
            if agent.id in record.recipient_ids and record.requires_reply and record.status in {"delivered", "read"}
        )
        open_tasks = sum(
            1
            for task in session.tasks.values()
            if task.assigned_agent_id == agent.id and task.status in {"pending", "in_progress"}
        )
        active_status = 1 if agent.status in {"working", "waiting", "blocked"} else 0
        retired = 1 if getattr(agent, "lifecycle_status", "active") == "retired" else 0
        return (active_status, waiting, open_tasks, len(agent.inbox), -retired)

    add_node(
        {
            "id": "session",
            "kind": "session",
            "label": session.title,
            "title": session.title,
            "detail": _compact(session.current_focus_task or session.goal, 220),
            "status": session.status,
            "tone": _status_tone(session.status),
            "badge": getattr(session, "workflow_mode", "adaptive_starter"),
            "x": 600,
            "y": 310,
            "workflow_mode": getattr(session, "workflow_mode", "adaptive_starter"),
            "architecture": getattr(session, "workflow_mode", "adaptive_starter"),
            "source_blueprint_id": (getattr(session, "blueprint", {}) or {}).get("id", ""),
        },
        rank=0,
    )

    budget_state = getattr(session, "budget_usage", {}) or {}
    add_node(
        {
            "id": "budget",
            "kind": "budget",
            "label": "Budget",
            "title": "Budget",
            "detail": _compact(f"calls {budget_state.get('agent_calls', 0)} / rounds {budget_state.get('rounds', 0)} / stop {getattr(session, 'stop_reason', '') or '-'}", 180),
            "status": "blocked" if getattr(session, "stop_reason", "") and "budget" in getattr(session, "stop_reason", "") else "active",
            "badge": getattr(session, "stop_reason", "") or "limits",
            "x": 980,
            "y": 82,
        },
        rank=8,
    )
    _add_edge(edges, "session", "budget", "has_budget")

    all_agents = sorted(session.agents.values(), key=agent_priority, reverse=True)
    agent_radius_x = max(260, min(390, 190 + min(len(all_agents), 12) * 16))
    agent_radius_y = max(165, min(245, 122 + min(len(all_agents), 12) * 9))
    for index, agent in enumerate(all_agents):
        angle = -math.pi / 2 + (math.tau * index) / max(len(all_agents), 1)
        x = 600 + math.cos(angle) * agent_radius_x
        y = 310 + math.sin(angle) * agent_radius_y
        pending_replies = [
            record.id
            for record in session.mailbox.values()
            if agent.id in record.recipient_ids and record.requires_reply and record.status in {"delivered", "read"}
        ]
        latest_direct = next(
            (
                record
                for record in reversed(list(session.mailbox.values()))
                if agent.id in record.recipient_ids or record.sender_id == agent.id
            ),
            None,
        )
        detail_parts = [
            agent.role or "Agent",
            agent.current_task_title or agent.goal,
            latest_direct.content if latest_direct else "",
        ]
        add_node(
            {
                "id": f"agent:{agent.id}",
                "entity_id": agent.id,
                "kind": "agent",
                "label": agent.name,
                "title": agent.name,
                "detail": _compact(" - ".join(part for part in detail_parts if part), 220),
                "status": agent.status,
                "tone": _status_tone(agent.status),
                "badge": f"in {len(agent.inbox)} / wait {len(pending_replies)} / r{agent.rounds or 0}",
                "x": round(x, 2),
                "y": round(y, 2),
                "parent_agent_id": getattr(agent, "parent_agent_id", None),
                "team_id": getattr(agent, "team_id", ""),
                "lifetime": getattr(agent, "lifetime", "persistent"),
                "lifecycle_status": getattr(agent, "lifecycle_status", "active"),
                "source_blueprint_id": getattr(agent, "source_blueprint_id", ""),
                "source_event_id": getattr(agent, "source_event_id", ""),
                "runtime_created": not bool(getattr(agent, "source_blueprint_id", "")),
            },
            rank=10 + index,
        )
        _add_edge(edges, "session", f"agent:{agent.id}", "member")
        parent_agent_id = getattr(agent, "parent_agent_id", None)
        if parent_agent_id and parent_agent_id in session.agents:
            _add_edge(edges, f"agent:{parent_agent_id}", f"agent:{agent.id}", "parent_of", team_id=getattr(agent, "team_id", ""))
            _add_edge(edges, f"agent:{parent_agent_id}", f"agent:{agent.id}", "spawned", source_event_id=getattr(agent, "source_event_id", ""))

    task_base_y = 72
    for index, task in enumerate(session.tasks.values()):
        add_node(
            {
                "id": f"task:{task.id}",
                "entity_id": task.id,
                "kind": "task",
                "label": task.title,
                "title": task.title,
                "detail": _compact(task.result_data.get("answer") or task.result or task.description, 220),
                "status": task.status,
                "tone": _status_tone(task.status),
                "badge": task.assigned_agent_id or "shared",
                "x": 160 + (index % 4) * 190,
                "y": task_base_y + (index // 4) * 74,
                "owner": task.assigned_agent_id,
                "dependencies": task.dependencies,
                "priority": getattr(task, "priority", 0),
                "review_required": getattr(task, "review_required", False),
                "review_status": getattr(task, "review_status", ""),
                "fanout_group_id": getattr(task, "fanout_group_id", ""),
                "merge_task_id": getattr(task, "merge_task_id", ""),
                "source_blueprint_id": getattr(task, "source_blueprint_id", ""),
                "source_event_id": getattr(task, "source_event_id", ""),
                "runtime_created": getattr(task, "runtime_created", False),
            },
            rank=40 + index,
        )
        _add_edge(edges, "session", f"task:{task.id}", "has_task")
        if task.assigned_agent_id and task.assigned_agent_id in session.agents:
            _add_edge(edges, f"task:{task.id}", f"agent:{task.assigned_agent_id}", "assigned_to")
        for dependency in task.dependencies:
            if dependency in session.tasks:
                _add_edge(edges, f"task:{dependency}", f"task:{task.id}", "depends_on")
        if getattr(task, "merge_task_id", "") and task.merge_task_id in session.tasks:
            _add_edge(edges, f"task:{task.id}", f"task:{task.merge_task_id}", "synthesizes")

    for index, thread in enumerate(session.threads.values()):
        add_node(
            {
                "id": f"thread:{thread.id}",
                "entity_id": thread.id,
                "kind": "thread",
                "label": thread.topic,
                "title": thread.topic,
                "detail": _compact(thread.summary or f"{len(thread.message_ids)} message(s)", 160),
                "status": thread.status,
                "badge": f"{len(thread.message_ids)} msg",
                "x": 1080,
                "y": 170 + index * 58,
            },
            rank=70 + index,
        )
        _add_edge(edges, "session", f"thread:{thread.id}", "has_thread")
        for participant_id in thread.participant_ids:
            if participant_id in session.agents:
                _add_edge(edges, f"thread:{thread.id}", f"agent:{participant_id}", "participant")

    for index, message in enumerate(list(session.messages.values())[-80:]):
        node_id = f"message:{message.id}"
        add_node(
            {
                "id": node_id,
                "entity_id": message.id,
                "kind": "message",
                "label": _compact(message.content, 52),
                "title": _compact(message.content, 80),
                "detail": _compact(message.content, 220),
                "status": "read" if message.read_by else "delivered",
                "badge": message.sender_id,
                "x": 1040,
                "y": 430 + (index % 8) * 42,
                "thread_id": message.thread_id,
            },
            rank=110 + index,
        )
        sender = f"agent:{message.sender_id}" if message.sender_id in session.agents else "session"
        _add_edge(edges, sender, node_id, "sent")
        if message.thread_id in session.threads:
            _add_edge(edges, f"thread:{message.thread_id}", node_id, "contains")
        for recipient_id in message.recipient_ids:
            if recipient_id in session.agents:
                _add_edge(edges, node_id, f"agent:{recipient_id}", "delivered_to")

    for index, record in enumerate(session.mailbox.values()):
        node_id = f"mailbox:{record.id}"
        add_node(
            {
                "id": node_id,
                "entity_id": record.id,
                "kind": "mailbox",
                "label": record.request_type or record.kind or "mailbox",
                "title": record.request_type or record.kind or "Mailbox",
                "detail": _compact(record.content, 220),
                "status": record.status,
                "badge": "reply" if record.requires_reply else record.kind,
                "x": 72 + (index % 5) * 145,
                "y": 540 + (index // 5) * 48,
                "sender_id": record.sender_id,
                "recipient_ids": record.recipient_ids,
                "topic": getattr(record, "topic", ""),
                "event_type": getattr(record, "event_type", ""),
                "request_type": record.request_type,
                "correlation_id": record.correlation_id,
                "lineage_id": getattr(record, "lineage_id", None),
                "reply_to_envelope_id": record.reply_to_envelope_id,
                "caused_by_envelope_id": getattr(record, "caused_by_envelope_id", None),
                "blocking_task_id": record.blocking_task_id,
                "escalated_at": getattr(record, "escalated_at", None),
            },
            rank=90 + index,
        )
        sender = f"agent:{record.sender_id}" if record.sender_id in session.agents else "session"
        _add_edge(edges, sender, node_id, "sent", topic=getattr(record, "topic", ""), event_type=getattr(record, "event_type", ""))
        for recipient_id in record.recipient_ids:
            if recipient_id not in session.agents:
                continue
            _add_edge(
                edges,
                node_id,
                f"agent:{recipient_id}",
                "delivered_to",
                pulse=record.requires_reply,
                status=record.status,
                topic=getattr(record, "topic", ""),
                event_type=getattr(record, "event_type", ""),
                lineage_id=getattr(record, "lineage_id", None),
                request_type=record.request_type,
                requires_reply=record.requires_reply,
                detail=_compact(record.content, 180),
            )
        if record.message_id in session.messages:
            _add_edge(edges, node_id, f"message:{record.message_id}", "materialized_as")
        if record.thread_id in session.threads:
            _add_edge(edges, node_id, f"thread:{record.thread_id}", "in_thread")
        if record.reply_to_envelope_id and record.reply_to_envelope_id in session.mailbox:
            _add_edge(edges, node_id, f"mailbox:{record.reply_to_envelope_id}", "replied_to")
        if record.caused_by_envelope_id and record.caused_by_envelope_id in session.mailbox:
            _add_edge(edges, node_id, f"mailbox:{record.caused_by_envelope_id}", "caused_by")
        if record.blocking_task_id and record.blocking_task_id in session.tasks:
            _add_edge(edges, node_id, f"task:{record.blocking_task_id}", "blocks")

    for index, artifact in enumerate(_session_artifacts(session)):
        artifact_id = artifact["id"]
        add_node(
            {
                "id": artifact_id,
                "kind": "artifact",
                "label": _compact(artifact["value"], 70),
                "title": artifact["value"],
                "detail": artifact["value"],
                "status": "completed",
                "badge": artifact.get("kind", "artifact"),
                "x": 980,
                "y": 520 + index * 46,
                "source_task_id": artifact.get("source_task_id"),
                "source_agent_id": artifact.get("source_agent_id"),
            },
            rank=130 + index,
        )
        if artifact.get("source_task_id") in session.tasks:
            _add_edge(edges, f"task:{artifact['source_task_id']}", artifact_id, "produced")
        if artifact.get("source_agent_id") in session.agents:
            _add_edge(edges, f"agent:{artifact['source_agent_id']}", artifact_id, "produced")

    memory_index = 0
    memory = getattr(session, "shared_memory", {}) or {}
    if isinstance(memory, dict):
        for bucket, entries in memory.items():
            if not isinstance(entries, list):
                continue
            for entry in entries[-12:]:
                if not isinstance(entry, dict):
                    continue
                memory_id = f"memory:{bucket}:{memory_index}"
                memory_index += 1
                add_node(
                    {
                        "id": memory_id,
                        "kind": "memory",
                        "label": bucket,
                        "title": bucket,
                        "detail": _compact(entry.get("text", ""), 220),
                        "status": "completed",
                        "badge": bucket,
                        "x": 56 + (memory_index % 6) * 150,
                        "y": 24 + (memory_index // 6) * 50,
                        "source_task_id": entry.get("source_task_id"),
                        "author": entry.get("author"),
                    },
                    rank=140 + memory_index,
                )
                if entry.get("source_task_id") in session.tasks:
                    _add_edge(edges, f"task:{entry['source_task_id']}", memory_id, "uses_memory")
                if entry.get("author") in session.agents:
                    _add_edge(edges, f"agent:{entry['author']}", memory_id, "uses_memory")

    if getattr(session, "completion_decision", None):
        decision = session.completion_decision
        add_node(
            {
                "id": "decision:latest",
                "kind": "decision",
                "label": decision.get("next_action", "decision"),
                "title": decision.get("next_action", "Decision"),
                "detail": _compact(decision.get("reason", ""), 220),
                "status": "completed" if decision.get("ready_to_finish") else "pending",
                "badge": getattr(session, "stop_reason", "") or "latest",
                "x": 600,
                "y": 36,
            },
            rank=5,
        )
        _add_edge(edges, "session", "decision:latest", "has_decision")
        for blocker in decision.get("blocked", []) if isinstance(decision.get("blocked", []), list) else []:
            if blocker.get("id") in session.mailbox:
                _add_edge(edges, f"mailbox:{blocker['id']}", "decision:latest", "blocks")

    total_nodes = len(nodes)
    total_edges = len(edges)
    visible_ids = _focused_node_ids(nodes, node_rank, node_limit)
    visible_nodes = [node for node in nodes if node["id"] in visible_ids]
    visible_edges = [edge for edge in edges if edge.get("from") in visible_ids and edge.get("to") in visible_ids][:edge_limit]
    hidden_nodes = max(0, total_nodes - len(visible_nodes))
    hidden_edges = max(0, total_edges - len(visible_edges))
    node_kinds = Counter(node["kind"] for node in nodes)
    edge_kinds = Counter(edge["kind"] for edge in edges)
    visible_node_kinds = Counter(node["kind"] for node in visible_nodes)
    visible_edge_kinds = Counter(edge["kind"] for edge in visible_edges)

    return {
        "schema_version": "cowork.graph.v2",
        "generated_at": now_iso(),
        "nodes": visible_nodes,
        "edges": visible_edges,
        "stats": {
            "nodes": len(visible_nodes),
            "edges": len(visible_edges),
            "total_nodes": total_nodes,
            "total_edges": total_edges,
            "hidden_nodes": hidden_nodes,
            "hidden_edges": hidden_edges,
            "node_kinds": dict(sorted(visible_node_kinds.items())),
            "edge_kinds": dict(sorted(visible_edge_kinds.items())),
            "total_node_kinds": dict(sorted(node_kinds.items())),
            "total_edge_kinds": dict(sorted(edge_kinds.items())),
            "agents": visible_node_kinds.get("agent", 0),
            "total_agents": len(session.agents),
            "tasks": len(session.tasks),
            "threads": len(session.threads),
            "mailbox": len(session.mailbox),
            "artifacts": len(_session_artifacts(session)),
            "memory": sum(len(entries) for entries in (getattr(session, "shared_memory", {}) or {}).values() if isinstance(entries, list)),
            "communications": sum(1 for edge in visible_edges if edge.get("kind") in {"delivered_to", "sent"}),
        },
        "truncated": {
            "nodes": hidden_nodes > 0,
            "edges": hidden_edges > 0,
            "hidden_nodes": hidden_nodes,
            "hidden_edges": hidden_edges,
            "limits": {"nodes": node_limit, "edges": edge_limit},
        },
    }


def _focused_node_ids(nodes: list[dict[str, Any]], node_rank: dict[str, int], limit: int) -> set[str]:
    if len(nodes) <= limit:
        return {node["id"] for node in nodes}
    per_kind_limits = {
        "session": 1,
        "budget": 1,
        "decision": 2,
        "agent": 40,
        "task": 50,
        "thread": 18,
        "mailbox": 34,
        "message": 28,
        "artifact": 18,
        "memory": 18,
    }
    by_kind: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for node in nodes:
        by_kind[str(node.get("kind") or "other")].append(node)
    selected: set[str] = set()
    for kind, items in by_kind.items():
        kind_limit = per_kind_limits.get(kind, 10)
        ordered = sorted(items, key=lambda node: node_rank.get(node["id"], 1000))
        selected.update(node["id"] for node in ordered[:kind_limit])
    if len(selected) > limit:
        ordered_selected = sorted((node for node in nodes if node["id"] in selected), key=lambda node: node_rank.get(node["id"], 1000))
        selected = {node["id"] for node in ordered_selected[:limit]}
    elif len(selected) < limit:
        for node in sorted(nodes, key=lambda item: node_rank.get(item["id"], 1000)):
            selected.add(node["id"])
            if len(selected) >= limit:
                break
    return selected


def _session_artifacts(session: CoworkSession) -> list[dict[str, Any]]:
    artifacts: list[dict[str, Any]] = []
    seen: set[tuple[str, str]] = set()
    for task in session.tasks.values():
        for value in _task_artifacts(task):
            key = (task.id, value)
            if key in seen:
                continue
            seen.add(key)
            artifacts.append(
                {
                    "id": f"artifact:{hashlib.sha1(f'{task.id}:{value}'.encode()).hexdigest()[:12]}",
                    "value": value,
                    "kind": _artifact_kind(value),
                    "source_task_id": task.id,
                    "source_agent_id": task.assigned_agent_id,
                }
            )
    for value in getattr(session, "artifacts", []):
        text = str(value or "").strip()
        if not text or any(item["value"] == text for item in artifacts):
            continue
        artifacts.append(
            {
                "id": f"artifact:{hashlib.sha1(text.encode('utf-8')).hexdigest()[:12]}",
                "value": text,
                "kind": _artifact_kind(text),
                "source_task_id": None,
                "source_agent_id": None,
            }
        )
    return artifacts


def _event_stage(event: CoworkEvent) -> str:
    event_type = str(event.type or "")
    if event_type.startswith("scheduler."):
        return "scheduler"
    if event_type.startswith("agent."):
        return "agent"
    if event_type.startswith("task."):
        return "task"
    if event_type.startswith("mailbox.") or event_type.startswith("message."):
        return "message"
    if event_type.startswith("session."):
        return "session"
    return "event"


def _event_action(event: CoworkEvent) -> str:
    labels = {
        "session.created": "Session created",
        "scheduler.round": "Scheduler round",
        "scheduler.idle": "Scheduler idle",
        "scheduler.lead_synthesis": "Lead synthesis",
        "agent.started": "Agent started",
        "agent.ran": "Agent finished",
        "agent.failed": "Agent failed",
        "task.added": "Task added",
        "task.assigned": "Task assigned",
        "task.completed": "Task completed",
        "mailbox.delivered": "Mailbox delivered",
        "mailbox.read": "Mailbox read",
    }
    return labels.get(str(event.type or ""), str(event.type or "Event").replace(".", " ").title())


def build_cowork_trace(session: CoworkSession, *, limit: int = 80) -> list[dict[str, Any]]:
    """Return timeline trace cards derived from events, spans, and scheduler decisions."""

    trace: list[dict[str, Any]] = []
    for step in getattr(session, "agent_steps", [])[-limit:]:
        summary = step.summary or _step_summary_from_step(step)
        trace.append(
            {
                "id": step.id,
                "type": f"agent_step.{step.action_kind}",
                "stage": "agent_step",
                "action": summary.purpose or step.action_kind.replace("_", " ").title(),
                "detail": summary.outcome_summary or step.output_summary or summary.input_summary,
                "actor_id": step.agent_id,
                "actor_name": session.agents.get(step.agent_id).name if step.agent_id in session.agents else step.agent_id,
                "at": step.ended_at or step.started_at,
                "status": step.status,
                "node_id": f"agent:{step.agent_id}" if step.agent_id in session.agents else "session",
                "next_node_id": f"task:{step.task_id}" if step.task_id in session.tasks else "",
                "payload": {
                    "branch_id": step.branch_id,
                    "architecture": step.architecture,
                    "task_id": step.task_id,
                    "work_unit_id": step.work_unit_id,
                    "scheduler_reason": step.scheduler_reason,
                    "linked_message_ids": step.linked_message_ids,
                    "linked_artifact_refs": step.linked_artifact_refs,
                    "linked_task_ids": step.linked_task_ids,
                    "linked_envelope_ids": step.linked_envelope_ids,
                    "tool_observation_ids": [item.id for item in step.tool_observations],
                    "browser_observation_ids": [item.id for item in step.browser_observations],
                    "source_span_id": step.source_span_id,
                    "source_event_id": step.source_event_id,
                    "projected": step.projected,
                    "error": step.error,
                },
                "source": "agent_step",
            }
        )
    for event in session.events[-limit:]:
        actor_id = event.actor_id or str(event.data.get("agent_id") or "")
        actor = session.agents.get(actor_id)
        node_id = f"agent:{actor_id}" if actor_id in session.agents else "session"
        next_node_id = ""
        data_next = event.data.get("next_agent_id") or event.data.get("assigned_agent_id")
        if data_next in session.agents:
            next_node_id = f"agent:{data_next}"
        task_id = event.data.get("task_id") or event.data.get("blocking_task_id")
        if task_id in session.tasks:
            next_node_id = f"task:{task_id}"

        trace.append(
            {
                "id": event.id,
                "type": event.type,
                "stage": _event_stage(event),
                "action": _event_action(event),
                "detail": event.message,
                "actor_id": actor_id or None,
                "actor_name": actor.name if actor else None,
                "at": event.created_at,
                "status": _status_tone(str(event.data.get("status") or event.type.rsplit(".", 1)[-1])),
                "node_id": node_id,
                "next_node_id": next_node_id,
                "payload": event.data,
                "source": "event",
            }
        )
    for decision in getattr(session, "scheduler_decisions", [])[-limit:]:
        selected = decision.get("selected_agent_ids", [])
        reasons = {
            item.get("agent_id"): item.get("activation_reasons", [])
            for item in decision.get("candidate_scores", [])
            if isinstance(item, dict)
        }
        trace.append(
            {
                "id": decision.get("id"),
                "type": "scheduler.decision",
                "stage": "scheduler",
                "action": "Scheduler selected agents",
                "detail": decision.get("reason", ""),
                "actor_id": "scheduler",
                "actor_name": "Scheduler",
                "at": decision.get("created_at", ""),
                "status": "active" if selected else "skipped",
                "node_id": "session",
                "next_node_id": f"agent:{selected[0]}" if selected else "",
                "payload": {
                    **decision,
                    "readiness_scores": decision.get("candidate_scores", []),
                    "activation_reasons": reasons,
                    "budget_usage": getattr(session, "budget_usage", {}),
                },
                "source": "scheduler_decision",
            }
        )
    for span in getattr(session, "trace_spans", [])[-limit:]:
        payload = dict(getattr(span, "data", {}) or {})
        stage = getattr(span, "kind", "trace") or "trace"
        trace.append(
            {
                "id": span.id,
                "type": f"span.{stage}",
                "stage": stage,
                "action": span.name,
                "detail": span.summary or span.output_ref or span.input_ref,
                "actor_id": span.actor_id,
                "actor_name": session.agents.get(span.actor_id).name if span.actor_id in session.agents else span.actor_id,
                "at": span.ended_at or span.started_at,
                "status": span.status,
                "node_id": f"agent:{span.actor_id}" if span.actor_id in session.agents else "session",
                "next_node_id": f"task:{payload.get('task_id')}" if payload.get("task_id") in session.tasks else "",
                "payload": {
                    **payload,
                    "run_id": span.run_id,
                    "round_id": span.round_id,
                    "input_ref": span.input_ref,
                    "output_ref": span.output_ref,
                    "error": span.error,
                },
                "source": "trace_span",
            }
        )
    if getattr(session, "stop_reason", "") and not any(item.get("payload", {}).get("stop_reason") == session.stop_reason for item in trace):
        trace.append(
            {
                "id": f"stop:{session.id}:{session.stop_reason}",
                "type": "scheduler.stop",
                "stage": "scheduler",
                "action": "Stop reason",
                "detail": session.stop_reason.replace("_", " "),
                "actor_id": "scheduler",
                "actor_name": "Scheduler",
                "at": session.updated_at,
                "status": "blocked" if "budget" in session.stop_reason or "blocker" in session.stop_reason else "completed",
                "node_id": "session",
                "next_node_id": "",
                "payload": {"stop_reason": session.stop_reason, "budget_usage": getattr(session, "budget_usage", {})},
                "source": "derived",
            }
        )
    return sorted(trace, key=lambda item: str(item.get("at") or ""))[-limit:]


def build_cowork_agent_steps(session: CoworkSession, *, limit: int = 120) -> list[dict[str, Any]]:
    """Return native Agent Steps or a bounded legacy projection from existing observability data."""

    native_steps = getattr(session, "agent_steps", []) or []
    if native_steps:
        return [_agent_step_payload(step) for step in native_steps[-limit:]]
    return [_agent_step_payload(step) for step in _project_legacy_agent_steps(session, limit=limit)]


def _agent_step_payload(step: CoworkAgentStep) -> dict[str, Any]:
    payload = asdict(step)
    summary = step.summary or _step_summary_from_step(step)
    payload["summary"] = asdict(summary)
    payload["tool_observations"] = [asdict(item) for item in step.tool_observations]
    payload["browser_observations"] = [asdict(item) for item in step.browser_observations]
    return payload


def _step_summary_from_step(step: CoworkAgentStep) -> CoworkStepSummary:
    outcome = step.output_summary or step.error or step.status.replace("_", " ")
    return CoworkStepSummary(
        id=f"summary:{step.id}",
        step_id=step.id,
        purpose=step.scheduler_reason or step.action_kind.replace("_", " ").title(),
        action_kind=step.action_kind,
        input_summary=_compact(step.input_summary, 220),
        outcome_summary=_compact(outcome, 240),
        next_effect=_compact(_step_next_effect(step), 180),
        has_full_detail=bool(step.detail_ref),
        detail_ref=step.detail_ref,
        redacted=False,
        created_at=step.ended_at or step.started_at,
    )


def _step_next_effect(step: CoworkAgentStep) -> str:
    if step.linked_task_ids:
        return f"Updated task(s): {', '.join(step.linked_task_ids[:4])}"
    if step.linked_message_ids:
        return f"Linked message(s): {', '.join(step.linked_message_ids[:4])}"
    if step.linked_artifact_refs:
        return f"Linked artifact(s): {', '.join(step.linked_artifact_refs[:4])}"
    return ""


def _project_legacy_agent_steps(session: CoworkSession, *, limit: int) -> list[CoworkAgentStep]:
    steps: list[CoworkAgentStep] = []
    branch_id = getattr(session, "current_branch_id", "default") or "default"
    branches = getattr(session, "branches", {}) if isinstance(getattr(session, "branches", {}), dict) else {}
    architecture = getattr(branches.get(branch_id), "architecture", getattr(session, "workflow_mode", "adaptive_starter"))

    for span in getattr(session, "trace_spans", [])[-limit:]:
        data = getattr(span, "data", {}) or {}
        if getattr(span, "kind", "") == "session":
            continue
        step = CoworkAgentStep(
            id=f"legacy-step:span:{span.id}",
            session_id=session.id,
            branch_id=str(data.get("branch_id") or branch_id),
            architecture=str(data.get("architecture") or data.get("workflow_mode") or architecture),
            agent_id=span.actor_id,
            action_kind=str(span.kind or "trace"),
            scheduler_reason=str(data.get("scheduler_reason") or span.name or ""),
            status=_step_status(span.status),
            started_at=span.started_at,
            ended_at=span.ended_at,
            duration_ms=span.duration_ms,
            task_id=str(data.get("task_id")) if data.get("task_id") else None,
            work_unit_id=str(data.get("work_unit_id")) if data.get("work_unit_id") else None,
            input_summary=_compact(span.input_ref, 240),
            output_summary=_compact(span.summary or span.output_ref, 260),
            error=span.error,
            linked_task_ids=[str(data.get("task_id"))] if data.get("task_id") else [],
            linked_artifact_refs=[str(item) for item in data.get("artifact_refs", []) if item]
            if isinstance(data.get("artifact_refs"), list)
            else [],
            source_span_id=span.id,
            projected=True,
        )
        step.summary = _step_summary_from_step(step)
        steps.append(step)

    for event in getattr(session, "events", [])[-limit:]:
        data = getattr(event, "data", {}) or {}
        task_id = data.get("task_id") or data.get("blocking_task_id")
        step = CoworkAgentStep(
            id=f"legacy-step:event:{event.id}",
            session_id=session.id,
            branch_id=str(data.get("branch_id") or branch_id),
            architecture=str(data.get("architecture") or data.get("workflow_mode") or architecture),
            agent_id=event.actor_id,
            action_kind=str(event.type or "event"),
            scheduler_reason=_event_action(event),
            status=_step_status(str(data.get("status") or event.type.rsplit(".", 1)[-1])),
            started_at=event.created_at,
            ended_at=event.created_at,
            task_id=str(task_id) if task_id else None,
            input_summary="",
            output_summary=_compact(event.message, 260),
            linked_message_ids=[str(data.get("message_id"))] if data.get("message_id") else [],
            linked_task_ids=[str(task_id)] if task_id else [],
            linked_envelope_ids=[str(data.get("envelope_id"))] if data.get("envelope_id") else [],
            source_event_id=event.id,
            projected=True,
        )
        step.summary = _step_summary_from_step(step)
        steps.append(step)

    for decision in getattr(session, "scheduler_decisions", [])[-limit:]:
        selected = decision.get("selected_agent_ids") or []
        first_agent = str(selected[0]) if selected else None
        step = CoworkAgentStep(
            id=f"legacy-step:scheduler:{decision.get('id')}",
            session_id=session.id,
            branch_id=branch_id,
            architecture=architecture,
            agent_id=first_agent,
            action_kind="scheduler_decision",
            scheduler_reason=str(decision.get("reason") or "Scheduler selected agents"),
            status="completed" if selected else "blocked",
            started_at=str(decision.get("created_at") or now_iso()),
            ended_at=str(decision.get("created_at") or now_iso()),
            input_summary=_compact(decision.get("candidate_scores", []), 260),
            output_summary=_compact(f"Selected: {', '.join(str(item) for item in selected)}", 180),
            projected=True,
        )
        step.summary = _step_summary_from_step(step)
        steps.append(step)

    return sorted(steps, key=lambda item: str(item.ended_at or item.started_at))[-limit:]


def _step_status(value: str) -> str:
    status = str(value or "").lower()
    if status in {"running", "in_progress", "pending"}:
        return "running"
    if status in {"failed", "error"}:
        return "failed"
    if status in {"blocked", "expired"}:
        return "blocked"
    if status in {"stopped", "cancelled", "canceled"}:
        return "stopped"
    return "completed"


def build_cowork_task_dag(session: CoworkSession) -> dict[str, Any]:
    """Return a task-first DAG projection for observable Cowork planning."""

    nodes: list[dict[str, Any]] = [
        {
            "id": "goal",
            "kind": "goal",
            "label": session.title,
            "title": session.title,
            "detail": _compact(session.goal, 260),
            "status": session.status,
            "tone": _status_tone(session.status),
        }
    ]
    edges: list[dict[str, Any]] = []

    for task in session.tasks.values():
        nodes.append(
            {
                "id": f"task:{task.id}",
                "entity_id": task.id,
                "kind": "task",
                "label": task.title,
                "title": task.title,
                "detail": _compact(task.result_data.get("answer") or task.result or task.description, 260),
                "status": task.status,
                "tone": _status_tone(task.status),
                "owner": task.assigned_agent_id,
                "confidence": task.confidence,
                "updated_at": task.updated_at,
            }
        )
        if task.dependencies:
            for dependency in task.dependencies:
                if dependency in session.tasks:
                    _add_edge(edges, f"task:{dependency}", f"task:{task.id}", "depends_on")
                else:
                    _add_edge(edges, "goal", f"task:{task.id}", "root")
        else:
            _add_edge(edges, "goal", f"task:{task.id}", "root")
        if task.assigned_agent_id and task.assigned_agent_id in session.agents:
            agent = session.agents[task.assigned_agent_id]
            agent_node_id = f"agent:{agent.id}"
            if not any(node["id"] == agent_node_id for node in nodes):
                nodes.append(
                    {
                        "id": agent_node_id,
                        "entity_id": agent.id,
                        "kind": "agent",
                        "label": agent.name,
                        "title": agent.name,
                        "detail": _compact(agent.role, 120),
                        "status": agent.status,
                        "tone": _status_tone(agent.status),
                    }
                )
            _add_edge(edges, agent_node_id, f"task:{task.id}", "owns")
        for artifact in _task_artifacts(task):
            artifact_id = f"artifact:{hashlib.sha1(f'{task.id}:{artifact}'.encode()).hexdigest()[:12]}"
            nodes.append(
                {
                    "id": artifact_id,
                    "kind": "artifact",
                    "label": _compact(artifact, 80),
                    "title": artifact,
                    "detail": artifact,
                    "status": "completed",
                    "tone": "completed",
                    "source_task_id": task.id,
                }
            )
            _add_edge(edges, f"task:{task.id}", artifact_id, "produced")

    for record in session.mailbox.values():
        if not record.requires_reply or record.status not in {"delivered", "read"}:
            continue
        blocked_task_id = record.blocking_task_id
        if not blocked_task_id or blocked_task_id not in session.tasks:
            continue
        blocker_id = f"blocker:{record.id}"
        nodes.append(
            {
                "id": blocker_id,
                "entity_id": record.id,
                "kind": "blocker",
                "label": record.request_type or "Reply needed",
                "title": record.request_type or "Reply needed",
                "detail": _compact(record.content, 220),
                "status": record.status,
                "tone": _status_tone(record.status),
                "sender_id": record.sender_id,
                "recipient_ids": record.recipient_ids,
            }
        )
        _add_edge(edges, blocker_id, f"task:{blocked_task_id}", "blocks")

    return {
        "nodes": nodes,
        "edges": edges,
        "stats": {
            "nodes": len(nodes),
            "edges": len(edges),
            "tasks": len(session.tasks),
            "blocked_tasks": sum(1 for node in nodes if node.get("kind") == "blocker"),
            "artifacts": sum(1 for node in nodes if node.get("kind") == "artifact"),
        },
    }


def build_cowork_artifact_index(session: CoworkSession) -> list[dict[str, Any]]:
    """Return artifacts linked back to their source tasks, work units, and agents."""

    artifacts: list[dict[str, Any]] = []
    seen: set[str] = set()
    for task in session.tasks.values():
        for value in _task_artifacts(task):
            key = _artifact_key(value)
            if key in seen:
                continue
            seen.add(key)
            artifacts.append(_artifact_record(
                session,
                artifact_id=f"artifact_{len(artifacts) + 1}",
                path_or_url=value,
                source_agent_id=task.assigned_agent_id,
                source_task_id=task.id,
                source_task_title=task.title,
                created_at=task.updated_at,
                summary=_compact(value, 160),
                confidence=task.confidence,
            )
            )
    for value in getattr(session, "artifacts", []):
        text = str(value or "").strip()
        key = _artifact_key(text)
        if not text or key in seen:
            continue
        seen.add(key)
        artifacts.append(_artifact_record(
            session,
            artifact_id=f"artifact_{len(artifacts) + 1}",
            path_or_url=text,
            source_agent_id=None,
            source_task_id=None,
            source_task_title="",
            created_at=session.updated_at,
            summary=_compact(text, 160),
            confidence=None,
        )
        )
    plan = getattr(session, "swarm_plan", {}) if isinstance(getattr(session, "swarm_plan", {}), dict) else {}
    span_by_work_unit: dict[str, str] = {}
    for span in getattr(session, "trace_spans", []):
        data = getattr(span, "data", {}) or {}
        work_unit_id = data.get("work_unit_id")
        if work_unit_id:
            span_by_work_unit[str(work_unit_id)] = span.id
    for unit in plan.get("work_units", []) if isinstance(plan.get("work_units", []), list) else []:
        if not isinstance(unit, dict):
            continue
        for artifact in unit.get("artifacts") or []:
            value = artifact.get("path_or_url") or artifact.get("path") or artifact.get("url") if isinstance(artifact, dict) else artifact
            text = str(value or "").strip()
            if not text:
                continue
            key = _artifact_key(text)
            if key in seen:
                continue
            seen.add(key)
            artifacts.append(_artifact_record(
                session,
                artifact_id=f"artifact_{len(artifacts) + 1}",
                path_or_url=text,
                source_agent_id=unit.get("assigned_agent_id"),
                source_task_id=unit.get("source_task_id"),
                source_work_unit_id=unit.get("id"),
                source_span_id=span_by_work_unit.get(str(unit.get("id") or "")),
                source_task_title=unit.get("title", ""),
                created_at=unit.get("updated_at") or session.updated_at,
                summary=artifact.get("summary") if isinstance(artifact, dict) else _compact(text, 160),
                confidence=unit.get("confidence"),
                kind=(artifact.get("kind") if isinstance(artifact, dict) else "") or "",
            )
            )
    return artifacts


def build_cowork_large_swarm_summary(session: CoworkSession) -> dict[str, Any]:
    plan = getattr(session, "swarm_plan", {}) if isinstance(getattr(session, "swarm_plan", {}), dict) else {}
    units = [unit for unit in plan.get("work_units", []) if isinstance(unit, dict)]
    status_counts = Counter(str(unit.get("status") or "unknown") for unit in units)
    workstreams: dict[str, dict[str, Any]] = {}
    for unit in units:
        key = str(unit.get("team_id") or unit.get("fanout_group_id") or unit.get("kind") or "default")
        group = workstreams.setdefault(
            key,
            {"id": key, "title": key.replace("_", " "), "count": 0, "status_counts": Counter(), "sample_unit_ids": []},
        )
        group["count"] += 1
        group["status_counts"][str(unit.get("status") or "unknown")] += 1
        if len(group["sample_unit_ids"]) < 8:
            group["sample_unit_ids"].append(unit.get("id"))
    return {
        "schema_version": "cowork.large_swarm.v1",
        "enabled": len(units) >= 40,
        "total_work_units": len(units),
        "status_counts": dict(status_counts),
        "workstreams": [
            {**group, "status_counts": dict(group["status_counts"])}
            for group in sorted(workstreams.values(), key=lambda item: item["count"], reverse=True)
        ],
        "render_limit": 60,
        "generated_at": now_iso(),
    }


def build_cowork_swarm_organization(session: CoworkSession) -> dict[str, Any]:
    """Return the grouped swarm organization projection for snapshots and UI."""

    plan = getattr(session, "swarm_plan", {}) if isinstance(getattr(session, "swarm_plan", {}), dict) else {}
    units = [unit for unit in plan.get("work_units", []) if isinstance(unit, dict)]
    main_units = [unit for unit in units if unit.get("kind") not in {"reducer", "reviewer"}]
    metrics = build_swarm_parallel_metrics(session) if getattr(session, "workflow_mode", "") == "swarm" else {}
    workstreams = _swarm_workstream_groups(main_units, metrics)
    return {
        "schema_version": "cowork.swarm_organization.v1",
        "generated_at": now_iso(),
        "plan_id": plan.get("id", ""),
        "plan_status": plan.get("status", ""),
        "enabled": bool(units) and (len(units) >= 20 or len(workstreams) > 1),
        "total_work_units": len(units),
        "workstreams": workstreams,
        "grouped_counts": {
            "workstreams": len(workstreams),
            "work_units": len(main_units),
            "gates": sum(1 for unit in units if unit.get("kind") in {"reducer", "reviewer"}),
            "agents": len({str(unit.get("assigned_agent_id")) for unit in units if unit.get("assigned_agent_id")}),
        },
        "gates": _swarm_gate_summary(plan, units, session),
        "metrics": metrics,
        "blockers": _swarm_blocker_summaries(units, session),
    }


def _swarm_workstream_groups(units: list[dict[str, Any]], metrics: dict[str, Any]) -> list[dict[str, Any]]:
    groups: dict[str, dict[str, Any]] = {}
    critical_depth = int(metrics.get("critical_path_depth", 0) or 0)
    for unit in units:
        group_id = _swarm_workstream_id(unit)
        group = groups.setdefault(
            group_id,
            {
                "id": group_id,
                "title": _swarm_workstream_title(group_id, unit),
                "status": "pending",
                "unit_counts": Counter(),
                "agent_ids": set(),
                "critical": False,
                "coverage": 0.0,
                "risk": "low",
                "blockers": [],
                "sample_unit_ids": [],
            },
        )
        status = str(unit.get("status") or "unknown")
        group["unit_counts"][status] += 1
        if unit.get("assigned_agent_id"):
            group["agent_ids"].add(str(unit.get("assigned_agent_id")))
        if len(group["sample_unit_ids"]) < 8:
            group["sample_unit_ids"].append(unit.get("id"))
        blockers = _work_unit_blockers(unit)
        if blockers:
            group["blockers"].append({"work_unit_id": unit.get("id", ""), "blocked_by": blockers, "status": status})
        if status in {"failed", "blocked", "needs_revision"}:
            group["critical"] = True

    for group in groups.values():
        counts = group["unit_counts"]
        total = sum(counts.values())
        completed = counts.get("completed", 0) + counts.get("skipped", 0)
        blocked = counts.get("failed", 0) + counts.get("blocked", 0) + counts.get("needs_revision", 0)
        running = counts.get("in_progress", 0)
        if blocked:
            status = "blocked"
            risk = "high"
        elif running:
            status = "active"
            risk = "medium" if group["blockers"] else "low"
        elif total and completed == total:
            status = "completed"
            risk = "low"
        else:
            status = "pending"
            risk = "medium" if group["blockers"] else "low"
        group["status"] = status
        group["risk"] = risk
        group["coverage"] = round(completed / max(1, total), 3)
        group["critical"] = bool(group["critical"] or (critical_depth > 0 and blocked))
        group["unit_counts"] = dict(sorted(counts.items()))
        group["agent_ids"] = sorted(group["agent_ids"])

    return sorted(groups.values(), key=lambda item: (-sum(item["unit_counts"].values()), item["id"]))


def _swarm_gate_summary(plan: dict[str, Any], units: list[dict[str, Any]], session: CoworkSession) -> dict[str, Any]:
    reducer_units = [unit for unit in units if unit.get("kind") == "reducer"]
    reviewer_units = [unit for unit in units if unit.get("kind") == "reviewer"]
    evaluations = [
        item
        for item in ((getattr(session, "runtime_state", {}) or {}).get("swarm_evaluations", []))
        if isinstance(item, dict)
    ]
    blocking_evaluations = [item for item in evaluations if item.get("status") in {"block", "error"}]
    completion = getattr(session, "completion_decision", {}) if isinstance(getattr(session, "completion_decision", {}), dict) else {}
    return {
        "reducer": _swarm_gate("reducer", reducer_units, plan.get("reducer", {})),
        "reviewer": _swarm_gate("reviewer", reviewer_units, plan.get("review", {})),
        "evaluations": {
            "status": "blocked" if blocking_evaluations else "pass" if evaluations else "not_ready",
            "total": len(evaluations),
            "blocking": len(blocking_evaluations),
            "blocking_ids": [item.get("id", "") for item in blocking_evaluations],
        },
        "final_deliverable": {
            "status": "ready" if completion.get("ready_to_finish") else "not_ready",
            "next_action": completion.get("next_action", ""),
            "reason": completion.get("reason", ""),
        },
    }


def _swarm_gate(kind: str, units: list[dict[str, Any]], config: Any) -> dict[str, Any]:
    unit = units[-1] if units else {}
    data = config if isinstance(config, dict) else {}
    return {
        "status": unit.get("status") or ("pending" if data.get("required") else "not_ready"),
        "required": bool(data.get("required") or kind == "reducer"),
        "agent_id": unit.get("assigned_agent_id") or data.get("agent_id") or "",
        "work_unit_id": unit.get("id", ""),
        "source_work_unit_ids": unit.get("source_work_unit_ids", []),
        "source_artifact_refs": unit.get("source_artifact_refs", []),
        "coverage_by_workstream": unit.get("coverage_by_workstream", {}),
        "confidence_by_section": unit.get("confidence_by_section", {}),
    }


def _swarm_blocker_summaries(units: list[dict[str, Any]], session: CoworkSession) -> list[dict[str, Any]]:
    completed = {
        str(unit.get("id"))
        for unit in units
        if unit.get("status") in {"completed", "skipped", "cancelled"} and unit.get("id")
    }
    completed.update(
        task_id
        for task_id, task in (getattr(session, "tasks", {}) or {}).items()
        if getattr(task, "status", "") in {"completed", "skipped"}
    )
    blockers: list[dict[str, Any]] = []
    for unit in units:
        blocked_by = [dep for dep in _work_unit_blockers(unit) if dep not in completed]
        status = str(unit.get("status") or "")
        if not blocked_by and status not in {"failed", "blocked", "needs_revision"}:
            continue
        blockers.append(
            {
                "work_unit_id": unit.get("id", ""),
                "title": unit.get("title", ""),
                "workstream_id": _swarm_workstream_id(unit),
                "status": status,
                "blocked_by": blocked_by,
                "error": unit.get("error", ""),
            }
        )
    return blockers[:40]


def _swarm_workstream_id(unit: dict[str, Any]) -> str:
    for key in ("workstream_id", "workstream", "fanout_group_id", "team_id", "source_kind", "kind"):
        value = str(unit.get(key) or "").strip()
        if value:
            return value
    source_task_id = str(unit.get("source_task_id") or "").strip()
    return source_task_id or "default"


def _swarm_workstream_title(group_id: str, unit: dict[str, Any]) -> str:
    title = str(unit.get("workstream_title") or unit.get("fanout_group_title") or "").strip()
    return title or group_id.replace("_", " ").replace("-", " ").title()


def _work_unit_blockers(unit: dict[str, Any]) -> list[str]:
    blockers = unit.get("blocked_by") if isinstance(unit.get("blocked_by"), list) else []
    dependencies = unit.get("dependencies") if isinstance(unit.get("dependencies"), list) else []
    return [str(item) for item in [*blockers, *dependencies] if str(item or "").strip()]


def _artifact_key(path_or_url: str) -> str:
    return hashlib.sha1(str(path_or_url or "").strip().lower().encode("utf-8")).hexdigest()


def _artifact_record(
    session: CoworkSession,
    *,
    artifact_id: str,
    path_or_url: str,
    source_agent_id: str | None,
    source_task_id: str | None,
    source_task_title: str,
    created_at: str,
    summary: str,
    confidence: float | None,
    kind: str = "",
    source_work_unit_id: str | None = None,
    source_span_id: str | None = None,
) -> dict[str, Any]:
    status, file_hash, mime_type = _artifact_file_metadata(session, path_or_url)
    return {
        "id": artifact_id,
        "kind": kind or _artifact_kind(path_or_url),
        "path_or_url": path_or_url,
        "source_agent_id": source_agent_id,
        "source_task_id": source_task_id,
        "source_work_unit_id": source_work_unit_id,
        "source_span_id": source_span_id,
        "source_task_title": source_task_title,
        "created_at": created_at,
        "summary": summary,
        "hash": file_hash,
        "mime_type": mime_type,
        "confidence": confidence,
        "status": status,
        "missing": status == "missing",
    }


def _artifact_file_metadata(session: CoworkSession, path_or_url: str) -> tuple[str, str, str]:
    value = str(path_or_url or "").strip()
    if not value or value.startswith(("http://", "https://")):
        return ("linked" if value else "unresolved", "", mimetypes.guess_type(value)[0] or "")
    path = Path(value)
    if not path.is_absolute() and getattr(session, "workspace_dir", ""):
        path = Path(session.workspace_dir) / path
    mime_type = mimetypes.guess_type(value)[0] or ""
    if not path.exists() or not path.is_file():
        return "missing", "", mime_type
    try:
        digest = hashlib.sha256()
        with path.open("rb") as handle:
            for chunk in iter(lambda: handle.read(1024 * 1024), b""):
                digest.update(chunk)
        return "available", digest.hexdigest(), mime_type
    except Exception:
        return "unresolved", "", mime_type


def _task_artifacts(task: Any) -> list[str]:
    data = getattr(task, "result_data", {}) or {}
    values: list[Any] = []
    for key in ("artifacts", "artifact_paths", "generated_files", "files", "paths"):
        raw = data.get(key)
        if isinstance(raw, list):
            values.extend(raw)
        elif isinstance(raw, str):
            values.append(raw)
    return [str(value).strip() for value in values if str(value or "").strip()]


def _artifact_kind(value: str) -> str:
    text = value.lower()
    if text.startswith(("http://", "https://")):
        return "url"
    if text.endswith((".png", ".jpg", ".jpeg", ".webp", ".gif", ".svg")):
        return "image"
    if text.endswith((".csv", ".tsv", ".xlsx", ".xls")):
        return "table"
    if text.endswith((".patch", ".diff")):
        return "diff"
    return "file"
