Spaces:
Running
Running
| from __future__ import annotations | |
| import json | |
| from trace2mcp.analysis import diff_traces, explain_workflow, learn_traces | |
| from trace2mcp.compiler import compile_trace | |
| from trace2mcp.models import Trace, TraceEvent | |
| def event(event_id, tool, arguments, result, *, side_effect="read", status="success"): | |
| return TraceEvent( | |
| event_id, | |
| tool, | |
| arguments, | |
| result, | |
| side_effect=side_effect, | |
| status=status, | |
| ) | |
| def test_explain_workflow_reports_evidenced_dependencies_and_parallelism(): | |
| workflow = compile_trace( | |
| Trace( | |
| inputs={"company": "Acme"}, | |
| events=[ | |
| event("resolve", "resolve", {"company": "Acme"}, {"id": 7}), | |
| event("news", "news", {"id": 7}, ["launch"]), | |
| event("profile", "profile", {"id": 7}, {"sector": "tools"}), | |
| ], | |
| ), | |
| name="company_report", | |
| ) | |
| report = explain_workflow(workflow) | |
| assert report["layers"] == [["resolve"], ["news", "profile"]] | |
| assert report["max_parallelism"] == 2 | |
| news = next(node for node in report["nodes"] if node["id"] == "news") | |
| assert news["parallel_with"] == ["profile"] | |
| assert news["dependencies"] == [ | |
| { | |
| "node_id": "resolve", | |
| "reason": "data_reference", | |
| "matches": [ | |
| { | |
| "argument_path": "arguments.id", | |
| "reference": "nodes.resolve.result.id", | |
| } | |
| ], | |
| } | |
| ] | |
| json.dumps(report) | |
| def test_diff_traces_is_stable_and_distinguishes_missing_from_null(): | |
| before = Trace( | |
| inputs={"query": "one", "optional": None}, | |
| events=[ | |
| event("a", "search", {"query": "one"}, {"id": 1}), | |
| event("b", "format", {"id": 1}, "done"), | |
| ], | |
| ) | |
| after = Trace( | |
| inputs={"query": "two", "added": None}, | |
| events=[ | |
| event("b", "format", {"id": 2}, "done"), | |
| event("a", "search_v2", {"query": "two"}, {"id": 2}), | |
| event("c", "audit", {}, {"ok": True}), | |
| ], | |
| ) | |
| report = diff_traces(before, after) | |
| assert report["summary"] == { | |
| "added": 1, | |
| "removed": 0, | |
| "changed": 2, | |
| "input_changes": 3, | |
| "order_changed": True, | |
| } | |
| assert [step["id"] for step in report["added_steps"]] == ["c"] | |
| assert report["input_changes"]["optional"]["before_present"] is True | |
| assert report["input_changes"]["optional"]["after_present"] is False | |
| assert report == diff_traces(before, after) | |
| json.dumps(report) | |
| def test_learn_traces_clusters_shapes_and_only_infers_evidenced_values(): | |
| traces = [ | |
| Trace( | |
| inputs={"customer_id": 10}, | |
| events=[ | |
| event("lookup-1", "lookup", {"id": 10, "country": "BR"}, {"name": "A"}), | |
| event("render-1", "render", {"name": "A", "format": "md"}, "A"), | |
| ], | |
| ), | |
| Trace( | |
| inputs={"customer_id": 20}, | |
| events=[ | |
| event("lookup-2", "lookup", {"id": 20, "country": "BR"}, {"name": "B"}), | |
| event("render-2", "render", {"name": "B", "format": "md"}, "B"), | |
| ], | |
| ), | |
| Trace( | |
| inputs={"customer_id": 30}, | |
| events=[event("write", "store", {"id": 30}, {"ok": True}, side_effect="write")], | |
| ), | |
| ] | |
| report = learn_traces(traces) | |
| assert report["cluster_count"] == 2 | |
| cluster = report["clusters"][0] | |
| assert cluster["trace_indexes"] == [0, 1] | |
| assert cluster["signature"] == [ | |
| {"position": 0, "tool": "lookup", "side_effect": "read", "status": "success"}, | |
| {"position": 1, "tool": "render", "side_effect": "read", "status": "success"}, | |
| ] | |
| assert any(item["argument_path"] == "country" for item in cluster["constants"]) | |
| inferred = next(item for item in cluster["inferred_inputs"] if item["argument_path"] == "id") | |
| assert inferred["input_path"] == "customer_id" | |
| assert inferred["observed_distinct_values"] == 2 | |
| assert inferred["evidence"] == {"matches": 2, "total": 2, "confidence": 1.0} | |
| assert not any(item["argument_path"] == "name" for item in cluster["inferred_inputs"]) | |
| assert report["branch_inference"]["performed"] is False | |
| json.dumps(report) | |
| def test_learn_traces_excludes_runs_without_successful_steps(): | |
| report = learn_traces( | |
| [Trace(inputs={}, events=[event("failed", "search", {}, None, status="error")])] | |
| ) | |
| assert report["analyzed_trace_count"] == 0 | |
| assert report["excluded_trace_indexes"] == [0] | |
| assert report["clusters"] == [] | |