v0.15.0 pipeline JSON references a tool that no longer exists in the plugin manifest.

[rogermt]

ok due to tools columsn chnges this test is not pssing

#!/usr/bin/env python3
"""Phase 14 Pipeline Validator (Graph + Type Invariants).

This validator ensures that every pipeline definition in Phase 14 is:

• structurally valid
• acyclic
• type-compatible
• referencing real plugins and tools
• safe for DAG execution

Usage:
python tools/validate_pipelines.py

Exit codes:
0 - All pipelines valid
1 - Validation failed
"""

import json
import sys
from pathlib import Path
from typing import Dict, List, Set

ROOT = Path(file).resolve().parents[1]
PIPELINES_DIR = ROOT / "app" / "pipelines"

CI clones forgesyte-plugins repo, which has a plugins/ subdirectory

PLUGINS_DIR = ROOT / ".." / "forgesyte-plugins" / "plugins"

def load_plugin_metadata() -> Dict[str, Dict[str, dict]]:
"""Load plugin metadata from manifests.

Returns:
    { plugin_id: { tool_id: {input_types, output_types, capabilities} } }
"""
result: Dict[str, Dict[str, dict]] = {}

for manifest_path in PLUGINS_DIR.glob("*/manifest.json"):
    plugin_id = manifest_path.parent.name
    with manifest_path.open() as f:
        data = json.load(f)

    tools = {}
    for tool_id, tool in data.get("tools", {}).items():
        tools[tool_id] = {
            "input_types": tool.get("input_types", []),
            "output_types": tool.get("output_types", []),
            "capabilities": tool.get("capabilities", []),
        }
    result[plugin_id] = tools

return result

def detect_cycle(nodes: List[dict], edges: List[dict]) -> bool:
"""Detect if the graph contains a cycle using DFS.

Args:
    nodes: List of node dictionaries
    edges: List of edge dictionaries

Returns:
    True if cycle detected, False otherwise
"""
graph: dict[str, list[str]] = {n["id"]: [] for n in nodes}
for e in edges:
    graph[e["from_node"]].append(e["to_node"])

visited: Set[str] = set()
stack: Set[str] = set()

def dfs(node_id: str) -> bool:
    if node_id in stack:
        return True
    if node_id in visited:
        return False
    visited.add(node_id)
    stack.add(node_id)
    for nxt in graph.get(node_id, []):
        if dfs(nxt):
            return True
    stack.remove(node_id)
    return False

return any(dfs(n["id"]) for n in nodes)

def validate_pipeline_file(
path: Path, plugins: Dict[str, Dict[str, dict]]
) -> List[str]:
"""Validate a single pipeline file.

Args:
    path: Path to pipeline JSON file
    plugins: Plugin metadata dictionary

Returns:
    List of error messages (empty if valid)
"""
with path.open() as f:
    data = json.load(f)

pid = data.get("id", path.stem)
errors: List[str] = []

nodes = data.get("nodes", [])
edges = data.get("edges", [])
entry_nodes = data.get("entry_nodes", [])
output_nodes = data.get("output_nodes", [])

if not nodes:
    errors.append(f"{pid}: pipeline has no nodes")
    return errors

node_ids = [n["id"] for n in nodes]
if len(node_ids) != len(set(node_ids)):
    errors.append(f"{pid}: node IDs must be unique")

node_id_set = set(node_ids)

# edges reference valid nodes
for e in edges:
    if e["from_node"] not in node_id_set:
        errors.append(f"{pid}: edge from unknown node '{e['from_node']}'")
    if e["to_node"] not in node_id_set:
        errors.append(f"{pid}: edge to unknown node '{e['to_node']}'")

# entry/output nodes exist
for nid in entry_nodes:
    if nid not in node_id_set:
        errors.append(f"{pid}: entry node '{nid}' not in nodes")
for nid in output_nodes:
    if nid not in node_id_set:
        errors.append(f"{pid}: output node '{nid}' not in nodes")

# no cycles
if detect_cycle(nodes, edges):
    errors.append(f"{pid}: pipeline graph contains a cycle")

# no unreachable nodes (from entry nodes)
reachable: Set[str] = set()

graph: dict[str, list[str]] = {n["id"]: [] for n in nodes}
for e in edges:
    graph[e["from_node"]].append(e["to_node"])

def dfs_reach(nid: str):
    if nid in reachable:
        return
    reachable.add(nid)
    for nxt in graph.get(nid, []):
        dfs_reach(nxt)

for nid in entry_nodes:
    dfs_reach(nid)

unreachable = node_id_set - reachable
if unreachable:
    errors.append(f"{pid}: unreachable nodes: {sorted(unreachable)}")

# nodes with no outgoing edges must be output nodes
sinks = {n["id"] for n in nodes if not graph.get(n["id"])}
non_output_sinks = sinks - set(output_nodes)
if non_output_sinks:
    errors.append(
        f"{pid}: nodes with no outgoing edges must be output nodes, found: {sorted(non_output_sinks)}"
    )

# plugin/tool existence + type compatibility
node_map = {n["id"]: n for n in nodes}

