Spaces:
Running
Running
File size: 4,633 Bytes
d771e01 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | 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"] == []
|