Graph Updates

akms.graph.update_graph

update_graph.py — PCD/AgentMemory → Local State Mutations (§2.7 of system design).

Applies the persistent zone of Phase Completion Documents (or individual AgentMemories) to the graph. Pure algorithmic mutation with deterministic threshold-based dedup (token Jaccard + exact-id signal), no LLM calls.

Writes exclusively to local_state.yaml and local-nodes/ — never touches global node files.

Mutation pipeline

1. Process nodes_used → confidence boost/decay + activations
2. Propagate confidence hits to neighbors via edge weights
3. Process pitfalls_discovered → local_edges
4. Process new_knowledge → dedup check → local-nodes/
5. Create session node entries
6. Write local_state.yaml
7. Call build_graph() to recompile graph.json

update_graph

update_graph(
    source: AgentMemory | PCD | dict,
    repo_root: str | Path,
    config: PropagationConfig | None = None,
    global_vault: str | Path | None = None,
    recompile: bool = True,
) -> dict

Apply persistent zone mutations from a PCD or AgentMemory to the graph.

This is the main entry point for Phase 4. It: 1. Loads the current compiled graph 2. Processes nodes_used → confidence mutations 3. Propagates decay to neighbors 4. Processes pitfalls → local_edges 5. Creates session node entry 6. Processes new_knowledge → local-nodes/ 7. Writes local_state.yaml 8. Recompiles graph.json via build_graph()

Parameters:

Name	Type	Description	Default
source	AgentMemory | PCD | dict	AgentMemory, PCD, or persistent zone dict.	required
repo_root	str | Path	Path to the repository root.	required
config	PropagationConfig | None	PropagationConfig (loads from file or uses defaults if None).	None
global_vault	str | Path | None	Override path to global vault.	None
recompile	bool	Whether to call build_graph() after mutations. Default True.	True

Returns:

Type	Description
dict	Dict with mutation summary:
dict	{ "confidence_events": [...], "propagation_events": [...], "pitfall_events": [...], "knowledge_events": [...], "session_node_id": str,
dict	}

Source code in packages/akms/src/akms/graph/update_graph.py

