Agenthub

#!/usr/bin/env python3
"""Manage agent task boards with status tracking for multi-agent workflows.

Provides a visual dashboard of agent states within an orchestration session.
Supports multiple view modes: board (kanban), timeline, summary.

Usage:
    python board_manager.py --session session.json --view board
    python board_manager.py --session session.json --view timeline
    python board_manager.py --session session.json --agent researcher --detail
    python board_manager.py --session session.json --json

Expected session.json format:
{
    "session_id": "abc123",
    "workflow_name": "market-analysis",
    "started_at": "2026-04-02T10:30:00",
    "agents": {
        "researcher": {"state": "COMPLETED", "started_at": "...", "completed_at": "...", "duration_s": 90, "outputs": {...}},
        "data_collector": {"state": "RUNNING", "started_at": "...", "duration_s": null},
        "analyst": {"state": "PENDING", "dependencies": ["data_collector", "researcher"]},
        "writer": {"state": "PENDING", "dependencies": ["analyst"]}
    }
}
"""

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path


STATE_SYMBOLS = {
    "COMPLETED": "[x]",
    "RUNNING": "[>]",
    "PENDING": "[ ]",
    "FAILED": "[!]",
    "SKIPPED": "[-]",
    "READY": "[~]",
    "EVALUATING": "[?]",
}

STATE_ORDER = {
    "RUNNING": 0,
    "READY": 1,
    "EVALUATING": 2,
    "PENDING": 3,
    "COMPLETED": 4,
    "FAILED": 5,
    "SKIPPED": 6,
}


def load_session(path):
    """Load session state from JSON file."""
    try:
        with open(path, "r") as f:
            return json.load(f)
    except (json.JSONDecodeError, OSError) as e:
        print(f"Error loading session: {e}", file=sys.stderr)
        sys.exit(1)


def compute_elapsed(session):
    """Compute elapsed time since session start."""
    started = session.get("started_at", "")
    if not started:
        return 0
    try:
        start_dt = datetime.fromisoformat(started)
        return (datetime.now() - start_dt).total_seconds()
    except ValueError:
        return 0


