diff --git a/packages/agent-mesh/src/agentmesh/cli/trust_cli.py b/packages/agent-mesh/src/agentmesh/cli/trust_cli.py index 3662aad6..f0af7657 100644 --- a/packages/agent-mesh/src/agentmesh/cli/trust_cli.py +++ b/packages/agent-mesh/src/agentmesh/cli/trust_cli.py @@ -136,6 +136,11 @@ def _get_demo_history(agent_id: str) -> list[dict]: }) return history +def _categorize_tasks(history): + successful = [h['event'] for h in history if h['delta'] > 0] + failed = [h['event'] for h in history if h['delta'] < 0] + return successful, failed + def _output_json(data: object) -> None: """Print data as JSON to stdout.""" @@ -569,3 +574,75 @@ def attest(agent_id: str, note: str, score_boost: int, fmt: str, json_flag: bool console.print(f" New Level: [{style}]{level}[/{style}]") console.print(f" Attested At: {attestation_info['attested_at']}") console.print() + + +@trust.command("report") +@click.option( + "--format", "fmt", + type=click.Choice(["table", "json"]), + default="table", + help="Output format (table or json).", +) +@click.option("--json", "json_flag", is_flag=True, help="Output as JSON (shorthand for --format json).") +def report_agents(fmt: str, json_flag: bool): + """Report trust score summary for registered agents""" + if json_flag: + fmt = "json" + + peers = _get_demo_peers() + filtered: list[PeerInfo] = list(peers.values()) + + if fmt == "json": + data = [] + for p in filtered: + history = _get_demo_history(p.peer_did) + successful_tasks, failure_tasks = _categorize_tasks(history); + for h in history: + if h["delta"] >= 0: + successful_tasks.append(h["event"]) + elif h["delta"] < 0: + failure_tasks.append(h["event"]) + + data.append({ + "agent_id": p.peer_did, + "trust_score": p.trust_score, + "trust_level": _trust_level_label(p.trust_score), + "successful_tasks": successful_tasks, + "failure_tasks": failure_tasks, + "last_activity": history[-1]["timestamp"] if history else "N/A" + }) + _output_json(data) + return + + # Table output + console.print("\n[bold blue]🛡️ Trust Network — Agent Report[/bold blue]\n") + table = Table(box=box.ROUNDED) + table.add_column("Agent ID", style="cyan", no_wrap=True) + table.add_column("Score", justify="right") + table.add_column("Level") + table.add_column("Successful Tasks") + table.add_column("Failure Tasks") + table.add_column("Last Activity", style="dim") + + for p in filtered: + history = _get_demo_history(p.peer_did) + successful_tasks, failure_tasks = _categorize_tasks(history); + for h in history: + if h["delta"] >= 0: + successful_tasks.append(h["event"]) + elif h["delta"] < 0: + failure_tasks.append(h["event"]) + + level = _trust_level_label(p.trust_score) + style = _trust_level_style(level) + table.add_row( + p.peer_did, + str(p.trust_score), + f"[{style}]{level}[/{style}]", + ", ".join(successful_tasks) if successful_tasks else "N/A", + ", ".join(failure_tasks) if failure_tasks else "N/A", + history[-1]["timestamp"] if history else "N/A", + ) + + console.print(table) + console.print(f"\n Total agents: {len(peers)}\n") diff --git a/packages/agent-mesh/tests/test_trust_cli.py b/packages/agent-mesh/tests/test_trust_cli.py index 368fb75f..764cf24e 100644 --- a/packages/agent-mesh/tests/test_trust_cli.py +++ b/packages/agent-mesh/tests/test_trust_cli.py @@ -329,6 +329,35 @@ def test_attest_custom_note(self, runner): data = json.loads(result.output) assert data["note"] == "Passed security audit" +# --------------------------------------------------------------------------- +# trust report +# --------------------------------------------------------------------------- + +class TestTrustReport: + def test_report_table(self, runner): + result = runner.invoke(app, ["trust", "report"]) + assert result.exit_code == 0 + assert "alpha" in result.output or "Agent" in result.output + + def test_report_json(self, runner): + result = runner.invoke(app, ["trust", "report", "--format", "json"]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert isinstance(data, list) + assert len(data) >= 3 + assert "agent_id" in data[0] + assert "trust_score" in data[0] + assert "trust_level" in data[0] + assert "successful_tasks" in data[0] + assert "failure_tasks" in data[0] + assert "last_activity" in data[0] + + def test_report_json_flag(self, runner): + result = runner.invoke(app, ["trust", "report", "--json"]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert isinstance(data, list) + # --------------------------------------------------------------------------- # trust help @@ -341,6 +370,6 @@ def test_trust_help(self, runner): assert "trust" in result.output.lower() def test_subcommand_help(self, runner): - for cmd in ("list", "inspect", "history", "graph", "revoke", "attest"): + for cmd in ("list", "inspect", "history", "graph", "revoke", "attest", "report"): result = runner.invoke(app, ["trust", cmd, "--help"]) assert result.exit_code == 0, f"Help for 'trust {cmd}' failed"