GraFOSa Pipeline Framework — Advisory & Documentation

GraFOSa Pipeline Framework — Advisory & Documentation

Date: 2026-02-27
Status: Implemented, 843 tests passing
Files: src/grafosa/pipeline/ (7 files), refactored intel.py + strategy.py

What We Built

A config-driven pipeline framework that sits between GraFOSa's raw graph
analysis and downstream consumers (learning pipelines, LLM agents, Conductor
workflows). It solves three problems:

1. Hardcoded thresholds — All scoring formulas, risk levels, and grading
thresholds were baked into intel.py and strategy.py. Now they're
overridable via .grafosa.yaml or programmatic PipelineConfig.

2. No structured output for automation — GraFOSa's analysis was human-readable
text. Pipelines need structured dicts (feature vectors, impact metrics,
anomaly lists) that can feed into AutoML, DataFrames, or Conductor tasks.

3. No DAG execution model — Steps had no dependency ordering. The runner
provides topological sorting so that downstream tools (Conductor, Airflow)
can adopt the same step contracts.

Architecture

                    .grafosa.yaml
                         |
                   PipelineConfig
                    (overrides)
                         |
    GraFOSa Graph -------+-------> PipelineRunner (DAG)
    (63K nodes,          |            |
     88K edges)          |            +-- Step 1: Feature Extraction
                         |            |     file, service, symbol vectors
                         |            |
                         |            +-- Step 2: Change Impact
                         |            |     blast radius, affected endpoints
                         |            |
                         |            +-- Step 3: Anomaly Detection
                         |            |     z-score outliers (config thresholds)
                         |            |
                         |            +-- Step 4: Summary Formatter
                         |                  compact output for LLM context
                         |
                         +-------> GraphIntel (L7) — config-driven scoring
                         +-------> GraphStrategy (L8) — config-driven risk

Components

PipelineConfig (pipeline/config.py)

Typed, dot-path accessible config with deep-merge over defaults:

from grafosa.pipeline import PipelineConfig

# All built-in defaults
cfg = PipelineConfig.defaults()

# Override from dict (for testing or Conductor)
cfg = PipelineConfig.from_dict({
    "scoring": {"rag_thresholds": {"green": 85}},
    "anomaly": {"z_score_threshold": 3.0},
})

# Override from .grafosa.yaml (for repo-specific tuning)
cfg = PipelineConfig.load(".grafosa.yaml")

# Dot-path access
cfg.get("scoring.section_weights.architecture")  # 0.25
cfg.get("strategy.risk_levels.critical.services") # 4

All configurable values (with current defaults):

PipelineRunner (pipeline/runner.py)

DAG-based step runner with Kahn's topological sort:

from grafosa.pipeline import PipelineRunner, Step, PipelineConfig

runner = PipelineRunner(config=PipelineConfig.defaults())
runner.add_step(Step(name="extract", fn=my_extract_fn))
runner.add_step(Step(name="transform", fn=my_transform_fn, depends_on=["extract"]))

result = runner.run({"code_stack": stack}, pipeline_name="my-pipeline")
# result.status: "completed" | "failed"
# result.steps[0].data: dict output from each step
# result.total_duration_ms: timing

Step contract (Conductor-compatible):

def my_step(input_data: dict, config: PipelineConfig) -> dict:
    # input_data contains initial_input + upstream step outputs keyed by name
    # return dict is passed to downstream steps
    return {"my_output": ...}

Pre-Built Pipelines

from grafosa.pipeline import build_analysis_pipeline, build_change_pipeline

# Analysis: extract → anomaly → summary
runner = build_analysis_pipeline()
result = runner.run({"code_stack": stack})

# Change impact: extract → impact + anomaly → summary
runner = build_change_pipeline()
result = runner.run({"code_stack": stack, "changed_files": ["symposium/conductor.py"]})

Pipeline Steps

How This Fits in Conductor / Symphony

The Problem

Symphony (maestro-platform) is an AI orchestration platform with 12+ services,
6K+ files, and 88K graph edges. When Conductor orchestrates code changes across
this platform, it needs answers to:

1. "What breaks if I change symposium/conductor.py?" — change impact
2. "Is this codebase healthy enough to deploy?" — health scoring
3. "Which files are anomalous and need attention?" — anomaly detection
4. "Give me a compact summary for the LLM agent" — formatted output

Previously, these answers came from hardcoded formulas. A learning pipeline
couldn't tune the thresholds. Now it can.

Conductor Integration Pattern

Conductor Workflow
    |
    +-- Task 1: Build Graph
    |     grafosa build --config .grafosa.yaml
    |     → examples/maestro_platform_full.json
    |
    +-- Task 2: Run Analysis Pipeline
    |     input:  { code_stack, pipeline_type: "analysis" }
    |     output: { features, anomalies, summary }
    |     → feed to learning pipeline for threshold tuning
    |
    +-- Task 3: Run Change Pipeline (on PR)
    |     input:  { code_stack, changed_files: [...] }
    |     output: { impact_metrics, affected_services, summary }
    |     → gate: reject if blast_radius > threshold
    |
    +-- Task 4: Feed to Learning Pipeline
    |     input:  features (AutoML-ready dicts)
    |     output: updated config thresholds
    |     → write back to .grafosa.yaml [pipelines] section
    |
    +-- Task 5: Validate & Deploy
          grafosa validate --max-severity high <services>
          → exit 0 = pass, exit 1 = fail