def format_duration(seconds):
    """Format seconds as Xm Ys."""
    if seconds is None or seconds == 0:
        return "---"
    minutes = int(seconds // 60)
    secs = int(seconds % 60)
    if minutes > 0:
        return f"{minutes}m {secs:02d}s"
    return f"{secs}s"


def group_by_state(agents):
    """Group agents by their current state."""
    groups = {}
    for agent_id, agent_data in agents.items():
        state = agent_data.get("state", "PENDING")
        if state not in groups:
            groups[state] = []
        groups[state].append({"id": agent_id, **agent_data})
    return groups


def render_board_view(session):
    """Render kanban-style board view."""
    agents = session.get("agents", {})
    groups = group_by_state(agents)
    elapsed = compute_elapsed(session)

    lines = []
    lines.append(f"AgentHub Board - Session: {session.get('session_id', '?')}")
    lines.append(f"Workflow: {session.get('workflow_name', '?')}")
    lines.append(f"Started: {session.get('started_at', '?')[:19]} | Elapsed: {format_duration(elapsed)}")
    lines.append("")

    # Column layout
    column_order = ["RUNNING", "READY", "EVALUATING", "PENDING", "COMPLETED", "FAILED", "SKIPPED"]
    active_columns = [s for s in column_order if s in groups]

    if not active_columns:
        lines.append("  No agents in session.")
        return "\n".join(lines)

    # Header
    col_width = 22
    header = ""
    for state in active_columns:
        count = len(groups[state])
        header += f"{state} ({count})".ljust(col_width)
    lines.append(header)
    lines.append("─" * (col_width * len(active_columns)))

    # Find max rows
    max_rows = max(len(groups[s]) for s in active_columns)

    for row in range(max_rows):
        row_line = ""
        for state in active_columns:
            if row < len(groups[state]):
                agent = groups[state][row]
                symbol = STATE_SYMBOLS.get(state, "[?]")
                duration = format_duration(agent.get("duration_s"))
                entry = f"{symbol} {agent['id'][:14]}"
                if state in ("COMPLETED", "RUNNING"):
                    entry += f" {duration}"
                row_line += entry.ljust(col_width)
            else:
                row_line += " " * col_width
        lines.append(row_line)

    # Show pending dependencies
    pending = groups.get("PENDING", [])
    if pending:
        lines.append("")
        lines.append("WAITING ON:")
        for agent in pending:
            deps = agent.get("dependencies", [])
            if deps:
                lines.append(f"  {agent['id']} <- {', '.join(deps)}")

    return "\n".join(lines)


def render_timeline_view(session):
    """Render timeline/gantt-style view."""
    agents = session.get("agents", {})
    elapsed = compute_elapsed(session)

    lines = []
    lines.append(f"AgentHub Timeline - Session: {session.get('session_id', '?')}")
    lines.append(f"Elapsed: {format_duration(elapsed)}")
    lines.append("")

    # Sort by start time, then by state
    sorted_agents = sorted(
        agents.items(),
        key=lambda x: (
            STATE_ORDER.get(x[1].get("state", "PENDING"), 9),
            x[1].get("started_at", "z"),
        ),
    )

    max_name_len = max(len(aid) for aid, _ in sorted_agents) if sorted_agents else 10
    bar_width = 40

    for agent_id, agent_data in sorted_agents:
        state = agent_data.get("state", "PENDING")
        duration = agent_data.get("duration_s", 0) or 0
        symbol = STATE_SYMBOLS.get(state, "[?]")

        # Proportional bar
        if elapsed > 0 and duration > 0:
            bar_len = max(1, int(duration / max(elapsed, 1) * bar_width))
        elif state == "RUNNING":
            bar_len = max(1, int(bar_width * 0.3))
        else:
            bar_len = 0

        bar_char = {"COMPLETED": "=", "RUNNING": ">", "FAILED": "x", "PENDING": ".", "READY": "~"}
        char = bar_char.get(state, " ")
        bar = char * bar_len

        name = agent_id.ljust(max_name_len)
        dur_str = format_duration(duration if duration else None)
        lines.append(f"  {symbol} {name}  [{bar:<{bar_width}}]  {dur_str}")

    return "\n".join(lines)


def render_summary_view(session):
    """Render compact one-line-per-agent summary."""
    agents = session.get("agents", {})
    elapsed = compute_elapsed(session)

    total = len(agents)
    completed = sum(1 for a in agents.values() if a.get("state") == "COMPLETED")
    failed = sum(1 for a in agents.values() if a.get("state") == "FAILED")
    running = sum(1 for a in agents.values() if a.get("state") == "RUNNING")

    progress_pct = int(completed / total * 100) if total else 0
    bar_len = int(progress_pct / 5)
    progress_bar = "█" * bar_len + "░" * (20 - bar_len)

    lines = []
    lines.append(f"Session: {session.get('session_id', '?')} | {session.get('workflow_name', '?')}")
    lines.append(f"Progress: [{progress_bar}] {progress_pct}% ({completed}/{total})")
    lines.append(f"Running: {running} | Failed: {failed} | Elapsed: {format_duration(elapsed)}")
    lines.append("")
    lines.append(f"{'Agent':<20} {'State':<12} {'Duration':<10}")
    lines.append("─" * 42)

    for agent_id, agent_data in sorted(agents.items(), key=lambda x: STATE_ORDER.get(x[1].get("state", "PENDING"), 9)):
        state = agent_data.get("state", "PENDING")
        duration = format_duration(agent_data.get("duration_s"))
        lines.append(f"{agent_id:<20} {state:<12} {duration:<10}")

    return "\n".join(lines)


def render_agent_detail(session, agent_id):
    """Render detailed view for a specific agent."""
    agents = session.get("agents", {})
    if agent_id not in agents:
        return f"Agent '{agent_id}' not found in session."

    agent = agents[agent_id]
    lines = []
    lines.append(f"Agent Detail: {agent_id}")
    lines.append("=" * 50)
    lines.append(f"  State:        {agent.get('state', 'PENDING')}")
    lines.append(f"  Started:      {agent.get('started_at', '---')}")
    lines.append(f"  Completed:    {agent.get('completed_at', '---')}")
    lines.append(f"  Duration:     {format_duration(agent.get('duration_s'))}")
    lines.append(f"  Dependencies: {', '.join(agent.get('dependencies', [])) or 'none'}")
    lines.append(f"  Retries:      {agent.get('retries', 0)}")

    if agent.get("task"):
        lines.append(f"  Task:         {agent['task'][:60]}")
    if agent.get("error"):
        lines.append(f"  Error:        {agent['error'][:80]}")
    if agent.get("outputs"):
        lines.append(f"  Outputs:      {json.dumps(agent['outputs'], indent=2)[:200]}")
    if agent.get("eval_score") is not None:
        lines.append(f"  Eval score:   {agent['eval_score']}")

    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Manage agent task boards with status tracking.",
        epilog="Example: python board_manager.py --session session.json --view board",
    )
    parser.add_argument("--session", required=True, help="Path to session state JSON file")
    parser.add_argument("--view", choices=["board", "timeline", "summary"], default="board", help="View mode")
    parser.add_argument("--agent", help="Show detail for a specific agent")
    parser.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON")

    args = parser.parse_args()

    session_path = Path(args.session)
    if not session_path.exists():
        print(f"Error: Session file '{args.session}' not found.", file=sys.stderr)
        sys.exit(1)

    session = load_session(session_path)

    if args.json_output:
        agents = session.get("agents", {})
        result = {
            "session_id": session.get("session_id"),
            "workflow_name": session.get("workflow_name"),
            "elapsed_s": compute_elapsed(session),
            "total_agents": len(agents),
            "by_state": {},
            "agents": agents,
        }
        for state in set(a.get("state", "PENDING") for a in agents.values()):
            result["by_state"][state] = sum(1 for a in agents.values() if a.get("state") == state)
        print(json.dumps(result, indent=2))
    elif args.agent:
        print(render_agent_detail(session, args.agent))
    elif args.view == "board":
        print(render_board_view(session))
    elif args.view == "timeline":
        print(render_timeline_view(session))
    elif args.view == "summary":
        print(render_summary_view(session))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Analyze multi-agent DAG workflow definitions for structural issues.

Validates workflow definitions by checking for cycles, unreachable nodes,
missing input/output references, and bottlenecks. Also computes critical
path length and parallelization potential.

Usage:
    python dag_analyzer.py --workflow workflow.json --validate
    python dag_analyzer.py --workflow workflow.json --critical-path
    python dag_analyzer.py --workflow workflow.json --visualize
    python dag_analyzer.py --workflow workflow.json --json
