1
0
Fork 0
Code Issues Pull requests Projects Releases 2 Packages Wiki Activity Actions Pages

Trace complete directed graph lineage in Flow view

This commit is contained in:
Andraxion 2026-07-25 01:05:45 -04:00
parent 984a4fd993
commit e405326334
3 changed files with 139 additions and 61 deletions

View file

@ -33,9 +33,10 @@ from .errors import DocForgeError
from .index import APPLICATION_ID, INDEX_SCHEMA_VERSION, ProjectIndex, re_tokenize from .index import APPLICATION_ID, INDEX_SCHEMA_VERSION, ProjectIndex, re_tokenize
from .project import project_root_fingerprint from .project import project_root_fingerprint
VISUALIZATION_TEMPLATE = "graph-browser@10" VISUALIZATION_TEMPLATE = "graph-browser@11"
DEFAULT_EDGE_LIMIT = 100 DEFAULT_EDGE_LIMIT = 100
MAX_EDGE_LIMIT = 400 MAX_EDGE_LIMIT = 400
MAX_LINEAGE_EDGE_LIMIT = 1_000
DEFAULT_INITIAL_GRACE_SECONDS = 120.0 DEFAULT_INITIAL_GRACE_SECONDS = 120.0
DEFAULT_LEASE_SECONDS = 180.0 DEFAULT_LEASE_SECONDS = 180.0
LEASE_MONITOR_INTERVAL_SECONDS = 1.0 LEASE_MONITOR_INTERVAL_SECONDS = 1.0
@ -287,6 +288,76 @@ class VisualizationIndexSnapshot:
snapshot=True, snapshot=True,
) )
def lineage(self, node_id: str, *, limit: int) -> dict[str, object]:
"""Return every bounded, directed ancestry path terminating at ``node_id``.
A lineage follows stored edge direction only: ``source -> target``. This keeps
Flow literal and auditable. It does not reinterpret relationship meanings or
reverse dependency/data edges as the old client-side Flow view did.
"""
if type(limit) is not int or limit < 1 or limit > MAX_LINEAGE_EDGE_LIMIT:
raise DocForgeError(
"invalid_limit",
"Visualization lineage limit exceeds the fixed safety boundary",
maximum=MAX_LINEAGE_EDGE_LIMIT,
)
with self._connection() as connection:
root_row = connection.execute(
"SELECT * FROM nodes WHERE node_id = ?", (node_id,)
).fetchone()
if root_row is None:
raise DocForgeError(
"missing_node",
"No node has the requested stable ID",
node_id=node_id,
)
visited = {node_id}
frontier = {node_id}
selected: list[dict[str, str]] = []
truncated = False
while frontier and len(selected) < limit:
placeholders = ",".join("?" for _ in frontier)
remaining = limit - len(selected)
rows = connection.execute(
"SELECT source_id, relation, target_id FROM edges "
f"WHERE target_id IN ({placeholders}) "
"ORDER BY source_id, relation, target_id LIMIT ?",
(*sorted(frontier), remaining + 1),
).fetchall()
if len(rows) > remaining:
rows = rows[:remaining]
truncated = True
next_frontier: set[str] = set()
for row in rows:
edge = {
"source_id": row["source_id"],
"relation": row["relation"],
"target_id": row["target_id"],
}
selected.append(edge)
source_id = edge["source_id"]
if source_id not in visited:
visited.add(source_id)
next_frontier.add(source_id)
frontier = next_frontier
if frontier and len(selected) >= limit:
truncated = True
placeholders = ",".join("?" for _ in visited)
node_rows = connection.execute(
f"SELECT * FROM nodes WHERE node_id IN ({placeholders}) ORDER BY node_id",
tuple(sorted(visited)),
).fetchall()
return self._result(
root=node_id,
lineage=True,
edge_limit=limit,
truncated=truncated,
node=_node_dict(root_row),
nodes=[_node_dict(row, include_content=False) for row in node_rows],
edges=selected,
snapshot=True,
)
def require_node(self, node_id: str) -> None: def require_node(self, node_id: str) -> None:
with self._connection() as connection: with self._connection() as connection:
row = connection.execute("SELECT 1 FROM nodes WHERE node_id = ?", (node_id,)).fetchone() row = connection.execute("SELECT 1 FROM nodes WHERE node_id = ?", (node_id,)).fetchone()
@ -615,6 +686,9 @@ class VisualizationRunner:
elif parsed.path == f"{prefix}/api/node": elif parsed.path == f"{prefix}/api/node":
self._touch_lease() self._touch_lease()
payload = self._node(reader, params) payload = self._node(reader, params)
elif parsed.path == f"{prefix}/api/lineage":
self._touch_lease()
payload = self._lineage(reader, params)
else: else:
self._respond_error( self._respond_error(
handler, handler,
@ -682,6 +756,23 @@ class VisualizationRunner:
) )
return reader.node(node_id, depth=depth, limit=limit) return reader.node(node_id, depth=depth, limit=limit)
def _lineage(
self,
reader: VisualizationIndexSnapshot,
params: dict[str, list[str]],
) -> dict[str, object]:
node_id = _one(params, "id").strip()
if not node_id:
raise DocForgeError("missing_node", "One exact node ID is required")
limit = _integer(_one(params, "limit") or str(MAX_LINEAGE_EDGE_LIMIT))
if limit > MAX_LINEAGE_EDGE_LIMIT:
raise DocForgeError(
"invalid_limit",
"Visualization lineage limit exceeds the fixed safety boundary",
maximum=MAX_LINEAGE_EDGE_LIMIT,
)
return reader.lineage(node_id, limit=limit)
def _filter( def _filter(
self, self,
reader: VisualizationIndexSnapshot, reader: VisualizationIndexSnapshot,
@ -2046,32 +2137,18 @@ _GRAPH_BROWSER_HTML = r"""<!DOCTYPE html>
}, },
])); ]));
} }
function semanticFlowEdge(edge) {
const direction = relationStyle(edge.relation).flow;
if (direction === null) return null;
if (direction === "reverse") {
return {
...edge,
source_id: edge.target_id,
target_id: edge.source_id,
stored_source_id: edge.source_id,
stored_target_id: edge.target_id,
};
}
return {
...edge,
stored_source_id: edge.source_id,
stored_target_id: edge.target_id,
};
}
function buildFlowGraph(data) { function buildFlowGraph(data) {
const semanticEdges = data.edges.map(semanticFlowEdge).filter(Boolean); // The server has already supplied the complete directed ancestry for this
// focus. Keep every stored edge as-is: source -> target. Flow must show
// what literally leads to the selected terminal, not infer an alternate
// direction from a relationship label.
const lineageEdges = data.edges;
const upstreamHops = new Map([[data.root, 0]]); const upstreamHops = new Map([[data.root, 0]]);
let frontier = [data.root]; let frontier = [data.root];
while (frontier.length) { while (frontier.length) {
const next = []; const next = [];
for (const targetId of frontier) { for (const targetId of frontier) {
for (const edge of semanticEdges) { for (const edge of lineageEdges) {
if (edge.target_id !== targetId || upstreamHops.has(edge.source_id)) continue; if (edge.target_id !== targetId || upstreamHops.has(edge.source_id)) continue;
upstreamHops.set(edge.source_id, upstreamHops.get(targetId) + 1); upstreamHops.set(edge.source_id, upstreamHops.get(targetId) + 1);
next.push(edge.source_id); next.push(edge.source_id);
@ -2081,7 +2158,7 @@ _GRAPH_BROWSER_HTML = r"""<!DOCTYPE html>
} }
const nodes = data.nodes.filter((node) => upstreamHops.has(node.node_id)); const nodes = data.nodes.filter((node) => upstreamHops.has(node.node_id));
const nodeIds = new Set(nodes.map((node) => node.node_id)); const nodeIds = new Set(nodes.map((node) => node.node_id));
const edges = semanticEdges.filter( const edges = lineageEdges.filter(
(edge) => nodeIds.has(edge.source_id) && nodeIds.has(edge.target_id), (edge) => nodeIds.has(edge.source_id) && nodeIds.has(edge.target_id),
); );
const topology = new Map(nodes.map((node) => [ const topology = new Map(nodes.map((node) => [
@ -2532,11 +2609,18 @@ _GRAPH_BROWSER_HTML = r"""<!DOCTYPE html>
} }
async function loadNode(nodeId) { async function loadNode(nodeId) {
try { try {
setStatus(`Loading ${nodeId}`); const showingFlow = state.mode === "flow";
const params = new URLSearchParams({id: nodeId, depth: String(state.depth), limit: "100"}); setStatus(`${showingFlow ? "Tracing lineage for" : "Loading"} ${nodeId}`);
const data = await api(`node?${params}`); const params = new URLSearchParams(
showingFlow
? {id: nodeId, limit: "1000"}
: {id: nodeId, depth: String(state.depth), limit: "100"},
);
const data = await api(`${showingFlow ? "lineage" : "node"}?${params}`);
renderGraph(data); renderGraph(data);
setStatus(`${data.nodes.length} nodes · ${data.edges.length} edges in neighborhood`); const scope = showingFlow ? "directed ancestry" : "neighborhood";
const suffix = data.truncated ? " · truncated at the safety limit" : "";
setStatus(`${data.nodes.length} nodes · ${data.edges.length} edges in ${scope}${suffix}`);
history.replaceState( history.replaceState(
null, null,
"", "",
@ -2546,22 +2630,13 @@ _GRAPH_BROWSER_HTML = r"""<!DOCTYPE html>
setStatus(error.message, true); setStatus(error.message, true);
} }
} }
function setViewMode(mode) { async function setViewMode(mode) {
if (mode !== "nodes" && mode !== "flow") return; if (mode !== "nodes" && mode !== "flow") return;
state.mode = mode; state.mode = mode;
$("view-switch").dataset.mode = mode; $("view-switch").dataset.mode = mode;
$("view-nodes").setAttribute("aria-pressed", String(mode === "nodes")); $("view-nodes").setAttribute("aria-pressed", String(mode === "nodes"));
$("view-flow").setAttribute("aria-pressed", String(mode === "flow")); $("view-flow").setAttribute("aria-pressed", String(mode === "flow"));
if (state.graph) { if (state.root) await loadNode(state.root);
renderGraph(state.graph, true);
const suffix = mode === "flow" ? "upstream flow" : "node neighborhood";
setStatus(`Showing ${suffix} for ${state.root}`);
history.replaceState(
null,
"",
`?node=${encodeURIComponent(state.root)}&depth=${state.depth}&view=${state.mode}`,
);
}
} }
function clamp(value, minimum, maximum) { function clamp(value, minimum, maximum) {
return Math.min(maximum, Math.max(minimum, value)); return Math.min(maximum, Math.max(minimum, value));

View file

@ -130,7 +130,7 @@ class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase):
visualization = results[11].structuredContent["visualization"] visualization = results[11].structuredContent["visualization"]
self.assertTrue(visualization["read_only"]) self.assertTrue(visualization["read_only"])
self.assertTrue(visualization["project_bound"]) self.assertTrue(visualization["project_bound"])
self.assertEqual("graph-browser@10", visualization["template"]) self.assertEqual("graph-browser@11", visualization["template"])
self.assertEqual("managed_idle", visualization["lifetime"]["policy"]) self.assertEqual("managed_idle", visualization["lifetime"]["policy"])
self.assertEqual("docforge_stop_visualization", visualization["lifetime"]["stop_tool"]) self.assertEqual("docforge_stop_visualization", visualization["lifetime"]["stop_tool"])
self.assertTrue(visualization["url"].startswith("http://127.0.0.1:")) self.assertTrue(visualization["url"].startswith("http://127.0.0.1:"))

View file

@ -177,47 +177,34 @@ if (centered.width !== 500 || centered.height !== 300) fail("center preserves zo
if (centered.x !== positions.get("child-two").x - 250) fail("center x"); if (centered.x !== positions.get("child-two").x - 250) fail("center x");
if (centered.y !== positions.get("child-two").y - 150) fail("center y"); if (centered.y !== positions.get("child-two").y - 150) fail("center y");
if (relationStyle("calls").family !== "Execution") fail("calls family"); if (relationStyle("calls").family !== "Execution") fail("calls family");
if (relationStyle("contains").flow !== "forward") fail("contains flow direction");
if (relationStyle("reads").flow !== "reverse") fail("reads flow direction");
if (relationStyle("documents").flow !== null) fail("documents excluded from flow");
if (relationStyle("unknown_relation").family !== "Other") fail("fallback relation family"); if (relationStyle("unknown_relation").family !== "Other") fail("fallback relation family");
const flowData = { const flowData = {
root: "primary", root: "primary",
depth: 2,
nodes: [ nodes: [
{node_id: "primary"}, {node_id: "primary"},
{node_id: "caller"}, {node_id: "package"},
{node_id: "dependency"}, {node_id: "module"},
{node_id: "reader"},
{node_id: "test-class"}, {node_id: "test-class"},
{node_id: "test-file"}, {node_id: "test-file"},
{node_id: "document"}, {node_id: "evidence"},
], ],
edges: [ edges: [
{source_id: "caller", relation: "calls", target_id: "primary"},
{source_id: "primary", relation: "depends_on", target_id: "dependency"},
{source_id: "primary", relation: "reads", target_id: "reader"},
{source_id: "test-class", relation: "contains", target_id: "primary"}, {source_id: "test-class", relation: "contains", target_id: "primary"},
{source_id: "test-file", relation: "contains", target_id: "test-class"}, {source_id: "test-file", relation: "contains", target_id: "test-class"},
{source_id: "document", relation: "documents", target_id: "primary"}, {source_id: "module", relation: "contains", target_id: "test-file"},
{source_id: "package", relation: "contains", target_id: "module"},
{source_id: "evidence", relation: "verifies", target_id: "primary"},
], ],
}; };
const flow = buildFlowGraph(flowData); const flow = buildFlowGraph(flowData);
const flowIds = new Set(flow.nodes.map((node) => node.node_id)); const flowIds = new Set(flow.nodes.map((node) => node.node_id));
if (!flowIds.has("caller") || !flowIds.has("dependency") || !flowIds.has("reader") if (!flowIds.has("package") || !flowIds.has("module") || !flowIds.has("test-class")
|| !flowIds.has("test-class") || !flowIds.has("test-file")) { || !flowIds.has("test-file") || !flowIds.has("evidence")) fail("full lineage membership");
fail("upstream flow membership");
}
if (flowIds.has("document")) fail("evidence leaked into flow");
const flowPositions = layoutFlow(flow.nodes, flow.root, flow.topology); const flowPositions = layoutFlow(flow.nodes, flow.root, flow.topology);
if (flowPositions.get("primary").x !== 0) fail("flow destination position"); if (flowPositions.get("primary").x !== 0) fail("flow destination position");
if (flowPositions.get("caller").x >= flowPositions.get("primary").x) { if (flowPositions.get("package").x >= flowPositions.get("primary").x) {
fail("flow upstream direction"); fail("flow upstream direction");
} }
const dependencyEdge = flow.edges.find((edge) => edge.relation === "depends_on");
if (dependencyEdge.source_id !== "dependency" || dependencyEdge.target_id !== "primary") {
fail("dependency semantic reversal");
}
const containmentEdge = flow.edges.find((edge) => edge.relation === "contains" const containmentEdge = flow.edges.find((edge) => edge.relation === "contains"
&& edge.source_id === "test-class"); && edge.source_id === "test-class");
if (!containmentEdge || containmentEdge.target_id !== "primary") { if (!containmentEdge || containmentEdge.target_id !== "primary") {
@ -307,9 +294,9 @@ if (!containmentEdge || containmentEdge.target_id !== "primary") {
self.assertIn("relationStyles", html) self.assertIn("relationStyles", html)
self.assertIn("appendRelationMarker", html) self.assertIn("appendRelationMarker", html)
self.assertIn("renderRelationshipKey", html) self.assertIn("renderRelationshipKey", html)
self.assertIn("semanticFlowEdge", html)
self.assertIn("buildFlowGraph", html) self.assertIn("buildFlowGraph", html)
self.assertIn("layoutFlow", html) self.assertIn("layoutFlow", html)
self.assertIn('api(`${showingFlow ? "lineage" : "node"}?${params}`)', html)
self.assertIn("filterByDescriptor", html) self.assertIn("filterByDescriptor", html)
self.assertIn("api(`filter?${params}`)", html) self.assertIn("api(`filter?${params}`)", html)
self.assertIn('setViewMode("flow")', html) self.assertIn('setViewMode("flow")', html)
@ -371,6 +358,22 @@ if (!containmentEdge || containmentEdge.target_id !== "primary") {
self.assertEqual("guide.workflow", node["node"]["node_id"]) self.assertEqual("guide.workflow", node["node"]["node_id"])
self.assertEqual("guide.workflow", node["root"]) self.assertEqual("guide.workflow", node["root"])
lineage_query = urllib.parse.urlencode({"id": "guide.workflow", "limit": 1000})
with urllib.request.urlopen(
f"{base}api/lineage?{lineage_query}", timeout=2
) as response:
lineage = json.load(response)
self.assertTrue(lineage["lineage"])
self.assertEqual("guide.workflow", lineage["root"])
self.assertIn(
{
"source_id": "proof.validation",
"relation": "proves",
"target_id": "guide.workflow",
},
lineage["edges"],
)
wrong_token = f"{first_url.scheme}://{first_url.netloc}/wrong-token/api/overview" wrong_token = f"{first_url.scheme}://{first_url.netloc}/wrong-token/api/overview"
with self.assertRaises(urllib.error.HTTPError) as missing: with self.assertRaises(urllib.error.HTTPError) as missing:
urllib.request.urlopen(wrong_token, timeout=2) urllib.request.urlopen(wrong_token, timeout=2)