feat(knowledge): resolution reuse — RCA becomes a reusable ARM asset with Next-Best-Action

Makefile

@@ -1,6 +1,6 @@
 # SourceOS Continuum — lifecycle entry points.
 # Control-plane targets delegate to Makefile.porter (the rehomed Porter control plane).
-.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute mesh-demo grant commons mcp spine run loop verify lease sphere push push-webhook rollback edge login sso provision deploy inference availability
+.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute mesh-demo grant commons reuse mcp spine run loop verify lease sphere push push-webhook rollback edge login sso provision deploy inference availability
 
 validate: ## repo hygiene + CapD validity
 	python3 tools/validate.py
@@ -27,6 +27,9 @@ grant: ## demo the zero-trust attach flow: Attest -> Decide -> Grant -> verify-a
 commons: ## reproducible knowledge commons: ingest the estate's CapDs + workloads as citable records
 	python3 tools/commons.py
 
+reuse: ## resolution reuse (ARM): deposit an RCA Resolution asset + Next-Best-Action lookup + feedback
+	cd tools && python3 resolution_registry.py
+
 spine: ## run the full execution spine demo: place -> grant -> verify -> dispatch -> sealed receipt
 	cd tools && python3 executor.py
 

capd/resolution-reuse.mesh.capd.json

@@ -0,0 +1,27 @@
+{
+  "capability_id": "caps.knowledge.resolution-reuse@0.1.0",
+  "kind": "knowledge.arm-resolution",
+  "status": "experimental",
+  "name": "Resolution reuse — RCA becomes a reusable ARM/commons asset with Next-Best-Action",
+  "description": "Closes the ARM feedback loop for root-cause work (the IBM Asset-Reuse-Manager pattern: Domain → Category → Asset → Recommendation → Feedback; and 'Next Best Action for a Case'). A Resolution asset produced by the RCA pipeline — a tagged, content-addressed record carrying the root-cause graph SVG + remediation — is deposited into the real continuum commons as asset_type='resolution', keyed by its failure-class category, and marked reproducible iff it carries the graph digest that rebuilds its evidence. When the same failure class recurs, next_best_action() retrieves the prior resolution to apply (recommendation, not a fresh investigation), and apply_resolution() feeds the ARM reuse loop (commons.record_use) so the most-reused, most-successful resolutions rank first. A thin adapter over the existing commons — no parallel store — so resolutions are first-class, searchable, reusable estate knowledge.",
+  "links": {
+    "engine": "tools/resolution_registry.py",
+    "commons": "tools/commons.py",
+    "producer": "prophet-platform tools/resolution_asset.py (the RCA→Resolution asset + SVG)",
+    "graph_ingest": "prophet-platform tools/rca_to_knowledge_update.py (RCA→HellGraph)",
+    "reference_pattern": "IBM Asset Reuse Manager (ARM) — Domain/Category/Asset/Recommendation/Feedback + Next-Best-Action — met sovereign: content-addressed, reproducibility-gated, reuse-scored"
+  },
+  "composes_with": {
+    "commons": "caps.knowledge.commons@0.1.0",
+    "control_plane": "caps.infra.paas.continuum-local@0.1.0",
+    "scales_up_to": "caps.infra.cluster-scaleup.hyperswarm@0.1.0"
+  },
+  "policy": {
+    "availability": "needs-work",
+    "reproducibility_gated": true,
+    "next_best_action": true,
+    "reuse_feedback_loop": true,
+    "no_parallel_store": true,
+    "evidence_emitting": true
+  }
+}