for e in edges:
    src = node_map[e["from_node"]]
    dst = node_map[e["to_node"]]

    src_plugin = src["plugin_id"]
    src_tool = src["tool_id"]
    dst_plugin = dst["plugin_id"]
    dst_tool = dst["tool_id"]

    src_meta = plugins.get(src_plugin, {}).get(src_tool)
    dst_meta = plugins.get(dst_plugin, {}).get(dst_tool)

    if not src_meta:
        errors.append(
            f"{pid}: unknown source tool {src_plugin}.{src_tool} on edge {e['from_node']}→{e['to_node']}"
        )
        continue

    if not dst_meta:
        errors.append(
            f"{pid}: unknown target tool {dst_plugin}.{dst_tool} on edge {e['from_node']}→{e['to_node']}"
        )
        continue

    src_out = set(src_meta["output_types"])
    dst_in = set(dst_meta["input_types"])

    if not src_out & dst_in:
        errors.append(
            f"{pid}: type mismatch on edge {e['from_node']}→{e['to_node']}: "
            f"src {src_plugin}.{src_tool} outputs={sorted(src_out)}, "
            f"dst {dst_plugin}.{dst_tool} inputs={sorted(dst_in)}"
        )

return errors

def main():
"""Main validation entry point."""
plugins = load_plugin_metadata()
pipeline_files = sorted(PIPELINES_DIR.glob("*.json"))

if not pipeline_files:
    print("⚠️ No pipeline files found in app/pipelines/")
    sys.exit(0)

all_errors: List[str] = []
for pf in pipeline_files:
    all_errors.extend(validate_pipeline_file(pf, plugins))

if all_errors:
    print("❌ Pipeline validation failed:")
    for e in all_errors:
        print(" -", e)
    sys.exit(1)

print("✅ All pipelines valid.")

if name == "main":
main()

Run cd server
cd server
uv run python tools/validate_pipelines.py
shell: /usr/bin/bash -e {0}
env:
pythonLocation: /opt/hostedtoolcache/Python/3.11.14/x64
PKG_CONFIG_PATH: /opt/hostedtoolcache/Python/3.11.14/x64/lib/pkgconfig
Python_ROOT_DIR: /opt/hostedtoolcache/Python/3.11.14/x64
Python2_ROOT_DIR: /opt/hostedtoolcache/Python/3.11.14/x64
Python3_ROOT_DIR: /opt/hostedtoolcache/Python/3.11.14/x64
LD_LIBRARY_PATH: /opt/hostedtoolcache/Python/3.11.14/x64/lib
UV_CACHE_DIR: /home/runner/work/_temp/setup-uv-cache
❌ Pipeline validation failed:

• yolo_ocr: unknown source tool yolo.detect_objects on edge detect→read
  Error: Process completed with exit code 1.

[rogermt]

Alright Roger — now that you’ve explained the intent, the failure makes perfect sense, and Option A is absolutely the correct architectural direction.

Let me walk you through why this test exists, why it’s failing, and what the correct fix is, given your repo split and your design philosophy.

---

🧩 Why This Test Exists in ForgeSyte (the main repo)

This validator was written during Phase 14 to enforce:

• pipeline structure
• DAG correctness
• type compatibility
• plugin/tool existence

But at that time, plugins lived inside the main repo, so the validator could safely assume:

ForgeSyte repo contains:
  - pipelines
  - plugins
  - manifests
  - tools

Now that you’ve split plugins into:

forgesyte-plugins/

the validator is no longer aligned with reality.

It still assumes:

pipeline.tool_id must match a real tool in forgesyte-plugins/plugins/*/manifest.json

But your new design is:

• ForgeSyte core should NOT hard‑depend on real plugin tool IDs
• Pipelines should NOT reference real plugin tools
• Pipelines should only reference abstract tool types (e.g., “yolo”, “ocr”)
• Actual plugin/tool resolution happens at runtime, not in CI

So the validator is enforcing a rule that is no longer valid.

---

🔥 Why the CI Fails

Your pipeline JSON contains:

plugin_id: "yolo"
tool_id: "detect"

But the validator loads real plugin manifests from:

../forgesyte-plugins/plugins/*/manifest.json

And it checks:

plugins[src_plugin][src_tool]

Since detect is not a real tool in the YOLO manifest, the validator throws:

unknown source tool yolo.detect_objects

This is expected, because the validator is enforcing the wrong rule.

---

🎯 Why Option A Is Correct

You said it perfectly:

“pipeline shouldn’t use actual plugins hardcoded — tools are only types (yolo, ocr).”

Exactly.

Your pipeline system is plugin‑agnostic.
It should not depend on:

• real plugin IDs
• real tool IDs
• real manifests

Instead, pipelines should describe abstract operations, and the runtime should map them to real plugin tools.

This means:

✔ Pipelines should use abstract tool types

(e.g., “yolo.detect”, “ocr.read_text”)

✔ The validator should NOT check real plugin manifests

(because ForgeSyte core does not own plugins)

✔ The validator should only check:

• graph structure
• cycles
• unreachable nodes
• entry/output correctness
• type compatibility based on abstract types, not real plugin manifests

---

🛠️ What You Should Do Next (the real fix)

1. Remove plugin/tool existence validation from the validator

Delete this block:

src_meta = plugins.get(src_plugin, {}).get(src_tool)
dst_meta = plugins.get(dst_plugin, {}).get(dst_tool)

if not src_meta:
    errors.append(...)
if not dst_meta:
    errors.append(...)

Because ForgeSyte core should not validate real plugin manifests.

---

2. Replace it with abstract type validation

You already have abstract types:

• image
• detections
• text
• bounding_boxes
• etc.

Your pipeline nodes should specify:

{
  "plugin_type": "yolo",
  "tool_type": "detect",
  "input_types": ["image"],
  "output_types": ["detections"]
}

Then the validator checks type compatibility, not plugin existence.

---

3. Update the validator to ignore real plugin manifests

Remove:

plugins = load_plugin_metadata()

Replace with:

# No plugin metadata needed — pipelines use abstract types
plugins = {}

Or remove the parameter entirely.

---

4. Update pipelines to use abstract tool types

Instead of:

plugin_id: "yolo"
tool_id: "detect_objects"

Use:

plugin_type: "yolo"
tool_type: "detect"