"""

import argparse
import json
import sys
from collections import defaultdict, deque
from pathlib import Path


def load_workflow(path):
    """Load workflow definition from JSON file."""
    try:
        with open(path, "r") as f:
            return json.load(f)
    except (json.JSONDecodeError, OSError) as e:
        print(f"Error loading workflow: {e}", file=sys.stderr)
        sys.exit(1)


def build_graph(workflow):
    """Build adjacency list and reverse graph from workflow agents."""
    agents = workflow.get("agents", {})
    graph = defaultdict(list)      # agent -> list of dependents
    reverse = defaultdict(list)    # agent -> list of dependencies
    all_nodes = set(agents.keys())

    for agent_id, agent_def in agents.items():
        deps = agent_def.get("dependencies", [])
        for dep in deps:
            graph[dep].append(agent_id)
            reverse[agent_id].append(dep)

    return graph, reverse, all_nodes


def detect_cycles(graph, all_nodes):
    """Detect cycles using DFS coloring (white/gray/black)."""
    WHITE, GRAY, BLACK = 0, 1, 2
    color = {n: WHITE for n in all_nodes}
    cycles = []

    def dfs(node, path):
        color[node] = GRAY
        path.append(node)
        for neighbor in graph.get(node, []):
            if color[neighbor] == GRAY:
                # Found a cycle
                cycle_start = path.index(neighbor)
                cycles.append(path[cycle_start:] + [neighbor])
            elif color[neighbor] == WHITE:
                dfs(neighbor, path)
        path.pop()
        color[node] = BLACK

    for node in all_nodes:
        if color[node] == WHITE:
            dfs(node, [])

    return cycles


def topological_sort(graph, reverse, all_nodes):
    """Kahn's algorithm for topological sort."""
    in_degree = {n: 0 for n in all_nodes}
    for node in all_nodes:
        for dep in reverse.get(node, []):
            in_degree[node] = in_degree.get(node, 0)
        in_degree[node] = len(reverse.get(node, []))

    queue = deque([n for n in all_nodes if in_degree[n] == 0])
    order = []

    while queue:
        node = queue.popleft()
        order.append(node)
        for dependent in graph.get(node, []):
            in_degree[dependent] -= 1
            if in_degree[dependent] == 0:
                queue.append(dependent)

    if len(order) != len(all_nodes):
        return None  # Cycle exists
    return order


def find_roots_and_terminals(graph, reverse, all_nodes):
    """Identify root nodes (no deps) and terminal nodes (no dependents)."""
    roots = [n for n in all_nodes if not reverse.get(n)]
    terminals = [n for n in all_nodes if not graph.get(n)]
    return sorted(roots), sorted(terminals)


def compute_critical_path(workflow, graph, reverse, all_nodes):
    """Compute the critical path (longest path through the DAG)."""
    agents = workflow.get("agents", {})
    config = workflow.get("config", {})
    default_timeout = config.get("timeout_per_agent", 300)

    # Estimate duration per agent (use timeout as upper bound)
    durations = {}
    for agent_id in all_nodes:
        agent_config = agents.get(agent_id, {}).get("config", {})
        durations[agent_id] = agent_config.get("timeout", default_timeout)

    # Compute longest path from each node
    longest_to = {n: 0 for n in all_nodes}
    predecessor = {n: None for n in all_nodes}

    topo = topological_sort(graph, reverse, all_nodes)
    if topo is None:
        return None, 0  # Cycle

    for node in topo:
        for dependent in graph.get(node, []):
            new_dist = longest_to[node] + durations[node]
            if new_dist > longest_to[dependent]:
                longest_to[dependent] = new_dist
                predecessor[dependent] = node

    # Find the terminal with the longest path
    terminals = [n for n in all_nodes if not graph.get(n)]
    if not terminals:
        return [], 0

    end_node = max(terminals, key=lambda n: longest_to[n] + durations[n])
    total_time = longest_to[end_node] + durations[end_node]

    # Reconstruct path
    path = [end_node]
    current = end_node
    while predecessor[current] is not None:
        current = predecessor[current]
        path.append(current)
    path.reverse()

    return path, total_time


def check_io_references(workflow):
    """Verify all input references resolve to upstream outputs."""
    agents = workflow.get("agents", {})
    issues = []

    # Map each agent to its outputs
    output_map = {}
    for agent_id, agent_def in agents.items():
        for output in agent_def.get("outputs", []):
            output_map[output] = agent_id

    # Check inputs
    for agent_id, agent_def in agents.items():
        deps = set(agent_def.get("dependencies", []))
        for inp in agent_def.get("inputs", []):
            if inp in output_map:
                producer = output_map[inp]
                # Check that the producer is an upstream dependency
                if producer not in deps and producer != agent_id:
                    issues.append({
                        "type": "missing_dependency",
                        "agent": agent_id,
                        "input": inp,
                        "producer": producer,
                        "message": f"Agent '{agent_id}' uses input '{inp}' from '{producer}' but doesn't list it as a dependency",
                    })
            # Inputs might be workflow-level inputs (not produced by agents)

    # Check for unused outputs
    all_inputs = set()
    for agent_def in agents.values():
        all_inputs.update(agent_def.get("inputs", []))

    for output_name, producer in output_map.items():
        if output_name not in all_inputs:
            # Terminal output -- expected
            if not any(
                producer in agents[a].get("dependencies", [])
                for a in agents
            ):
                pass  # Terminal agent output, this is fine

    return issues