Concrete Conductor Task Definitions

Task: grafosa_analysis

{
  "name": "grafosa_analysis",
  "taskReferenceName": "analyze",
  "type": "SIMPLE",
  "inputParameters": {
    "graph_path": "${build.output.graph_path}",
    "pipeline_type": "analysis"
  }
}

Conductor worker calls:

from grafosa.pipeline import build_analysis_pipeline, PipelineConfig

cfg = PipelineConfig.load(".grafosa.yaml")
runner = build_analysis_pipeline(cfg)
result = runner.run({"code_stack": load_stack(graph_path)})
return result.steps[-1].data  # summary

Task: grafosa_change_gate

{
  "name": "grafosa_change_gate",
  "taskReferenceName": "gate",
  "type": "SIMPLE",
  "inputParameters": {
    "graph_path": "${build.output.graph_path}",
    "changed_files": "${pr.changed_files}"
  }
}

Conductor worker calls:

from grafosa.pipeline import build_change_pipeline, PipelineConfig

cfg = PipelineConfig.load(".grafosa.yaml")
runner = build_change_pipeline(cfg)
result = runner.run({
    "code_stack": load_stack(graph_path),
    "changed_files": changed_files,
})
impact = result.steps[1].data  # impact step
if impact["impact_metrics"]["total_blast_radius"] > cfg.get("strategy.risk_levels.critical.radius"):
    return {"status": "FAILED", "reason": "blast radius too high"}
return {"status": "COMPLETED", "summary": result.steps[-1].data}

Task: grafosa_features (for learning pipeline)

{
  "name": "grafosa_features",
  "taskReferenceName": "features",
  "type": "SIMPLE",
  "inputParameters": {
    "graph_path": "${build.output.graph_path}",
    "entity_type": "file"
  }
}

Returns dict-of-dicts directly convertible to DataFrame:

features = result["features"]["file"]
# {"symposium/conductor.py": {"lines_of_code": 2847, "pagerank": 0.042, ...}, ...}
import pandas as pd
df = pd.DataFrame.from_dict(features, orient="index")
# → AutoML-ready DataFrame with 23 features per file

The Learning Pipeline Loop

This is the key architectural insight: *the pipeline framework creates a
feedback loop between GraFOSa's structural analysis and the learning pipeline*.

                     .grafosa.yaml [pipelines]
                            ^
                            | (4) write updated thresholds
                            |
    GraFOSa Graph ──(1)──> Pipeline Steps ──(2)──> Feature Vectors
                                                         |
                                                    (3) AutoML
                                                         |
                                                   Learned Thresholds
                                                   (z_score, weights,
                                                    risk levels)

1. GraFOSa builds the graph from source code
2. Pipeline steps extract features (23 per file, 4 per service, 3 per symbol)
3. Learning pipeline trains on features → produces tuned thresholds
4. Thresholds written back to .grafosa.yaml → next analysis uses learned values

What the learning pipeline can tune:

• anomaly.z_score_threshold — what counts as an outlier (currently fixed at 2.0)
• scoring.section_weights.* — how much architecture vs code quality matters
• strategy.risk_levels.* — when to flag critical vs medium risk
• scoring.grade_thresholds.* — what score earns an A vs D

What stays fixed (graph structure):

• Node/edge counts, PageRank, import chains, cycle detection
• These are structural facts, not tunable parameters

MCP Tools (for Claude Code agents)

Two new tools available when the MCP server is running:

CLI Commands

# Run analysis pipeline (no changed files)
PYTHONPATH=src python3 -m grafosa --graph examples/maestro_platform_full.json pipeline --type analysis

# Run change pipeline for specific files
PYTHONPATH=src python3 -m grafosa --graph examples/maestro_platform_full.json pipeline --type change \
  --changed symposium/src/symposium/conductor.py

# Extract file features (AutoML-ready)
PYTHONPATH=src python3 -m grafosa --graph examples/maestro_platform_full.json features --entity-type file

# Extract service features
PYTHONPATH=src python3 -m grafosa --graph examples/maestro_platform_full.json features --entity-type service

Feature Vector Schema

Per-File (23 features)

Per-Service (4 features)

Per-Symbol (3 features)

Backward Compatibility

All changes are backward-compatible:

• GraphIntel(config=None) — behaves identically to pre-change code
• GraphStrategy(config=None) — behaves identically to pre-change code
• PipelineConfig.defaults() produces the exact same values as the previous

hardcoded constants

• All 764 existing tests pass unchanged
• 79 new tests added (843 total)

What's Next

The pipeline framework is the foundation. The next steps are:

1. Conductor workers — Wrap build_analysis_pipeline() and
build_change_pipeline() as Conductor SIMPLE task workers

2. Learning pipeline integration — Feed extract_features() output into
the learning pipeline as training data; write learned thresholds back to
.grafosa.yaml

3. Custom pipeline steps — Add domain-specific steps (e.g., security
scoring, compliance checking) by implementing the Step contract

4. Remote execution — The runner's Step contract (fn(dict, config) → dict)
is designed to map directly to Conductor's inputParameters → outputParameters.
When Conductor support is added, steps can run as distributed tasks with
no contract changes.