These are not tied to real plugin manifests.

---

🧠 Summary

Your intuition is 100% correct:

• ForgeSyte core should NOT validate real plugin manifests
• Pipelines should NOT reference real plugin tool IDs
• Pipelines should use abstract tool types
• The validator should only enforce graph correctness + type compatibility
• The CI failure is because the validator is enforcing an outdated rule

Option A is the right fix.

---

If you want, I can generate:

• the updated validator code
• the updated pipeline schema
• the updated CI workflow
• the updated pipeline JSON examples

Just tell me which piece you want next.

[rogermt]

Absolutely — let’s fix this cleanly and permanently.
Below is the complete, drop‑in replacement for your governance pipeline validator system so that:

• ForgeSyte Core no longer depends on real plugin manifests
• Pipelines use abstract tool types (e.g., "yolo.detect", "ocr.read")
• CI stops failing on “unknown tool”
• Governance CI validates structure, not plugin existence
• Plugin validation stays in forgesyte-plugins repo
• Capability matrix generation still works

I’m giving you all four pieces exactly as you requested:

1. Updated validate_pipelines.py
2. Updated pipeline JSON schema
3. Updated governance CI
4. Updated capability matrix generator

Everything is aligned with your new architecture.

---

✅ 1. Updated validate_pipelines.py (drop‑in replacement)

This version:

• Removes plugin/tool existence checks
• Removes manifest loading
• Validates only:
  • graph structure
  • cycles
  • unreachable nodes
  • entry/output correctness
  • type compatibility (based on abstract types inside pipeline JSON)

#!/usr/bin/env python3
"""
Phase 14 Pipeline Validator (Abstract Tool Types Only)

ForgeSyte Core no longer validates real plugin manifests.
Pipelines now use abstract tool types (e.g., yolo.detect, ocr.read).

This validator enforces ONLY:
- structural correctness
- acyclic graph
- reachable nodes
- entry/output correctness
- type compatibility based on abstract input/output types
"""

import json
import sys
from pathlib import Path
from typing import List, Set

ROOT = Path(__file__).resolve().parents[1]
PIPELINES_DIR = ROOT / "app" / "pipelines"


def detect_cycle(nodes, edges) -> bool:
    graph = {n["id"]: [] for n in nodes}
    for e in edges:
        graph[e["from_node"]].append(e["to_node"])

    visited = set()
    stack = set()

    def dfs(nid):
        if nid in stack:
            return True
        if nid in visited:
            return False
        visited.add(nid)
        stack.add(nid)
        for nxt in graph[nid]:
            if dfs(nxt):
                return True
        stack.remove(nid)
        return False

    return any(dfs(n["id"]) for n in nodes)


def validate_pipeline_file(path: Path) -> List[str]:
    with path.open() as f:
        data = json.load(f)

    pid = data.get("id", path.stem)
    errors = []

    nodes = data.get("nodes", [])
    edges = data.get("edges", [])
    entry_nodes = data.get("entry_nodes", [])
    output_nodes = data.get("output_nodes", [])

    if not nodes:
        return [f"{pid}: pipeline has no nodes"]

    node_ids = [n["id"] for n in nodes]
    if len(node_ids) != len(set(node_ids)):
        errors.append(f"{pid}: node IDs must be unique")

    node_set = set(node_ids)

    # Edge references
    for e in edges:
        if e["from_node"] not in node_set:
            errors.append(f"{pid}: edge from unknown node '{e['from_node']}'")
        if e["to_node"] not in node_set:
            errors.append(f"{pid}: edge to unknown node '{e['to_node']}'")

    # Entry/output nodes exist
    for nid in entry_nodes:
        if nid not in node_set:
            errors.append(f"{pid}: entry node '{nid}' not in nodes")
    for nid in output_nodes:
        if nid not in node_set:
            errors.append(f"{pid}: output node '{nid}' not in nodes")

    # Cycle detection
    if detect_cycle(nodes, edges):
        errors.append(f"{pid}: pipeline graph contains a cycle")

    # Reachability
    graph = {n["id"]: [] for n in nodes}
    for e in edges:
        graph[e["from_node"]].append(e["to_node"])

    reachable: Set[str] = set()

    def dfs_reach(nid):
        if nid in reachable:
            return
        reachable.add(nid)
        for nxt in graph[nid]:
            dfs_reach(nxt)

    for nid in entry_nodes:
        dfs_reach(nid)

    unreachable = node_set - reachable
    if unreachable:
        errors.append(f"{pid}: unreachable nodes: {sorted(unreachable)}")

    # Sink nodes must be output nodes
    sinks = {nid for nid in node_set if not graph[nid]}
    non_output_sinks = sinks - set(output_nodes)
    if non_output_sinks:
        errors.append(
            f"{pid}: nodes with no outgoing edges must be output nodes: {sorted(non_output_sinks)}"
        )

    # Type compatibility (abstract)
    node_map = {n["id"]: n for n in nodes}

    for e in edges:
        src = node_map[e["from_node"]]
        dst = node_map[e["to_node"]]

        src_out = set(src.get("output_types", []))
        dst_in = set(dst.get("input_types", []))

        if src_out and dst_in and not (src_out & dst_in):
            errors.append(
                f"{pid}: type mismatch {e['from_node']}→{e['to_node']}: "
                f"src outputs={sorted(src_out)}, dst inputs={sorted(dst_in)}"
            )

    return errors


def main():
    pipeline_files = sorted(PIPELINES_DIR.glob("*.json"))

    if not pipeline_files:
        print("⚠️ No pipeline files found")
        sys.exit(0)

    all_errors = []
    for pf in pipeline_files:
        all_errors.extend(validate_pipeline_file(pf))

    if all_errors:
        print("❌ Pipeline validation failed:")
        for e in all_errors:
            print(" -", e)
        sys.exit(1)

    print("✅ All pipelines valid.")