def compute_parallel_groups(graph, reverse, all_nodes):
    """Compute which agents can run in parallel at each level."""
    topo = topological_sort(graph, reverse, all_nodes)
    if topo is None:
        return []

    # Compute level (longest path from any root to this node)
    level = {n: 0 for n in all_nodes}
    for node in topo:
        for dependent in graph.get(node, []):
            level[dependent] = max(level[dependent], level[node] + 1)

    # Group by level
    groups = defaultdict(list)
    for node in all_nodes:
        groups[level[node]].append(node)

    return [{"level": lvl, "agents": sorted(agents)} for lvl, agents in sorted(groups.items())]


def validate_workflow(workflow):
    """Run all validation checks on a workflow definition."""
    agents = workflow.get("agents", {})
    if not agents:
        return {"valid": False, "errors": ["No agents defined in workflow"]}

    graph, reverse, all_nodes = build_graph(workflow)
    errors = []
    warnings = []

    # Check for unknown dependencies
    for agent_id, agent_def in agents.items():
        for dep in agent_def.get("dependencies", []):
            if dep not in all_nodes:
                errors.append(f"Agent '{agent_id}' depends on unknown agent '{dep}'")

    # Cycle detection
    cycles = detect_cycles(graph, all_nodes)
    for cycle in cycles:
        errors.append(f"Cycle detected: {' -> '.join(cycle)}")

    # IO reference check
    io_issues = check_io_references(workflow)
    for issue in io_issues:
        warnings.append(issue["message"])

    # Root/terminal check
    roots, terminals = find_roots_and_terminals(graph, reverse, all_nodes)
    if not roots:
        errors.append("No root agents found (all agents have dependencies)")
    if not terminals:
        warnings.append("No terminal agents found (all agents have dependents)")

    # Unreachable check
    if roots and not cycles:
        reachable = set()
        queue = deque(roots)
        while queue:
            node = queue.popleft()
            if node in reachable:
                continue
            reachable.add(node)
            queue.extend(graph.get(node, []))
        unreachable = all_nodes - reachable
        for node in unreachable:
            warnings.append(f"Agent '{node}' is unreachable from root agents")

    # Critical path
    crit_path, crit_time = compute_critical_path(workflow, graph, reverse, all_nodes)
    parallel_groups = compute_parallel_groups(graph, reverse, all_nodes)

    return {
        "valid": len(errors) == 0,
        "errors": errors,
        "warnings": warnings,
        "stats": {
            "total_agents": len(all_nodes),
            "root_agents": roots,
            "terminal_agents": terminals,
            "max_parallel": max(len(g["agents"]) for g in parallel_groups) if parallel_groups else 0,
            "depth": len(parallel_groups),
            "critical_path": crit_path if crit_path else [],
            "critical_path_time_s": crit_time,
        },
        "parallel_groups": parallel_groups,
    }


def visualize_dag(workflow):
    """Generate a text-based DAG visualization."""
    graph, reverse, all_nodes = build_graph(workflow)
    groups = compute_parallel_groups(graph, reverse, all_nodes)
    lines = []

    for group in groups:
        level_agents = group["agents"]
        level_line = "  |  ".join(f"[{a}]" for a in level_agents)
        lines.append(f"Level {group['level']}: {level_line}")

        # Show edges
        for agent in level_agents:
            dependents = graph.get(agent, [])
            for dep in dependents:
                lines.append(f"  {agent} --> {dep}")

    return "\n".join(lines)


def format_human(result, visualization=None):
    """Format validation result for human output."""
    output = []
    output.append("=" * 60)
    output.append("DAG WORKFLOW ANALYZER")
    output.append("=" * 60)

    if not result["valid"]:
        output.append("")
        output.append("VALIDATION: FAILED")
        output.append("-" * 60)
        for err in result["errors"]:
            output.append(f"  [ERROR] {err}")
    else:
        output.append("")
        output.append("VALIDATION: PASSED")

    if result["warnings"]:
        output.append("")
        output.append("WARNINGS")
        output.append("-" * 60)
        for warn in result["warnings"]:
            output.append(f"  [WARN] {warn}")

    stats = result["stats"]
    output.append("")
    output.append("STATISTICS")
    output.append("-" * 60)
    output.append(f"  Total agents:     {stats['total_agents']}")
    output.append(f"  Root agents:      {', '.join(stats['root_agents'])}")
    output.append(f"  Terminal agents:  {', '.join(stats['terminal_agents'])}")
    output.append(f"  Max parallelism:  {stats['max_parallel']}")
    output.append(f"  DAG depth:        {stats['depth']} levels")
    if stats["critical_path"]:
        output.append(f"  Critical path:    {' -> '.join(stats['critical_path'])}")
        output.append(f"  Est. time:        {stats['critical_path_time_s']}s")

    if result["parallel_groups"]:
        output.append("")
        output.append("EXECUTION GROUPS")
        output.append("-" * 60)
        for group in result["parallel_groups"]:
            agents_str = ", ".join(group["agents"])
            output.append(f"  Level {group['level']}: [{agents_str}]  ({len(group['agents'])} parallel)")

    if visualization:
        output.append("")
        output.append("DAG VISUALIZATION")
        output.append("-" * 60)
        output.append(visualization)

    return "\n".join(output)


