Runtime Behavior Correlation Implementation Plan

For Claude: REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

Goal: Build a production-grade runtime behavior correlation system that ties tri-signal telemetry (OTel traces/logs/metrics) to the ADR/PRD/SDS/SEA truth chain + manifests, detects behavioral drift, and surfaces results in Workbench UI and CI with high performance and privacy compliance.

Architecture: A telemetry ingest + normalization layer emits BehaviorEvidence, a correlation engine links evidence to provenance nodes, a drift classifier scores divergence, and results persist to the KG (authoritative) with a Postgres summary index. UI + CI consume summaries and drill into KG evidence.

Tech Stack: Python (workbench-bff), OTel/OTLP, OpenObserve, Knowledge Graph (Oxigraph), Postgres, React (Workbench), Nx, Vitest, Pytest.

Task 1: Define Spec Changes (ADR/PRD/SDS/SEA)

Files:

Step 1: Write the failing spec assertions

Add a new SDS section defining:

Step 2: Validate specs

Run:

1

just sds-validate docs/specs/shared/sds/0xx-runtime-behavior-correlation.md

Expected: PASS

Step 3: Commit

1
2
3
4
5
6

git add docs/specs/shared/adr/029-observability-stack-architecture.md \
  docs/specs/shared/sds/030-semantic-observability.md \
  docs/specs/shared/sds/0xx-runtime-behavior-correlation.md \
  docs/specs/shared/prd/025-workbench-ui.md

git commit -m "spec: add runtime behavior correlation SDS"

Task 2: Update Generators (if new spec outputs required)

Files:

Step 1: Add new manifest/IR mapping rules

Update AST/IR/Manifest mapping to include correlation schema if needed.

Step 2: Regenerate outputs

Run:

1

just pipeline shared

Expected: Deterministic output

Step 3: Commit

1
2
3

git add generators/ tools/

git commit -m "feat: extend generators for behavior correlation"

Task 3: Implement Behavior Evidence Normalizer

Files:

Step 1: Write the failing test

1
2
3
4

def test_normalize_trace_with_semantic_tags():
    normalizer = BehaviorNormalizer()
    evidence = normalizer.normalize_trace({"resource": {"attributes": {"sea.flow": "CreateOrder"}}})
    assert evidence.flow == "CreateOrder"

Step 2: Run test to verify it fails

Run:

1

cd services/workbench-bff && python -m pytest tests/test_behavior_normalizer.py -v

Expected: FAIL (BehaviorNormalizer not defined)

Step 3: Write minimal implementation

1
2
3

class BehaviorNormalizer:
    def normalize_trace(self, trace: dict) -> BehaviorEvidence:
        return BehaviorEvidence(flow=trace["resource"]["attributes"].get("sea.flow"))

Step 4: Run test to verify it passes

1

cd services/workbench-bff && python -m pytest tests/test_behavior_normalizer.py -v

Expected: PASS

Step 5: Commit

1
2
3
4

git add services/workbench-bff/src/adapters/behavior_normalizer.py \
  services/workbench-bff/tests/test_behavior_normalizer.py

git commit -m "feat: add behavior evidence normalizer"

Task 4: Implement Correlation Engine

Files:

Step 1: Write the failing test

1
2
3
4

def test_correlate_exact_flow_match():
    correlator = BehaviorCorrelator()
    result = correlator.correlate(flow="CreateOrder", context="semantic-core")
    assert result.confidence == 1.0

Step 2: Run test to verify it fails

Run:

1

cd services/workbench-bff && python -m pytest tests/test_behavior_correlator.py -v

Expected: FAIL

Step 3: Write minimal implementation

1
2
3

class BehaviorCorrelator:
    def correlate(self, flow: str, context: str) -> CorrelationResult:
        return CorrelationResult(confidence=1.0, spec_node_id=f"sea:{context}:{flow}")

Step 4: Run test to verify it passes

1

cd services/workbench-bff && python -m pytest tests/test_behavior_correlator.py -v

Expected: PASS

Step 5: Commit

1
2
3
4

git add services/workbench-bff/src/adapters/behavior_correlator.py \
  services/workbench-bff/tests/test_behavior_correlator.py