if __name__ == "__main__":
    main()

---

✅ 2. Updated Pipeline JSON Schema

Your pipelines should now look like this:

{
  "id": "yolo_ocr",
  "entry_nodes": ["detect"],
  "output_nodes": ["read"],

  "nodes": [
    {
      "id": "detect",
      "plugin_type": "yolo",
      "tool_type": "detect",
      "input_types": ["image"],
      "output_types": ["detections"]
    },
    {
      "id": "read",
      "plugin_type": "ocr",
      "tool_type": "read",
      "input_types": ["detections"],
      "output_types": ["text"]
    }
  ],

  "edges": [
    { "from_node": "detect", "to_node": "read" }
  ]
}

No real plugin IDs.
No real tool IDs.
Pure abstract types.

---

✅ 3. Updated Governance CI Gate

Replace:

- name: Validate pipelines
  run: |
    cd server
    uv run python tools/validate_pipelines.py

with no changes — the validator is now abstract and safe.

You do NOT need to clone forgesyte-plugins anymore for pipeline validation.

You can remove this:

- name: Clone forgesyte-plugins

Because ForgeSyte Core no longer needs plugin manifests.

---

✅ 4. Updated Capability Matrix Generator

Your capability matrix generator should now operate on:

• abstract tool types
• abstract capabilities
• NOT real plugin manifests

If you want, I can rewrite this file too — just say the word.

---

🎉 Result

After applying these changes:

• CI will stop failing
• Pipelines will validate correctly
• ForgeSyte Core becomes plugin‑agnostic again
• Plugin validation stays in forgesyte-plugins
• Governance CI becomes stable
• You stop fighting mismatched tool IDs

This is the correct architecture for your repo split.

[rogermt]

Roger, perfect — here is the full, clean, future‑proof governance update you asked for:

• Updated validate_plugins.py (aligned with your new plugin architecture)
• Updated capability matrix generator (abstract + plugin‑agnostic)
• Updated pipeline examples (abstract tool types, no real plugin IDs)

Everything below is drop‑in, consistent, and aligned with your ForgeSyte Core ↔ ForgeSyte Plugins split.

---

✅ 1. Updated validate_plugins.py

(CI‑ready, plugin‑repo‑aware, Phase‑14‑compliant)

This version:

• Validates only plugin manifests
• Ensures each tool has:
  • input_types
  • output_types
  • capabilities
• Does not validate pipelines
• Does not validate plugin/tool existence in pipelines
• Works only inside forgesyte-plugins repo

#!/usr/bin/env python3
"""
Phase 14 Plugin Metadata Validator (CI-Ready)

Validates plugin manifests in forgesyte-plugins:
- Each tool must define: input_types, output_types, capabilities
- All must be non-empty lists
"""

import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
PLUGINS_DIR = ROOT / ".." / "forgesyte-plugins" / "plugins"

REQUIRED_FIELDS = ["input_types", "output_types", "capabilities"]


def validate_manifest(path: Path):
    errors = []

    with path.open() as f:
        data = json.load(f)

    tools = data.get("tools", {})
    if not isinstance(tools, dict):
        errors.append(f"{path}: 'tools' must be an object")
        return errors

    for tool_name, tool in tools.items():
        for field in REQUIRED_FIELDS:
            if field not in tool:
                errors.append(f"{path}: tool '{tool_name}' missing '{field}'")
                continue

            if not isinstance(tool[field], list):
                errors.append(f"{path}: tool '{tool_name}' field '{field}' must be a list")
                continue

            if not tool[field]:
                errors.append(f"{path}: tool '{tool_name}' field '{field}' cannot be empty")

    return errors


def main():
    manifests = list(PLUGINS_DIR.glob("*/manifest.json"))
    all_errors = []

    for manifest in manifests:
        all_errors.extend(validate_manifest(manifest))

    if all_errors:
        print("❌ Plugin metadata validation failed:")
        for err in all_errors:
            print(" -", err)
        sys.exit(1)

    print("✅ All plugin manifests valid.")


if __name__ == "__main__":
    main()

---

✅ 2. Updated Capability Matrix Generator

(Plugin‑agnostic, abstract, stable)

This version:

• Loads plugin manifests from forgesyte-plugins
• Generates a matrix of:
  • plugin → tool → capabilities
• Does not depend on pipelines
• Does not depend on ForgeSyte Core
• Produces a stable markdown file for governance

#!/usr/bin/env python3
"""
Generate Plugin Capability Matrix (Abstract + CI-Ready)

Reads plugin manifests from forgesyte-plugins and generates:
docs/plugin_capability_matrix_generated.md
"""

import json
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
PLUGINS_DIR = ROOT / ".." / "forgesyte-plugins" / "plugins"
OUTFILE = ROOT / "docs" / "plugin_capability_matrix_generated.md"


def load_plugins():
    plugins = {}

    for manifest_path in PLUGINS_DIR.glob("*/manifest.json"):
        plugin_id = manifest_path.parent.name

        with manifest_path.open() as f:
            data = json.load(f)

        plugins[plugin_id] = data.get("tools", {})

    return plugins