def main():
    parser = argparse.ArgumentParser(
        description="Analyze multi-agent DAG workflow definitions.",
        epilog="Example: python dag_analyzer.py --workflow workflow.json --validate",
    )
    parser.add_argument("--workflow", required=True, help="Path to workflow JSON file")
    parser.add_argument("--validate", action="store_true", help="Run full validation")
    parser.add_argument("--critical-path", action="store_true", help="Show critical path only")
    parser.add_argument("--visualize", action="store_true", help="Show DAG visualization")
    parser.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON")

    args = parser.parse_args()

    wf_path = Path(args.workflow)
    if not wf_path.exists():
        print(f"Error: Workflow file '{args.workflow}' not found.", file=sys.stderr)
        sys.exit(1)

    workflow = load_workflow(wf_path)
    result = validate_workflow(workflow)
    visualization = visualize_dag(workflow) if args.visualize else None

    if args.json_output:
        output = result
        if visualization:
            output["visualization"] = visualization
        print(json.dumps(output, indent=2))
    else:
        print(format_human(result, visualization))

    # Exit with error code if invalid
    if not result["valid"]:
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Rank and merge outputs from multiple agents in a workflow.

Scores agent outputs on completeness, length, structure, and relevance.
Supports different merge strategies: synthesize (combine complementary),
rank-select (pick best), and chain (use final).

Usage:
    python result_ranker.py --session session.json --list-outputs
    python result_ranker.py --session session.json --rank
    python result_ranker.py --session session.json --merge synthesize
    python result_ranker.py --session session.json --merge rank-select --json