git commit -m "feat: add behavior correlation engine"

Task 5: Implement Drift Classifier

Files:

Step 1: Write the failing test

1
2
3
4

def test_classify_missing_db_span_is_medium():
    classifier = BehaviorDriftClassifier()
    severity = classifier.classify(expected=["db"], observed=["http"])
    assert severity == DriftSeverity.MEDIUM

Step 2: Run test to verify it fails

1

cd services/workbench-bff && python -m pytest tests/test_behavior_drift_classifier.py -v

Expected: FAIL

Step 3: Write minimal implementation

1
2
3

class BehaviorDriftClassifier:
    def classify(self, expected: list[str], observed: list[str]) -> DriftSeverity:
        return DriftSeverity.MEDIUM if "db" in expected and "db" not in observed else DriftSeverity.NONE

Step 4: Run test to verify it passes

1

cd services/workbench-bff && python -m pytest tests/test_behavior_drift_classifier.py -v

Expected: PASS

Step 5: Commit

1
2
3
4

git add services/workbench-bff/src/adapters/behavior_drift_classifier.py \
  services/workbench-bff/tests/test_behavior_drift_classifier.py

git commit -m "feat: add behavioral drift classifier"

Task 6: Persist Correlations (KG + Postgres)

Files:

Step 1: Write failing tests

Add tests to assert KG writer creates observed_behavior edge and Postgres summary persists.

Step 2: Implement KG writer + Postgres indexer

Step 3: Run tests

1

cd services/workbench-bff && python -m pytest tests/test_behavior_storage.py -v

Step 4: Commit

1
2
3
4
5

git add services/workbench-bff/src/adapters/behavior_kg_writer.py \
  services/workbench-bff/src/adapters/behavior_indexer.py \
  services/workbench-bff/src/models.py

git commit -m "feat: persist behavior correlation results"

Task 7: Add API Routes

Files:

Step 1: Write failing tests

1

GET /behavior/summary -> 200 with list

Step 2: Implement routes

Step 3: Run tests

1

cd services/workbench-bff && python -m pytest tests/test_behavior_routes.py -v

Step 4: Commit

1
2
3
4

git add services/workbench-bff/src/api/behavior_routes.py \
  services/workbench-bff/main.py

git commit -m "feat: add behavior correlation API routes"

Task 8: Workbench UI

Files:

Step 1: Write failing tests

1

RuntimeCorrelationDashboard renders summary cards

Step 2: Implement UI

Step 3: Run tests

1

pnpm nx test workbench --testFile=RuntimeCorrelationDashboard

Step 4: Commit

1
2
3
4
5
6
7
8

git add apps/workbench/src/pages/RuntimeCorrelationDashboard.tsx \
  apps/workbench/src/components/BehaviorDriftCard.tsx \
  apps/workbench/src/types/behavior.ts \
  apps/workbench/src/lib/behavior-api.ts \
  apps/workbench/src/pages/ProvenanceExplorer.tsx \
  apps/workbench/src/App.tsx

git commit -m "feat: add runtime correlation UI"

Task 9: CI Drift Gate

Files:

Step 1: Add script

Implement --warn and --fail modes with thresholds.

Step 2: Wire CI

Add a step:

1
2

- name: Behavior Drift Gate
  run: ./scripts/ci/behavior_drift_gate.sh --warn

Step 3: Commit

1
2
3

git add scripts/ci/behavior_drift_gate.sh .github/workflows/ci.yml

git commit -m "chore: add behavior drift gate"

Task 10: Validation + Determinism

Step 1: Run spec guard

1

just spec-guard

Step 2: Run tests

1
2

just test-python
just test-ts

Step 3: Determinism check

1

just ci-determinism

Step 4: Commit (if needed)

1

git status --porcelain

Expected: clean

Execution Options

Plan complete and saved to docs/plans/2026-01-23-runtime-behavior-correlation.md.

Two execution options:

  1. Subagent-Driven (this session) - I dispatch fresh subagent per task, review between tasks, fast iteration
  2. Parallel Session (separate) - Open new session with executing-plans, batch execution with checkpoints

Which approach?