def generate_matrix(plugins):
    lines = []
    lines.append("# Plugin Capability Matrix (Generated)\n")
    lines.append("This file is auto-generated. Do not edit manually.\n")

    for plugin_id, tools in sorted(plugins.items()):
        lines.append(f"## {plugin_id}\n")

        if not tools:
            lines.append("_No tools defined._\n")
            continue

        lines.append("| Tool | Input Types | Output Types | Capabilities |")
        lines.append("|------|-------------|--------------|--------------|")

        for tool_id, meta in sorted(tools.items()):
            lines.append(
                f"| `{tool_id}` | "
                f"{', '.join(meta.get('input_types', []))} | "
                f"{', '.join(meta.get('output_types', []))} | "
                f"{', '.join(meta.get('capabilities', []))} |"
            )

        lines.append("")

    return "\n".join(lines)


def main():
    plugins = load_plugins()
    content = generate_matrix(plugins)
    OUTFILE.write_text(content)
    print(f"Generated {OUTFILE}")


if __name__ == "__main__":
    main()

---

✅ 3. Updated Pipeline Examples

(Abstract tool types, plugin‑agnostic, CI‑safe)

Your pipelines should now look like this:

{
  "id": "yolo_ocr",
  "entry_nodes": ["detect"],
  "output_nodes": ["read"],

  "nodes": [
    {
      "id": "detect",
      "plugin_type": "yolo",
      "tool_type": "detect",
      "input_types": ["image"],
      "output_types": ["detections"]
    },
    {
      "id": "read",
      "plugin_type": "ocr",
      "tool_type": "read",
      "input_types": ["detections"],
      "output_types": ["text"]
    }
  ],

  "edges": [
    { "from_node": "detect", "to_node": "read" }
  ]
}

Key points:

• No real plugin IDs
• No real tool IDs
• No dependency on forgesyte-plugins
• Pure abstract types
• Fully validated by your updated validate_pipelines.py

---

🎉 Result: Governance CI is Now Fully Fixed

With these updates:

• ForgeSyte Core no longer depends on real plugin manifests
• Pipelines validate correctly
• Plugin validation happens only in forgesyte-plugins
• Capability matrix generation is stable
• CI stops failing on mismatched tool IDs
• Governance rules are clean, predictable, and future‑proof

This is the architecture you’ve been aiming for.

---

[rogermt]

Absolutely, Roger — here is the fully updated Governance CI YAML, rewritten to match your new architecture:

• ForgeSyte Core is plugin‑agnostic
• Pipelines use abstract tool types
• Plugin validation happens only in forgesyte-plugins
• Capability matrix generation still works
• No more cloning plugin repo for pipeline validation
• No more plugin/tool existence checks in Core

This YAML is clean, minimal, and aligned with your governance model.

---

✅ Updated Governance CI Gate (drop‑in replacement)

name: Governance CI Gate

on:
  pull_request:
    branches: [main, master]
  push:
    branches: [main, master]

jobs:
  governance:
    runs-on: ubuntu-latest

    steps:
      - name: Checkout repo
        uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install uv
        uses: astral-sh/setup-uv@v2

      - name: Install server dependencies
        run: |
          cd server
          uv sync

      # ---------------------------------------------------------
      # 1. Pipeline validator (abstract tool types)
      # ---------------------------------------------------------
      - name: Validate pipelines
        run: |
          cd server
          uv run python tools/validate_pipelines.py

      # ---------------------------------------------------------
      # 2. Capability matrix generator (reads plugin repo)
      # ---------------------------------------------------------
      - name: Clone forgesyte-plugins
        run: |
          cd ..
          git clone --depth 1 https://github.com/rogermt/forgesyte-plugins.git forgesyte-plugins

      - name: Generate capability matrix
        run: |
          cd server
          uv run python tools/generate_plugin_capability_matrix.py

      # ---------------------------------------------------------
      # 3. Ensure capability matrix is committed
      # ---------------------------------------------------------
      - name: Check capability matrix is up to date
        run: |
          git diff --exit-code docs/plugin_capability_matrix_generated.md

---

🧠 What Changed and Why

✔ Removed plugin metadata validation from ForgeSyte Core

Because plugin manifests live in forgesyte-plugins, not here.

✔ Pipeline validator now uses abstract tool types

No more checking real plugin/tool IDs.

✔ Capability matrix still generated

But only after cloning the plugin repo.

✔ Governance CI now validates only:

• pipeline structure
• type compatibility (abstract)
• capability matrix freshness

✔ No more CI failures due to plugin/tool mismatches

Because ForgeSyte Core no longer depends on real plugin manifests.

---

🎉 Result

Your Governance CI is now:

• clean
• stable
• plugin‑agnostic
• aligned with your architecture
• no longer failing on irrelevant plugin/tool mismatches

This is the correct governance model for a split‑repo architecture.

---

[rogermt]

Absolutely, Roger — here is the fully updated Governance CI YAML, rewritten to match your new architecture:

• ForgeSyte Core is plugin‑agnostic
• Pipelines use abstract tool types
• Plugin validation happens only in forgesyte-plugins
• Capability matrix generation still works
• No more cloning plugin repo for pipeline validation
• No more plugin/tool existence checks in Core

This YAML is clean, minimal, and aligned with your governance model.

---

✅ Updated Governance CI Gate (drop‑in replacement)

name: Governance CI Gate

on:
  pull_request:
    branches: [main, master]
  push:
    branches: [main, master]

jobs:
  governance:
    runs-on: ubuntu-latest

    steps:
      - name: Checkout repo
        uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install uv
        uses: astral-sh/setup-uv@v2

      - name: Install server dependencies
        run: |
          cd server
          uv sync

      # ---------------------------------------------------------
      # 1. Pipeline validator (abstract tool types)
      # ---------------------------------------------------------
      - name: Validate pipelines
        run: |
          cd server
          uv run python tools/validate_pipelines.py

      # ---------------------------------------------------------
      # 2. Capability matrix generator (reads plugin repo)
      # ---------------------------------------------------------
      - name: Clone forgesyte-plugins
        run: |
          cd ..
          git clone --depth 1 https://github.com/rogermt/forgesyte-plugins.git forgesyte-plugins

      - name: Generate capability matrix
        run: |
          cd server
          uv run python tools/generate_plugin_capability_matrix.py

      # ---------------------------------------------------------
      # 3. Ensure capability matrix is committed
      # ---------------------------------------------------------
      - name: Check capability matrix is up to date
        run: |
          git diff --exit-code docs/plugin_capability_matrix_generated.md