"""

import argparse
import json
import re
import sys
from collections import defaultdict
from pathlib import Path


def load_session(path):
    """Load session with agent outputs."""
    try:
        with open(path, "r") as f:
            return json.load(f)
    except (json.JSONDecodeError, OSError) as e:
        print(f"Error loading session: {e}", file=sys.stderr)
        sys.exit(1)


def extract_outputs(session):
    """Extract completed agent outputs from session."""
    agents = session.get("agents", {})
    outputs = []
    for agent_id, agent_data in agents.items():
        if agent_data.get("state") == "COMPLETED" and agent_data.get("outputs"):
            outputs.append({
                "agent_id": agent_id,
                "state": agent_data["state"],
                "outputs": agent_data["outputs"],
                "duration_s": agent_data.get("duration_s", 0),
                "eval_score": agent_data.get("eval_score"),
                "task": agent_data.get("task", ""),
            })
    return outputs


def score_output(output):
    """Score an individual agent output on quality dimensions."""
    scores = {}
    output_data = output.get("outputs", {})

    # Completeness: fraction of output fields that are non-empty
    total_fields = len(output_data)
    non_empty = sum(1 for v in output_data.values() if v)
    scores["completeness"] = non_empty / total_fields if total_fields > 0 else 0

    # Depth: total content length across all output fields
    total_length = 0
    for value in output_data.values():
        if isinstance(value, str):
            total_length += len(value)
        elif isinstance(value, (list, dict)):
            total_length += len(json.dumps(value))
    # Normalize: 500 chars is minimal, 5000 is good, cap at 1.0
    scores["depth"] = min(1.0, total_length / 5000) if total_length > 0 else 0

    # Structure: presence of structured data (lists, dicts, headers)
    has_structure = 0
    for value in output_data.values():
        if isinstance(value, (list, dict)):
            has_structure += 1
        elif isinstance(value, str):
            if re.search(r"^#+\s|\n-\s|\n\d+\.\s", value):
                has_structure += 1
    scores["structure"] = has_structure / total_fields if total_fields > 0 else 0

    # Use existing eval score if available
    if output.get("eval_score") is not None:
        scores["eval"] = output["eval_score"]
    else:
        scores["eval"] = None

    # Composite score
    weights = {"completeness": 0.3, "depth": 0.3, "structure": 0.2}
    weighted_sum = sum(scores[k] * w for k, w in weights.items())
    total_weight = sum(weights.values())

    if scores["eval"] is not None:
        weighted_sum += scores["eval"] * 0.2
        total_weight += 0.2

    scores["composite"] = round(weighted_sum / total_weight, 3) if total_weight > 0 else 0

    return scores


def rank_outputs(outputs):
    """Rank outputs by quality score."""
    ranked = []
    for output in outputs:
        scores = score_output(output)
        ranked.append({
            **output,
            "scores": scores,
        })
    ranked.sort(key=lambda x: -x["scores"]["composite"])
    return ranked


def merge_synthesize(ranked_outputs):
    """Synthesize complementary outputs into a unified result."""
    sections = {}
    for output in ranked_outputs:
        agent_id = output["agent_id"]
        task = output.get("task", agent_id)
        output_data = output.get("outputs", {})

        for key, value in output_data.items():
            section_key = key
            if section_key not in sections:
                sections[section_key] = {
                    "content": value,
                    "source": agent_id,
                    "score": output["scores"]["composite"],
                }
            else:
                # If existing has lower score, prefer new
                if output["scores"]["composite"] > sections[section_key]["score"]:
                    sections[section_key] = {
                        "content": value,
                        "source": agent_id,
                        "score": output["scores"]["composite"],
                    }

    merged = {}
    attribution = {}
    for key, section in sections.items():
        merged[key] = section["content"]
        attribution[key] = section["source"]

    return {
        "strategy": "synthesize",
        "merged_output": merged,
        "attribution": attribution,
        "sections_count": len(merged),
        "sources_count": len(set(attribution.values())),
    }


def merge_rank_select(ranked_outputs):
    """Select the best output from competing agents."""
    if not ranked_outputs:
        return {"strategy": "rank-select", "selected": None, "reason": "No outputs available"}

    best = ranked_outputs[0]
    return {
        "strategy": "rank-select",
        "selected_agent": best["agent_id"],
        "selected_output": best["outputs"],
        "score": best["scores"]["composite"],
        "runner_up": ranked_outputs[1]["agent_id"] if len(ranked_outputs) > 1 else None,
        "runner_up_score": ranked_outputs[1]["scores"]["composite"] if len(ranked_outputs) > 1 else None,
        "total_candidates": len(ranked_outputs),
    }


def merge_chain(session):
    """Use the output of the last agent in the pipeline."""
    agents = session.get("agents", {})
    # Find terminal agent (no dependents)
    all_deps = set()
    for agent_data in agents.values():
        all_deps.update(agent_data.get("dependencies", []))

    terminal_agents = [
        aid for aid in agents
        if aid not in all_deps and agents[aid].get("state") == "COMPLETED"
    ]

    if not terminal_agents:
        return {"strategy": "chain", "error": "No completed terminal agent found"}

    # If multiple terminals, pick the one with highest eval score
    best_terminal = None
    best_score = -1
    for aid in terminal_agents:
        score = agents[aid].get("eval_score", 0) or 0
        if score > best_score:
            best_score = score
            best_terminal = aid

    return {
        "strategy": "chain",
        "terminal_agent": best_terminal,
        "output": agents[best_terminal].get("outputs", {}),
        "eval_score": best_score,
    }


def format_human(result, action):
    """Format result for human output."""
    output = []
    output.append("=" * 60)
    output.append("RESULT RANKER")
    output.append("=" * 60)

    if action == "list":
        output.append(f"\nAgent Outputs ({len(result['outputs'])} completed)")
        output.append("-" * 60)
        for o in result["outputs"]:
            output_keys = list(o.get("outputs", {}).keys())
            output.append(f"  {o['agent_id']}")
            output.append(f"    State: {o['state']} | Duration: {o.get('duration_s', 0)}s")
            output.append(f"    Outputs: {', '.join(output_keys[:5])}")
            if o.get("eval_score") is not None:
                output.append(f"    Eval: {o['eval_score']}")

    elif action == "rank":
        output.append(f"\nRanked Outputs ({len(result['ranked'])} agents)")
        output.append("-" * 60)
        output.append(f"  {'Rank':<6} {'Agent':<20} {'Score':>8} {'Comp':>6} {'Depth':>6} {'Struct':>6}")
        output.append(f"  {'─' * 6} {'─' * 20} {'─' * 8} {'─' * 6} {'─' * 6} {'─' * 6}")
        for i, r in enumerate(result["ranked"], 1):
            s = r["scores"]
            output.append(
                f"  {i:<6} {r['agent_id']:<20} {s['composite']:>8.3f} "
                f"{s['completeness']:>5.2f} {s['depth']:>5.2f} {s['structure']:>5.2f}"
            )

    elif action == "merge":
        merge_result = result["merge_result"]
        strategy = merge_result.get("strategy", "?")
        output.append(f"\nMerge Strategy: {strategy}")
        output.append("-" * 60)

        if strategy == "synthesize":
            output.append(f"  Sections: {merge_result['sections_count']}")
            output.append(f"  Sources:  {merge_result['sources_count']} agents")
            output.append("\n  Attribution:")
            for key, source in merge_result.get("attribution", {}).items():
                output.append(f"    {key} <- {source}")

        elif strategy == "rank-select":
            output.append(f"  Selected: {merge_result.get('selected_agent', '?')} (score: {merge_result.get('score', 0):.3f})")
            if merge_result.get("runner_up"):
                output.append(f"  Runner-up: {merge_result['runner_up']} (score: {merge_result['runner_up_score']:.3f})")

        elif strategy == "chain":
            output.append(f"  Terminal agent: {merge_result.get('terminal_agent', '?')}")

    return "\n".join(output)


def main():
    parser = argparse.ArgumentParser(
        description="Rank and merge outputs from multiple agents.",
        epilog="Example: python result_ranker.py --session session.json --rank",
    )
    parser.add_argument("--session", required=True, help="Path to session JSON file")
    parser.add_argument("--list-outputs", action="store_true", help="List all agent outputs")
    parser.add_argument("--rank", action="store_true", help="Rank outputs by quality")
    parser.add_argument("--merge", choices=["synthesize", "rank-select", "chain"], help="Merge strategy")
    parser.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON")

    args = parser.parse_args()

    session_path = Path(args.session)
    if not session_path.exists():
        print(f"Error: Session file '{args.session}' not found.", file=sys.stderr)
        sys.exit(1)

    session = load_session(session_path)
    outputs = extract_outputs(session)

    if args.list_outputs:
        result = {"outputs": outputs}
        action = "list"
    elif args.rank:
        ranked = rank_outputs(outputs)
        result = {"ranked": ranked}
        action = "rank"
    elif args.merge:
        ranked = rank_outputs(outputs)
        if args.merge == "synthesize":
            merge_result = merge_synthesize(ranked)
        elif args.merge == "rank-select":
            merge_result = merge_rank_select(ranked)
        elif args.merge == "chain":
            merge_result = merge_chain(session)
        result = {"merge_result": merge_result}
        action = "merge"
    else:
        parser.print_help()
        sys.exit(1)

    if args.json_output:
        print(json.dumps(result, indent=2, default=str))
    else:
        print(format_human(result, action))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Manage orchestration sessions for multi-agent workflows.

Creates, updates, and queries session state. A session tracks the execution
lifecycle of a workflow including agent states, outputs, timing, and history.

Usage:
    python session_manager.py create --workflow workflow.json --output session.json
    python session_manager.py status --session session.json
    python session_manager.py update --session session.json --agent researcher --state COMPLETED --output-data '{"result": "..."}'
    python session_manager.py history --session session.json
    python session_manager.py list --dir ./sessions/
"""