tools/resolution_registry.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""Resolution reuse — deposit RCA Resolution assets into the commons/ARM and find them next time.
+
+Closes the ARM feedback loop for root-cause work (the IBM ARM diagram: Domain → Category → Asset →
+Recommendation → Feedback; and "Next Best Action for a Case"): a Resolution asset (produced by the
+RCA pipeline — a tagged, content-addressed record carrying the root-cause graph SVG + remediation) is
+**deposited into the real continuum commons** as `asset_type="resolution"`, keyed by its failure-class
+category. When the same failure class recurs, `next_best_action()` retrieves the prior resolution to
+apply — recommendation, not a fresh investigation — and `apply_resolution()` feeds the ARM reuse loop
+(`commons.record_use`) so the most-reused, most-successful resolutions rank first.
+
+Thin adapter over `commons.Commons` — no parallel store. stdlib only.
+"""
+from __future__ import annotations
+
+import commons as cm
+
+
+def deposit_resolution(commons: cm.Commons, resolution: dict) -> dict:
+    """Deposit a Resolution asset record into the commons as a first-class, reusable asset. The record
+    is the portable shape the RCA pipeline emits (asset_type='resolution', category=failure-class,
+    tags, content{...graph_digest, remediation, svg...})."""
+    content = resolution.get("content", resolution)
+    graph_digest = content.get("graph_digest")
+    provenance = {"source_ref": resolution.get("commons_id", "resolution"),
+                  "tags": resolution.get("tags", [])}
+    # a resolution is 'reproducible' iff it carries the graph digest that rebuilds its evidence.
+    if graph_digest:
+        provenance["source_digest"] = graph_digest
+        provenance["sbom_digest"] = graph_digest  # the graph IS the reproducible evidence bundle
+    return commons.deposit(
+        domain=resolution.get("domain", "governance/migration"),
+        name=resolution.get("name", "resolution-unknown"),
+        version=resolution.get("version", "v0.1"),
+        asset_type="resolution",
+        category=resolution.get("category", "resolution"),
+        content=content,
+        provenance=provenance,
+        semantic_action={"recommendation": resolution.get("recommendation"),
+                         "tags": resolution.get("tags", [])})
+
+
+def next_best_action(commons: cm.Commons, *, failure_class: str, from_lang: str | None = None,
+                     to_lang: str | None = None, limit: int = 3) -> list[dict]:
+    """ARM 'Next Best Action for a case': given a NEW failure's class (and optionally the swap langs),
+    return the prior Resolution assets to apply, most-reused/most-successful first. Empty = no prior
+    resolution — genuinely new failure."""
+    pool = commons.search(asset_type="resolution", category=failure_class)
+    if from_lang or to_lang:
+        def langs_match(r):
+            c = r.get("content", {})
+            return ((not from_lang or c.get("root_cause", "").find(from_lang) >= 0
+                     or from_lang in (r.get("semantic_action", {}).get("tags") or []))
+                    and (not to_lang or to_lang in (r.get("semantic_action", {}).get("tags") or [])))
+        pool = [r for r in pool if langs_match(r)]
+    return sorted(pool, key=lambda r: (r["reuse"]["score"], r["reuse"]["uses"]), reverse=True)[:limit]
+
+
+def apply_resolution(commons: cm.Commons, commons_id: str, *, outcome: str = "ok") -> dict | None:
+    """Record that a resolution was applied to a case — the ARM use/evaluate feedback that raises its
+    reuse score, so proven resolutions surface first next time."""
+    return commons.record_use(commons_id, outcome=outcome)
+
+
+if __name__ == "__main__":
+    import json
+
+    commons = cm.Commons()
+    # a Resolution asset from the RCA pipeline (Nix→Guix percolation).
+    resolution = {
+        "domain": "governance/migration", "name": "resolution-ADR-0001-nix-to-guix", "version": "v0.1",
+        "category": "dependency-swap-percolation", "tags": ["nix", "guix", "swap", "rca"],
+        "recommendation": "apply Firewall #1 (adr_swap_gate) before authoring in scope; reuse this plan",
+        "content": {"root_cause": "a swap nix→guix built no dependency graph, so no control caught new "
+                    "FROM artifacts", "graph_digest": "sha256:" + "ab" * 32, "residual": 63},
+    }
+    dep = deposit_resolution(commons, resolution)
+
+    # later: a NEW case of the same failure class arrives → Next Best Action finds the prior resolution.
+    nba = next_best_action(commons, failure_class="dependency-swap-percolation",
+                           from_lang="nix", to_lang="guix")
+    for r in nba:
+        apply_resolution(commons, r["commons_id"])  # applied to the new case (feedback)
+    applied = commons.resolve(dep["commons_id"])
+
+    print(json.dumps({
+        "deposited": {"commons_id": dep["commons_id"], "asset_type": dep["asset_type"],
+                      "category": dep["category"], "reproducibility": dep["reproducibility"]},
+        "next_best_action_found": [r["commons_id"] for r in nba],
+        "recommendation": nba[0]["semantic_action"]["recommendation"] if nba else None,
+        "reuse_after_apply": applied["reuse"],
+    }, indent=2))

tools/test_resolution_registry.py

@@ -0,0 +1,71 @@
+#!/usr/bin/env python3
+"""Tests for resolution reuse (deposit → Next-Best-Action → ARM feedback) over the real commons."""
+import commons as cm
+import resolution_registry as rr
+
+
+def _resolution(name="resolution-ADR-0001", cat="dependency-swap-percolation", digest="sha256:" + "ab" * 32,
+                tags=("nix", "guix", "swap")):
+    return {"domain": "governance/migration", "name": name, "version": "v0.1", "category": cat,
+            "tags": list(tags), "recommendation": "apply Firewall #1",
+            "content": {"root_cause": "swap nix→guix built no graph", "graph_digest": digest}}
+
+
+def test_deposit_is_a_reproducible_resolution_asset():
+    c = cm.Commons()
+    rec = rr.deposit_resolution(c, _resolution())
+    assert rec["asset_type"] == "resolution" and rec["category"] == "dependency-swap-percolation"
+    assert rec["reproducibility"] == "reproducible"  # carries graph_digest
+    assert c.search(asset_type="resolution") == [rec]
+
+
+def test_resolution_without_a_graph_digest_is_only_declared():
+    c = cm.Commons()
+    r = _resolution(digest=None)
+    r["content"]["graph_digest"] = None
+    rec = rr.deposit_resolution(c, r)
+    assert rec["reproducibility"] == "declared"
+
+
+def test_next_best_action_finds_the_prior_resolution_for_the_failure_class():
+    c = cm.Commons()
+    dep = rr.deposit_resolution(c, _resolution())
+    nba = rr.next_best_action(c, failure_class="dependency-swap-percolation", from_lang="nix", to_lang="guix")
+    assert [r["commons_id"] for r in nba] == [dep["commons_id"]]
+    assert nba[0]["semantic_action"]["recommendation"] == "apply Firewall #1"
+
+
+def test_a_genuinely_new_failure_class_has_no_prior_resolution():
+    c = cm.Commons()
+    rr.deposit_resolution(c, _resolution())
+    assert rr.next_best_action(c, failure_class="some-brand-new-failure") == []
+
+
+def test_apply_records_reuse_and_ranks_most_reused_first():
+    c = cm.Commons()
+    a = rr.deposit_resolution(c, _resolution(name="res-a"))
+    b = rr.deposit_resolution(c, _resolution(name="res-b"))
+    # apply b twice, a once → b should rank first
+    rr.apply_resolution(c, a["commons_id"])
+    rr.apply_resolution(c, b["commons_id"])
+    rr.apply_resolution(c, b["commons_id"])
+    ranked = rr.next_best_action(c, failure_class="dependency-swap-percolation", limit=5)
+    assert ranked[0]["commons_id"] == b["commons_id"]
+    assert ranked[0]["reuse"]["uses"] == 2
+
+
+def test_lang_filter_excludes_other_swaps():
+    c = cm.Commons()
+    rr.deposit_resolution(c, _resolution(tags=("python2", "python3", "swap")))
+    # a nix→guix case should not match a python2→3 resolution in the same failure class
+    nba = rr.next_best_action(c, failure_class="dependency-swap-percolation", from_lang="nix", to_lang="guix")
+    assert nba == []
+
+
+if __name__ == "__main__":
+    import sys
+    fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
+    for fn in fns:
+        fn()
+    print(f"ok: {len(fns)} resolution-registry tests passed")
+    sys.exit(0)

tools/validate.py

@@ -22,6 +22,7 @@
     "capd/devspace.local-dev.capd.json",
     "capd/cloudshell-fog.capd.json",
     "capd/knowledge-commons.mesh.capd.json",
+    "capd/resolution-reuse.mesh.capd.json",
     "capd/self-healing-loop.mesh.capd.json",
     "capd/volunteer-mesh-verification.mesh.capd.json",
     "capd/data-spheres.mesh.capd.json",
@@ -37,6 +38,7 @@
     "tools/mesh_telemetry.py",
     "tools/mcp_a2a_grant.py",
     "tools/commons.py",
+    "tools/resolution_registry.py",
     "tools/mcp_ops_server.py",
     "tools/executor.py",
     "tools/sourceosctl.py",
@@ -65,6 +67,7 @@
     "capd/devspace.local-dev.capd.json": "caps.dev.devspace-inner-loop",
     "capd/cloudshell-fog.capd.json": "caps.compute.cloudshell-fog",
     "capd/knowledge-commons.mesh.capd.json": "caps.knowledge.commons",
+    "capd/resolution-reuse.mesh.capd.json": "caps.knowledge.resolution-reuse",
     "capd/self-healing-loop.mesh.capd.json": "caps.compute.self-healing-loop",
     "capd/volunteer-mesh-verification.mesh.capd.json": "caps.compute.volunteer-mesh-verification",
     "capd/data-spheres.mesh.capd.json": "caps.data.spheres",