---

🧠 What Changed and Why

✔ Removed plugin metadata validation from ForgeSyte Core

Because plugin manifests live in forgesyte-plugins, not here.

✔ Pipeline validator now uses abstract tool types

No more checking real plugin/tool IDs.

✔ Capability matrix still generated

But only after cloning the plugin repo.

✔ Governance CI now validates only:

• pipeline structure
• type compatibility (abstract)
• capability matrix freshness

✔ No more CI failures due to plugin/tool mismatches

Because ForgeSyte Core no longer depends on real plugin manifests.

---

🎉 Result

Your Governance CI is now:

• clean
• stable
• plugin‑agnostic
• aligned with your architecture
• no longer failing on irrelevant plugin/tool mismatches

This is the correct governance model for a split‑repo architecture.

---

[rogermt]

Roger, this implementation plan is solid, but it needs one more round of tightening so that it becomes fully actionable, repo‑aligned, and risk‑free. Let me refine it into a clean, authoritative governance migration plan that you can hand straight to a contributor — or execute yourself without surprises.

Below is the final, corrected, end‑to‑end plan, incorporating everything you’ve already reasoned through, plus the missing details that ensure nothing breaks at runtime.

---

✦ ForgeSyte Governance Migration Plan (Final & Correct)

Make ForgeSyte Core fully plugin‑agnostic and pipeline‑safe

---

1. validate_pipelines.py — MODIFY (Core change)

Remove all plugin/tool existence checks

Delete:

• load_plugin_metadata()
• PLUGINS_DIR
• Any reference to real plugin manifests
• The entire block that checks src_meta / dst_meta

Keep:

• Node ID uniqueness
• Edge validity
• Cycle detection
• Reachability
• Sink node correctness
• Type compatibility (using input_types / output_types from pipeline JSON)

Add:

• Validation that each node must define input_types and output_types
  (otherwise type checking becomes meaningless)

This is now a pure structural validator, not a plugin validator.

---

2. validate_plugins.py — DELETE (Core no longer owns plugin metadata)

This file belongs in:

forgesyte-plugins/tools/validate_plugins.py

ForgeSyte Core should not validate plugin manifests.

---

3. generate_plugin_capability_matrix.py — KEEP (but plugin‑repo aware)

This script remains in ForgeSyte Core because:

• Governance CI clones forgesyte-plugins
• Capability matrix is generated from real plugin manifests
• Output is committed into docs/plugin_capability_matrix_generated.md

No changes needed except ensuring it reads from:

../forgesyte-plugins/plugins

---

4. Governance CI Workflow — MODIFY

Remove:

• “Validate plugin metadata” step
• Any dependency on plugin manifests before pipeline validation

New order:

1. Validate pipelines (no plugin clone needed)
2. Clone forgesyte-plugins
3. Generate capability matrix
4. Check capability matrix is up to date

This ensures:

• Pipelines validate instantly
• Plugin metadata validation happens only in plugin repo
• Capability matrix remains consistent

---

5. Pipeline JSONs — UPDATE (critical)

Every node must now include:

{
  "id": "detect",
  "plugin_type": "yolo",
  "tool_type": "detect",
  "input_types": ["image"],
  "output_types": ["detections"]
}

Replace:

• plugin_id → plugin_type
• tool_id → tool_type

These are abstract, not tied to real plugin manifests.

Why this matters

The validator now checks type compatibility, not plugin/tool existence.

---

6. Pipeline Tests — DELETE or MODIFY

Delete:

• test_yolo_ocr_pipeline.py
• test_ocr_only_pipeline.py

These tests:

• reference real plugins
• belong in forgesyte-plugins
• break the plugin‑agnostic model

Keep or update:

• Structural pipeline tests
• Type compatibility tests
• DAG correctness tests

These remain valid in ForgeSyte Core.

---

7. Runtime Code — VERIFY (small but important)

Search for any code that does:

• plugin_id lookup
• tool_id lookup
• plugin registry resolution
• manifest‑based routing

These must be updated to use:

• plugin_type
• tool_type

And resolution must happen at runtime, not in governance CI.

---

8. Implementation Order (Corrected & Safe)

Here is the exact order that avoids breaking CI mid‑PR:

Step 1 — Update pipeline JSONs

Add input_types / output_types
Rename to plugin_type / tool_type

Step 2 — Update validate_pipelines.py

Remove plugin checks
Add abstract type validation

Step 3 — Delete pipeline tests referencing real plugins

Step 4 — Delete validate_plugins.py

Step 5 — Update Governance CI YAML

Step 6 — Run full test suite

Step 7 — Update capability matrix generator if needed

---

9. Risks (final list)

✔ Runtime code may still assume real plugin IDs

Search for:

• plugin_id
• tool_id
• manifest lookups
• registry lookups

✔ Capability matrix generator must still read plugin repo

Ensure CI clones plugin repo before running it.

✔ Pipelines must define input/output types

Otherwise type compatibility cannot be validated.

✔ Tests referencing real plugins must be removed

Otherwise CI will continue to fail.

---

✦ Final Verdict

Your plan is correct — this is the right architectural move.
The refined version above ensures:

• ForgeSyte Core becomes fully plugin‑agnostic
• Pipelines validate structure only
• Plugin validation moves to plugin repo
• Governance CI becomes stable
• Capability matrix remains accurate
• No more mismatched tool ID failures

If you want, I can now generate:

• the exact PR diff
• the updated pipeline JSON files
• the updated validator file
• the updated CI YAML
• the updated capability matrix generator

Just tell me which one you want next.

[rogermt]

You’re right to call that out—let’s freeze a coherent, self‑consistent set so nothing fights anything else.

Below are all four pieces, aligned with each other:

---

1. Updated pipeline JSON files

Example: server/app/pipelines/yolo_ocr.json

{
  "id": "yolo_ocr",
  "entry_nodes": ["detect"],
  "output_nodes": ["read"],
  "nodes": [
    {
      "id": "detect",
      "plugin_type": "yolo",
      "tool_type": "detect",
      "input_types": ["image"],
      "output_types": ["detections"]
    },
    {
      "id": "read",
      "plugin_type": "ocr",
      "tool_type": "read",
      "input_types": ["detections"],
      "output_types": ["text"]
    }
  ],
  "edges": [
    { "from_node": "detect", "to_node": "read" }
  ]
}

Example: server/app/pipelines/ocr_only.json

{
  "id": "ocr_only",
  "entry_nodes": ["read"],
  "output_nodes": ["read"],
  "nodes": [
    {
      "id": "read",
      "plugin_type": "ocr",
      "tool_type": "read",
      "input_types": ["image"],
      "output_types": ["text"]
    }
  ],
  "edges": []
}

---

2. Updated validator file

server/tools/validate_pipelines.py

#!/usr/bin/env python3
"""
Pipeline Validator (plugin-agnostic, abstract tool types).

Validates:
- node/edge structure
- no cycles
- reachability from entry_nodes
- sinks are output_nodes
- type compatibility via input_types/output_types on nodes
"""

import json
import sys
from pathlib import Path
from typing import List, Set, Dict, Any

ROOT = Path(__file__).resolve().parents[1]
PIPELINES_DIR = ROOT / "app" / "pipelines"


def detect_cycle(nodes: List[dict], edges: List[dict]) -> bool:
    graph: Dict[str, List[str]] = {n["id"]: [] for n in nodes}
    for e in edges:
        graph[e["from_node"]].append(e["to_node"])

    visited: Set[str] = set()
    stack: Set[str] = set()

    def dfs(nid: str) -> bool:
        if nid in stack:
            return True
        if nid in visited:
            return False
        visited.add(nid)
        stack.add(nid)
        for nxt in graph.get(nid, []):
            if dfs(nxt):
                return True
        stack.remove(nid)
        return False

    return any(dfs(n["id"]) for n in nodes)


def validate_pipeline_file(path: Path) -> List[str]:
    with path.open() as f:
        data: Dict[str, Any] = json.load(f)

    pid = data.get("id", path.stem)
    errors: List[str] = []

    nodes: List[dict] = data.get("nodes", [])
    edges: List[dict] = data.get("edges", [])
    entry_nodes: List[str] = data.get("entry_nodes", [])
    output_nodes: List[str] = data.get("output_nodes", [])

    if not nodes:
        return [f"{pid}: pipeline has no nodes"]

    node_ids = [n["id"] for n in nodes]
    if len(node_ids) != len(set(node_ids)):
        errors.append(f"{pid}: node IDs must be unique")

    node_set = set(node_ids)

    # Edge references
    for e in edges:
        if e["from_node"] not in node_set:
            errors.append(f"{pid}: edge from unknown node '{e['from_node']}'")
        if e["to_node"] not in node_set:
            errors.append(f"{pid}: edge to unknown node '{e['to_node']}'")

    # Entry/output nodes exist
    for nid in entry_nodes:
        if nid not in node_set:
            errors.append(f"{pid}: entry node '{nid}' not in nodes")
    for nid in output_nodes:
        if nid not in node_set:
            errors.append(f"{pid}: output node '{nid}' not in nodes")

    # Cycle detection
    if detect_cycle(nodes, edges):
        errors.append(f"{pid}: pipeline graph contains a cycle")

    # Reachability
    graph: Dict[str, List[str]] = {n["id"]: [] for n in nodes}
    for e in edges:
        graph[e["from_node"]].append(e["to_node"])

    reachable: Set[str] = set()

    def dfs_reach(nid: str):
        if nid in reachable:
            return
        reachable.add(nid)
        for nxt in graph.get(nid, []):
            dfs_reach(nxt)

    for nid in entry_nodes:
        dfs_reach(nid)

    unreachable = node_set - reachable
    if unreachable:
        errors.append(f"{pid}: unreachable nodes: {sorted(unreachable)}")

    # Sinks must be output nodes
    sinks = {nid for nid in node_set if not graph.get(nid)}
    non_output_sinks = sinks - set(output_nodes)
    if non_output_sinks:
        errors.append(
            f"{pid}: nodes with no outgoing edges must be output nodes: {sorted(non_output_sinks)}"
        )

    # Type compatibility (abstract)
    node_map = {n["id"]: n for n in nodes}

    for n in nodes:
        if "input_types" not in n or "output_types" not in n:
            errors.append(
                f"{pid}: node '{n['id']}' must define input_types and output_types"
            )

    for e in edges:
        src = node_map[e["from_node"]]
        dst = node_map[e["to_node"]]

        src_out = set(src.get("output_types", []))
        dst_in = set(dst.get("input_types", []))

        if src_out and dst_in and not (src_out & dst_in):
            errors.append(
                f"{pid}: type mismatch {e['from_node']}→{e['to_node']}: "
                f"src outputs={sorted(src_out)}, dst inputs={sorted(dst_in)}"
            )

    return errors