import argparse
import json
import os
import sys
import uuid
from datetime import datetime
from pathlib import Path


def load_json(path):
    """Load JSON from file."""
    try:
        with open(path, "r") as f:
            return json.load(f)
    except (json.JSONDecodeError, OSError) as e:
        print(f"Error loading {path}: {e}", file=sys.stderr)
        sys.exit(1)


def save_json(data, path):
    """Save JSON to file."""
    with open(path, "w") as f:
        json.dump(data, f, indent=2, default=str)


def cmd_create(args):
    """Create a new orchestration session from a workflow definition."""
    workflow = load_json(args.workflow)
    agents_def = workflow.get("agents", {})
    config = workflow.get("config", {})

    session_id = str(uuid.uuid4())[:8]
    now = datetime.now().isoformat()

    agents = {}
    for agent_id, agent_def in agents_def.items():
        deps = agent_def.get("dependencies", [])
        agents[agent_id] = {
            "state": "READY" if not deps else "PENDING",
            "task": agent_def.get("task", ""),
            "inputs": agent_def.get("inputs", []),
            "expected_outputs": agent_def.get("outputs", []),
            "dependencies": deps,
            "outputs": None,
            "started_at": None,
            "completed_at": None,
            "duration_s": None,
            "retries": 0,
            "max_retries": agent_def.get("config", {}).get("retries", config.get("retry_on_failure", 1)),
            "eval_score": None,
            "error": None,
        }

    session = {
        "session_id": session_id,
        "workflow_name": workflow.get("name", "unnamed"),
        "workflow_description": workflow.get("description", ""),
        "created_at": now,
        "started_at": now,
        "completed_at": None,
        "state": "RUNNING",
        "config": config,
        "agents": agents,
        "history": [
            {"timestamp": now, "event": "session_created", "detail": f"Workflow: {workflow.get('name', 'unnamed')}"}
        ],
    }

    output_path = args.output or f"session-{session_id}.json"
    save_json(session, output_path)

    return {
        "action": "create",
        "session_id": session_id,
        "workflow_name": workflow.get("name"),
        "agents_count": len(agents),
        "ready_agents": sum(1 for a in agents.values() if a["state"] == "READY"),
        "output_file": output_path,
    }


def cmd_status(args):
    """Show current session status."""
    session = load_json(args.session)
    agents = session.get("agents", {})

    by_state = {}
    for agent_data in agents.values():
        state = agent_data.get("state", "UNKNOWN")
        by_state[state] = by_state.get(state, 0) + 1

    total = len(agents)
    completed = by_state.get("COMPLETED", 0)
    failed = by_state.get("FAILED", 0)
    running = by_state.get("RUNNING", 0)

    # Determine overall health
    if session.get("state") == "COMPLETED":
        health = "COMPLETE"
    elif failed > 0 and running == 0 and by_state.get("PENDING", 0) > 0:
        health = "BLOCKED"
    elif failed > 0:
        health = "AT_RISK"
    elif running > 0:
        health = "RUNNING"
    else:
        health = "IDLE"

    # Elapsed time
    started = session.get("started_at", "")
    elapsed = 0
    if started:
        try:
            elapsed = (datetime.now() - datetime.fromisoformat(started)).total_seconds()
        except ValueError:
            pass

    return {
        "action": "status",
        "session_id": session.get("session_id"),
        "workflow_name": session.get("workflow_name"),
        "state": session.get("state"),
        "health": health,
        "total_agents": total,
        "by_state": by_state,
        "progress_pct": round(completed / total * 100, 1) if total else 0,
        "elapsed_s": round(elapsed),
        "agents": {
            aid: {
                "state": adata.get("state"),
                "duration_s": adata.get("duration_s"),
                "eval_score": adata.get("eval_score"),
            }
            for aid, adata in agents.items()
        },
    }


def cmd_update(args):
    """Update an agent's state in the session."""
    session = load_json(args.session)
    agents = session.get("agents", {})

    if args.agent not in agents:
        return {"action": "update", "error": f"Agent '{args.agent}' not found"}

    agent = agents[args.agent]
    old_state = agent["state"]
    now = datetime.now().isoformat()

    # Update state
    agent["state"] = args.state

    if args.state == "RUNNING" and not agent.get("started_at"):
        agent["started_at"] = now
    elif args.state == "COMPLETED":
        agent["completed_at"] = now
        if agent.get("started_at"):
            try:
                started = datetime.fromisoformat(agent["started_at"])
                agent["duration_s"] = round((datetime.now() - started).total_seconds(), 1)
            except ValueError:
                pass
    elif args.state == "FAILED":
        agent["completed_at"] = now
        agent["error"] = args.error_msg or "Unknown error"

    # Update outputs if provided
    if args.output_data:
        try:
            agent["outputs"] = json.loads(args.output_data)
        except json.JSONDecodeError:
            agent["outputs"] = {"raw": args.output_data}

    # Update eval score if provided
    if args.eval_score is not None:
        agent["eval_score"] = args.eval_score

    # Check if dependents can now be set to READY
    newly_ready = []
    if args.state == "COMPLETED":
        for other_id, other_agent in agents.items():
            if other_agent["state"] == "PENDING":
                deps = other_agent.get("dependencies", [])
                all_met = all(
                    agents.get(d, {}).get("state") == "COMPLETED"
                    for d in deps
                )
                if all_met:
                    other_agent["state"] = "READY"
                    newly_ready.append(other_id)

    # Check if all agents are done
    all_done = all(
        a["state"] in ("COMPLETED", "FAILED", "SKIPPED")
        for a in agents.values()
    )
    if all_done:
        session["state"] = "COMPLETED"
        session["completed_at"] = now

    # Log event
    event = {
        "timestamp": now,
        "event": "agent_state_change",
        "agent": args.agent,
        "old_state": old_state,
        "new_state": args.state,
    }
    if newly_ready:
        event["newly_ready"] = newly_ready
    session.setdefault("history", []).append(event)

    save_json(session, args.session)

    return {
        "action": "update",
        "agent": args.agent,
        "old_state": old_state,
        "new_state": args.state,
        "newly_ready": newly_ready,
        "session_complete": all_done,
    }