@traced("akms.update_graph")
def update_graph(
    source: AgentMemory | PCD | dict,
    repo_root: str | Path,
    config: PropagationConfig | None = None,
    global_vault: str | Path | None = None,
    recompile: bool = True,
) -> dict:
    """Apply persistent zone mutations from a PCD or AgentMemory to the graph.

    This is the main entry point for Phase 4. It:
    1. Loads the current compiled graph
    2. Processes nodes_used → confidence mutations
    3. Propagates decay to neighbors
    4. Processes pitfalls → local_edges
    5. Creates session node entry
    6. Processes new_knowledge → local-nodes/
    7. Writes local_state.yaml
    8. Recompiles graph.json via build_graph()

    Args:
        source: AgentMemory, PCD, or persistent zone dict.
        repo_root: Path to the repository root.
        config: PropagationConfig (loads from file or uses defaults if None).
        global_vault: Override path to global vault.
        recompile: Whether to call build_graph() after mutations. Default True.

    Returns:
        Dict with mutation summary:
        {
            "confidence_events": [...],
            "propagation_events": [...],
            "pitfall_events": [...],
            "knowledge_events": [...],
            "session_node_id": str,
        }
    """
    repo_root = Path(repo_root)
    knowledge_dir = repo_root / "knowledge"
    graph_dir = knowledge_dir / "graph"
    overlay_path = graph_dir / "local_state.yaml"
    graph_json = graph_dir / "graph.json"

    # Load config
    if config is None:
        config_path = graph_dir / "propagation_config.yaml"
        if config_path.exists():
            config = parse_propagation_config(config_path)
        else:
            config = PropagationConfig()

    # Load compiled graph
    if graph_json.exists():
        G = load_graph(graph_json)
    else:
        # Build first if no graph exists
        G = build_graph(repo_root, global_vault=global_vault)

    # Load overlay
    overlay = _load_overlay(overlay_path)

    # Ensure required sections exist
    overlay.setdefault("akms_schema", AKMS_SCHEMA_VERSION)
    overlay.setdefault("nodes", {})
    overlay.setdefault("local_edges", [])
    overlay.setdefault("session_nodes", {})
    overlay.setdefault("suppressed_edges", [])

    # Extract persistent zone
    persistent = _extract_persistent_zone(source)
    source_id = _get_source_id(source)
    phase = _get_source_phase(source)
    today = date.today()

    # Replay ledger — same source_id applied twice is a no-op
    # (NFR-D03). Check before mutating anything; append post-commit. If the
    # ledger is missing (legacy overlay files), treat as empty and continue.
    processed_sources: list[str] = list(overlay.get("processed_sources") or [])
    if source_id and source_id in processed_sources:
        logger.info(
            "update_graph: source_id=%r already processed — no-op (replay ledger)",
            source_id,
        )
        return {
            "confidence_events": [],
            "propagation_events": [],
            "pitfall_events": [],
            "knowledge_events": [],
            "session_node_id": f"session-{source_id}",
            "replayed": True,
        }

    logger.info("update_graph: processing %s (phase %d)", source_id, phase)

    confidence_events = _process_nodes_used(
        G,
        overlay,
        persistent.get("nodes_used", []),
        config,
        source_id,
        today,
    )

    propagation_events = _propagate_to_neighbors(
        G,
        overlay,
        confidence_events,
        config,
    )

    session_node_id = _create_session_node(overlay, source_id, source, phase)

    pitfall_events = _process_pitfalls(
        overlay,
        persistent.get("pitfalls_discovered", []),
        session_node_id,
        source_id=source_id,
    )

    knowledge_events = _process_new_knowledge(
        G,
        repo_root,
        persistent.get("new_knowledge", []),
        config,
    )

    # Persist review/report categories consumed by graph_status().
    coverage_flags = overlay.setdefault("coverage_flags", [])
    for feedback in persistent.get("nodes_used", []):
        coverage_val = feedback.get("coverage", "")
        if isinstance(coverage_val, Coverage):
            coverage = coverage_val.value
        else:
            coverage = str(coverage_val)
        if coverage not in (Coverage.MISSING_DETAIL.value, Coverage.OUTDATED.value):
            continue
        coverage_flags.append(
            {
                "node_id": str(feedback.get("id", "")),
                "coverage": coverage,
                "source_id": source_id,
                "phase": phase,
                "date": str(today),
            }
        )

    dedup_events = overlay.setdefault("dedup_events", [])
    for event in knowledge_events:
        action = str(event.get("action", ""))
        if action not in ("dedup_append", "dedup_global_skip"):
            continue
        dedup_events.append(
            {
                "action": action,
                "merged_into": str(event.get("merged_into", event.get("node_id", ""))),
                "node_id": str(event.get("node_id", "")),
                "score": event.get("score"),
                "threshold": event.get("threshold"),
                "source_id": source_id,
                "phase": phase,
                "date": str(today),
            }
        )

    blocked_tasks = overlay.setdefault("blocked_tasks", [])
    if isinstance(source, dict):
        for item in source.get("blocked_tasks", []):
            if isinstance(item, dict):
                blocked_tasks.append(dict(item))
            elif isinstance(item, str):
                blocked_tasks.append({"task": item})
    elif isinstance(source, PCD):
        for issue in source.known_issues.failing_tests:
            if issue.impact_on_next_phase != ImpactOnNextPhase.BLOCKING:
                continue
            blocked_tasks.append(
                {
                    "task": issue.tests,
                    "reason": issue.reason,
                    "source_id": source_id,
                    "phase": phase,
                    "date": str(today),
                }
            )

    _prune_session_refs(overlay, config.graph.max_session_refs)

    # Append to the replay ledger post-commit so a crash mid-
    # write leaves the ledger intact and a clean retry can re-execute.
    if source_id and source_id not in processed_sources:
        processed_sources.append(source_id)
        overlay["processed_sources"] = processed_sources

    # Write overlay
    _write_overlay(overlay, overlay_path)

    # Recompile graph
    if recompile:
        build_graph(repo_root, global_vault=global_vault)
        logger.info("Graph recompiled after update")

    summary = {
        "confidence_events": confidence_events,
        "propagation_events": propagation_events,
        "pitfall_events": pitfall_events,
        "knowledge_events": knowledge_events,
        "session_node_id": session_node_id,
    }

    logger.info(
        "update_graph complete: %d confidence, %d propagation, "
        "%d pitfall, %d knowledge events",
        len(confidence_events),
        len(propagation_events),
        len(pitfall_events),
        len(knowledge_events),
    )

    return summary