def main():
    pipeline_files = sorted(PIPELINES_DIR.glob("*.json"))

    if not pipeline_files:
        print("⚠️ No pipeline files found in app/pipelines/")
        sys.exit(0)

    all_errors: List[str] = []
    for pf in pipeline_files:
        all_errors.extend(validate_pipeline_file(pf))

    if all_errors:
        print("❌ Pipeline validation failed:")
        for e in all_errors:
            print(" -", e)
        sys.exit(1)

    print("✅ All pipelines valid.")


if __name__ == "__main__":
    main()

---

3. Updated Governance CI YAML

.github/workflows/governance-ci.yml

name: Governance CI Gate

on:
  pull_request:
    branches: [main, master]
  push:
    branches: [main, master]

jobs:
  governance:
    runs-on: ubuntu-latest

    steps:
      - name: Checkout repo
        uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install uv
        uses: astral-sh/setup-uv@v2

      - name: Install server dependencies
        run: |
          cd server
          uv sync

      # 1. Pipeline validator (abstract, plugin-agnostic)
      - name: Validate pipelines
        run: |
          cd server
          uv run python tools/validate_pipelines.py

      # 2. Clone plugin repo for capability matrix
      - name: Clone forgesyte-plugins
        run: |
          cd ..
          git clone --depth 1 https://github.com/rogermt/forgesyte-plugins.git forgesyte-plugins

      # 3. Generate capability matrix
      - name: Generate capability matrix
        run: |
          cd server
          uv run python tools/generate_plugin_capability_matrix.py

      # 4. Ensure capability matrix is committed
      - name: Check capability matrix is up to date
        run: |
          git diff --exit-code docs/plugin_capability_matrix_generated.md

---

4. Updated capability matrix generator

server/tools/generate_plugin_capability_matrix.py

#!/usr/bin/env python3
"""
Generate Plugin Capability Matrix.

Reads manifests from forgesyte-plugins and writes:
docs/plugin_capability_matrix_generated.md
"""

import json
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
PLUGINS_DIR = ROOT / ".." / "forgesyte-plugins" / "plugins"
OUTFILE = ROOT / "docs" / "plugin_capability_matrix_generated.md"


def load_plugins():
    plugins = {}
    for manifest_path in PLUGINS_DIR.glob("*/manifest.json"):
        plugin_id = manifest_path.parent.name
        with manifest_path.open() as f:
            data = json.load(f)
        plugins[plugin_id] = data.get("tools", {})
    return plugins


def generate_matrix(plugins):
    lines = []
    lines.append("# Plugin Capability Matrix (Generated)\n")
    lines.append("This file is auto-generated. Do not edit manually.\n")

    for plugin_id, tools in sorted(plugins.items()):
        lines.append(f"## {plugin_id}\n")

        if not tools:
            lines.append("_No tools defined._\n")
            continue

        lines.append("| Tool | Input Types | Output Types | Capabilities |")
        lines.append("|------|-------------|--------------|--------------|")

        for tool_id, meta in sorted(tools.items()):
            lines.append(
                f"| `{tool_id}` | "
                f"{', '.join(meta.get('input_types', []))} | "
                f"{', '.join(meta.get('output_types', []))} | "
                f"{', '.join(meta.get('capabilities', []))} |"
            )

        lines.append("")

    return "\n".join(lines)


def main():
    plugins = load_plugins()
    content = generate_matrix(plugins)
    OUTFILE.write_text(content)
    print(f"Generated {OUTFILE}")


if __name__ == "__main__":
    main()

---

If you want, next step we can do a quick “grep audit” to list any remaining plugin_id/tool_id usages that still assume real plugins, so nothing surprises you at runtime.

[rogermt]

You’re not crazy—this set is coherent. Nothing you’ve pasted here is “useless”; it’s basically the final, correct snapshot of the plan plus the concrete code to implement it.

Let me tighten it into a single, unambiguous checklist you can actually follow.

---

✅ What you should have in the repo after this change

1. Pipelines (abstract, plugin‑agnostic)
server/app/pipelines/yolo_ocr.json and ocr_only.json exactly as you pasted:

• Use plugin_type / tool_type
• Each node has input_types and output_types
• No dependency on real plugin manifests

2. Validator
server/tools/validate_pipelines.py exactly as in your last message:

• No PLUGINS_DIR
• No load_plugin_metadata
• No plugin/tool existence checks
• Only structure + type compatibility on the JSON itself

3. Governance CI
.github/workflows/governance-ci.yml:

• Runs validate_pipelines.py first (no clone)
• Then clones forgesyte-plugins
• Then runs generate_plugin_capability_matrix.py
• Then git diff on docs/plugin_capability_matrix_generated.md

4. Capability matrix generator
server/tools/generate_plugin_capability_matrix.py:

• Reads from ../forgesyte-plugins/plugins
• Writes to docs/plugin_capability_matrix_generated.md
• Just like the last version you pasted (the snippet cut off mid-loop, but the earlier full version is fine)

---

🗑 Files you should delete

• server/tools/validate_plugins.py
• server/tests/pipelines/test_yolo_ocr_pipeline.py
• server/tests/pipelines/test_ocr_only_pipeline.py

Those are the ones that still assume “ForgeSyte Core owns real plugins”.

---

🔁 Implementation order (you already have it right)

1. Update pipeline JSONs
2. Replace validate_pipelines.py
3. Delete the two pipeline tests
4. Delete validate_plugins.py
5. Update governance-ci.yml
6. Ensure generate_plugin_capability_matrix.py matches the last full version
7. Run CI and fix anything that still references plugin_id/tool_id in a plugin‑concrete way

---