def cmd_history(args):
    """Show session event history."""
    session = load_json(args.session)
    history = session.get("history", [])
    return {
        "action": "history",
        "session_id": session.get("session_id"),
        "event_count": len(history),
        "events": history,
    }


def cmd_list(args):
    """List all sessions in a directory."""
    sessions_dir = Path(args.dir)
    if not sessions_dir.is_dir():
        return {"action": "list", "error": f"'{args.dir}' is not a directory"}

    sessions = []
    for jf in sorted(sessions_dir.glob("session-*.json")):
        try:
            data = json.loads(jf.read_text())
            sessions.append({
                "file": str(jf),
                "session_id": data.get("session_id"),
                "workflow_name": data.get("workflow_name"),
                "state": data.get("state"),
                "created_at": data.get("created_at"),
                "agents_count": len(data.get("agents", {})),
            })
        except (json.JSONDecodeError, OSError):
            continue

    return {"action": "list", "count": len(sessions), "sessions": sessions}


def format_human(result):
    """Format result for human output."""
    action = result.get("action", "unknown")
    lines = []

    if "error" in result:
        return f"Error: {result['error']}"

    if action == "create":
        lines.append(f"Session Created: {result['session_id']}")
        lines.append(f"  Workflow: {result['workflow_name']}")
        lines.append(f"  Agents:  {result['agents_count']} ({result['ready_agents']} ready)")
        lines.append(f"  File:    {result['output_file']}")

    elif action == "status":
        lines.append(f"Session: {result['session_id']} ({result['workflow_name']})")
        lines.append(f"  Health:   {result['health']}")
        lines.append(f"  Progress: {result['progress_pct']}% ({result['by_state']})")
        lines.append(f"  Elapsed:  {result['elapsed_s']}s")
        lines.append("")
        for aid, adata in result.get("agents", {}).items():
            dur = f"{adata['duration_s']}s" if adata.get("duration_s") else "---"
            lines.append(f"  {aid:<20} {adata['state']:<12} {dur}")

    elif action == "update":
        lines.append(f"Updated: {result['agent']} ({result['old_state']} -> {result['new_state']})")
        if result.get("newly_ready"):
            lines.append(f"  Newly ready: {', '.join(result['newly_ready'])}")
        if result.get("session_complete"):
            lines.append("  Session is now COMPLETE")

    elif action == "history":
        lines.append(f"Session History ({result['event_count']} events)")
        lines.append("-" * 50)
        for evt in result.get("events", []):
            ts = evt.get("timestamp", "")[:19]
            event_type = evt.get("event", "")
            detail = evt.get("detail", "")
            agent = evt.get("agent", "")
            if agent:
                lines.append(f"  {ts}  {event_type}: {agent} ({evt.get('old_state', '')} -> {evt.get('new_state', '')})")
            else:
                lines.append(f"  {ts}  {event_type}: {detail}")

    elif action == "list":
        lines.append(f"Sessions ({result['count']} found)")
        lines.append("-" * 60)
        for s in result.get("sessions", []):
            lines.append(f"  {s['session_id']}  {s['workflow_name']:<20}  {s['state']:<12}  {s['agents_count']} agents")

    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Manage orchestration sessions for multi-agent workflows.",
    )
    parser.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON")

    sub = parser.add_subparsers(dest="command")

    p_create = sub.add_parser("create", help="Create a new session")
    p_create.add_argument("--workflow", required=True, help="Path to workflow JSON")
    p_create.add_argument("--output", help="Output session file path")

    p_status = sub.add_parser("status", help="Show session status")
    p_status.add_argument("--session", required=True, help="Path to session JSON")

    p_update = sub.add_parser("update", help="Update agent state")
    p_update.add_argument("--session", required=True, help="Path to session JSON")
    p_update.add_argument("--agent", required=True, help="Agent ID to update")
    p_update.add_argument("--state", required=True,
                          choices=["PENDING", "READY", "RUNNING", "COMPLETED", "FAILED", "SKIPPED", "EVALUATING"])
    p_update.add_argument("--output-data", help="Agent output data (JSON string)")
    p_update.add_argument("--eval-score", type=float, help="Evaluation score (0-1)")
    p_update.add_argument("--error-msg", help="Error message (for FAILED state)")

    p_history = sub.add_parser("history", help="Show session event history")
    p_history.add_argument("--session", required=True, help="Path to session JSON")

    p_list = sub.add_parser("list", help="List sessions in a directory")
    p_list.add_argument("--dir", required=True, help="Directory containing session files")

    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        sys.exit(1)

    commands = {
        "create": cmd_create,
        "status": cmd_status,
        "update": cmd_update,
        "history": cmd_history,
        "list": cmd_list,
    }

    result = commands[args.command](args)

    if args.json_output:
        print(json.dumps(result, indent=2, default=str))
    else:
        print(format_human(result))


if __name__ == "__main__":
    main()