Orchestrata Ultimate › Plans › plan_automated_porting_harness › subplan-step0-ast-serializer-neo4j-sync-2026-04-11

Plan: Phase 1 — AST Serializer Extension + Neo4j Full Sync

Source: plans/plan_automated_porting_harness/subplan-step0-ast-serializer-neo4j-sync-2026-04-11.md at 8a2cb42ad · last changed 2026-07-27 · first committed 2026-04-11 · 1288 lines.

Prose pending.

Plan text

Plan: Phase 1 — AST Serializer Extension + Neo4j Full Sync

Context

We have scripts/serialize_codebase_ast.py (1,451 LOC) that captures modules, classes, methods, parameters, calls, imports, inheritance, and dependency graphs — but stops at method signatures. We need to extend it to capture statement-level AST (if/for/try/except/assign/return/raise/with) and sync the ENTIRE enriched AST to Neo4j.

This is Phase 1. Phase 2 (the harness context_builder that queries Neo4j) is deferred until Phase 1 is verified.

Neo4j: neo4j://10.222.9.188:7687 (user: neo4j, pass: knowledge123)

Scope: What Gets Built and Verified

Step 1: Extend /home/mohamed/workspace/orchestrata-ultimate/scripts/serialize_codebase_ast.py

Current state: 1,451 LOC. Captures declarations (modules, classes, methods, params, calls, imports, vars, inheritance, deps). Does NOT capture method/function body statements.

What to add (~200 LOC):

New dataclass:

@dataclass
class StatementInfo:
    type: str              # ALL 28 ast.stmt types handled (see list below)
    line: int
    col_offset: int = 0    # Column offset — needed for uniqueness (same-line semicolons)
    end_line: int | None = None
    depth: int = 0         # nesting level (0 = top-level body)
    # Type-specific fields
    test: str | None = None           # If/While/Match guard condition (ast.unparse)
    subject: str | None = None        # Match subject expression (ast.unparse)
    target: str | None = None         # For target, Assign targets, Delete target (ast.unparse)
    exception: str | None = None      # ExceptHandler/Raise exception type
    func: str | None = None           # Call function name (ast.unparse)
    value: str | None = None          # Return/Assign/TypeAlias value (ast.unparse, truncated 200 chars)
    items: str | None = None          # With/AsyncWith context manager expression
    op: str | None = None             # AugAssign operator (+=, -=, etc.)
    args_count: int = 0               # Call args count
    has_value: bool = False            # Return has a value
    name: str | None = None           # ExceptHandler `as name`, Global/Nonlocal names (joined)
    names: list[str] = field(default_factory=list)  # Global/Nonlocal name list
    children: list['StatementInfo'] = field(default_factory=list)  # body statements
    handlers: list['StatementInfo'] = field(default_factory=list)  # Try/TryStar except handlers
    orelse: list['StatementInfo'] = field(default_factory=list)    # If/Try/For else
    finalbody: list['StatementInfo'] = field(default_factory=list) # Try finally
    cases: list['StatementInfo'] = field(default_factory=list)     # Match cases (each case body in children)

# Statement types handled (Python 3.14 complete):
# From ast.stmt: If, For, AsyncFor, While, Try, TryStar, With, AsyncWith,
#   Match, Assign, AugAssign, AnnAssign, Return, Raise, Assert, Pass,
#   Delete, Global, Nonlocal, Break, Continue, TypeAlias, Expr
# From ast.excepthandler: ExceptHandler (not ast.stmt — broadened signature)
# NOT handled here (handled by other extractors): FunctionDef, AsyncFunctionDef,
#   ClassDef, Import, ImportFrom
# Yield/YieldFrom are ast.expr — detected inside ast.Expr wrapper, not top-level

New field on FunctionInfo:

body_statements: list[StatementInfo] = field(default_factory=list)

New field on ModuleInfo (the module IS the AST root per file — its body has top-level statements too):

body_statements: list[StatementInfo] = field(default_factory=list)  # module-level if/try/assign/call

This captures what the current serializer misses at module level:

• if TYPE_CHECKING: blocks (conditional imports — critical for forward references)
• try: import redis except ImportError: redis = None (optional dependency detection)
• Module-level function calls like configure_logging() (initialization side effects)
• Module-level assignments beyond all (constants, singletons, config)

The full path from Root to leaf is:

Root ("Projects")
  -[:HAS_PROJECT]-> Project ("orchestrata")
    -[:HAS_PACKAGE]-> Package ("orchestrata-kernel")
      -[:HAS_MODULE]-> Module ("orchestrata.kernel.event_bus")
        -[:HAS_BODY]-> Statement (module-level: if TYPE_CHECKING, try/except ImportError, etc.)
        -[:DEFINES_CLASS]-> Class ("EventBus")
          -[:HAS_METHOD]-> Method ("publish")
            -[:HAS_BODY]-> Statement (method-level: if/try/return/assign)
              -[:HAS_BODY]-> Statement (nested: deeper if/try/etc.)
            -[:HAS_PARAMETER]-> Parameter ("event_type")
          -[:HAS_VAR]-> ClassVar ("_handlers")
        -[:DEFINES_FUNCTION]-> Function ("create_event_bus")
          -[:HAS_BODY]-> Statement (function-level)
          -[:HAS_PARAMETER]-> Parameter ("redis_url")
        -[:IMPORTS]-> Module (another module)
        -[:HAS_MODULE_VAR]-> ModuleVar ("REDIS_HOST")

Every node is reachable from the Root via a deterministic path. No orphans.

New methods on CodebaseAnalyzer:

def _extract_statements(self, body: list[ast.stmt], depth: int = 0) -> list[StatementInfo]:
    """Recursively extract all statements from a body list."""

def _statement_to_info(self, node: ast.stmt | ast.excepthandler, depth: int) -> StatementInfo | None:
    """Convert AST node to StatementInfo. Broadened signature: accepts ast.excepthandler too.
    Handles ALL Python 3.14 stmt types + ExceptHandler. See StatementInfo comment for full list.
    Yield/YieldFrom detected inside ast.Expr wrapper (they are ast.expr, not ast.stmt).
    Recursively processes body/handlers/orelse/finalbody/cases."""

def flatten_statements(stmts: list[StatementInfo], owner_qname: str, parent_id: str | None = None) -> list[dict]:
    """Recursive DFS flattener for Neo4j UNWIND batching.
    Walks the StatementInfo tree, assigns statement_id using line:col_offset:depth,
    sets parent_statement_id, emits a flat list of dicts for UNWIND.
    ~30 LOC. Required because Cypher UNWIND cannot handle nested trees."""

Wire into _analyze_function(): after extracting params/calls/return_type, call self._extract_statements(node.body) and store in func.body_statements.

Wire into _analyze_file(): after existing ClassDef/FunctionDef/Import/AnnAssign/Assign handling, capture module-level statements:

module.body_statements = self._extract_statements([
    n for n in ast.iter_child_nodes(tree)
    if not isinstance(n, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef,
                           ast.Import, ast.ImportFrom))
    # Note: AnnAssign and Assign at module level are captured BOTH as module_vars
    # AND as body_statements. The var extractor gets the name/type; the statement
    # extractor gets the structure (for detecting if TYPE_CHECKING: blocks etc.)
])

Add a 14th verification query to test module-level statements:

-- 14. Module-level statements (if TYPE_CHECKING:, try/except ImportError, etc.)
MATCH (m:Module {project: 'orchestrata'})-[:HAS_BODY]->(s:Statement)
RETURN m.qualified_name, s.type, s.line LIMIT 10

Verification:

cd /home/mohamed/workspace/orchestrata-ultimate
python scripts/serialize_codebase_ast.py serialize \
    --source src/orchestrata_ultimate \
    --name orchestrata-ultimate \
    --output /tmp/test-ast-statements.json

# Verify statements are captured:
python -c "
import json
data = json.load(open('/tmp/test-ast-statements.json'))
total_stmts = 0
for mod in data['modules']:
    for cls in mod.get('classes', []):
        for method in cls.get('methods', []):
            stmts = method.get('body_statements', [])
            total_stmts += len(stmts)
    for func in mod.get('functions', []):
        stmts = func.get('body_statements', [])
        total_stmts += len(stmts)
print(f'Total top-level statements captured: {total_stmts}')
assert total_stmts > 0, 'No statements captured!'
print('PASS')
"

Also re-serialize orchestrata:

python scripts/serialize_codebase_ast.py serialize \
    --source /home/mohamed/workspace/orchestrata/packages \
    --name orchestrata \
    --output /tmp/test-ast-orch-statements.json \
    --monorepo

Step 2: Create /home/mohamed/workspace/orchestrata-ultimate/scripts/sync_ast_to_neo4j.py

New file: ~600 LOC. Reads enriched AST JSON, creates the full graph in Neo4j.

Graph hierarchy:

(:Root {name: "Projects"})
  -[:HAS_PROJECT]-> (:Project {name: "orchestrata"})
    -[:HAS_PACKAGE]-> (:Package {name: "orchestrata-kernel", tier: 2, loc: 15272})
      -[:HAS_MODULE]-> (:Module {qualified_name: "orchestrata.kernel.event_bus"})
        -[:DEFINES_CLASS]-> (:Class {name: "EventBus"})
          -[:HAS_METHOD]-> (:Method {name: "publish", line: 45})
            -[:HAS_PARAMETER]-> (:Parameter {name: "event_type", type: "str"})
            -[:HAS_BODY]-> (:Statement {type: "Try", line: 47})
              -[:HAS_BODY]-> (:Statement {type: "Assign", line: 48})
              -[:HAS_HANDLER]-> (:Statement {type: "ExceptHandler", exception: "ConnectionError"})
            -[:HAS_BODY]-> (:Statement {type: "Return", line: 53})
          -[:HAS_VAR]-> (:ClassVar {name: "_handlers", type: "dict[str, Callable]"})
        -[:DEFINES_FUNCTION]-> (:Function {name: "create_event_bus"})
        -[:IMPORTS]-> (:Module ...)
  -[:HAS_PROJECT]-> (:Project {name: "orchestrata-ultimate"})
    ...

Cross-project:
(:Class {name: "BaseVolk"})-[:MAPS_TO]->(:Class {name: "BaseAgent"})

Node labels and properties:

Relationships:

Schema DDL:

CREATE CONSTRAINT root_unique IF NOT EXISTS FOR (r:Root) REQUIRE r.name IS UNIQUE;
CREATE CONSTRAINT project_unique IF NOT EXISTS FOR (p:Project) REQUIRE p.name IS UNIQUE;
CREATE CONSTRAINT package_unique IF NOT EXISTS FOR (pkg:Package) REQUIRE (pkg.project, pkg.name) IS UNIQUE;
CREATE CONSTRAINT module_unique IF NOT EXISTS FOR (m:Module) REQUIRE (m.project, m.qualified_name) IS UNIQUE;
CREATE CONSTRAINT class_unique IF NOT EXISTS FOR (c:Class) REQUIRE (c.project, c.qualified_name) IS UNIQUE;
CREATE CONSTRAINT method_unique IF NOT EXISTS FOR (m:Method) REQUIRE (m.project, m.qualified_name) IS UNIQUE;
CREATE CONSTRAINT function_unique IF NOT EXISTS FOR (f:Function) REQUIRE (f.project, f.qualified_name) IS UNIQUE;
CREATE CONSTRAINT param_unique IF NOT EXISTS FOR (p:Parameter) REQUIRE (p.project, p.owner_qualified_name, p.name) IS UNIQUE;
CREATE CONSTRAINT classvar_unique IF NOT EXISTS FOR (v:ClassVar) REQUIRE (v.project, v.owner_qualified_name, v.name) IS UNIQUE;

CREATE INDEX method_name IF NOT EXISTS FOR (m:Method) ON (m.name);
CREATE INDEX class_name IF NOT EXISTS FOR (c:Class) ON (c.name);
CREATE INDEX class_pydantic IF NOT EXISTS FOR (c:Class) ON (c.is_pydantic);
CREATE INDEX module_qname IF NOT EXISTS FOR (m:Module) ON (m.qualified_name);
CREATE INDEX param_type IF NOT EXISTS FOR (p:Parameter) ON (p.type_annotation);
CREATE INDEX classvar_type IF NOT EXISTS FOR (v:ClassVar) ON (v.type_annotation);
-- NOTE: calls_raw and decorators are arrays — TEXT indexes don't support arrays in Neo4j 5.x
-- calls_raw is covered by Call nodes + call_func index. decorators queried via ANY() filter.
CREATE TEXT INDEX class_docstring IF NOT EXISTS FOR (c:Class) ON (c.docstring);
CREATE TEXT INDEX method_docstring IF NOT EXISTS FOR (m:Method) ON (m.docstring);
CREATE INDEX statement_type IF NOT EXISTS FOR (s:Statement) ON (s.type);
CREATE INDEX statement_exception IF NOT EXISTS FOR (s:Statement) ON (s.exception);
CREATE INDEX call_func IF NOT EXISTS FOR (c:Call) ON (c.func);

CLI subcommands:

# Create schema only
python scripts/sync_ast_to_neo4j.py schema \
    --uri neo4j://10.222.9.188:7687 --user neo4j --password knowledge123

# Full sync for a project (clears existing project data first)
python scripts/sync_ast_to_neo4j.py full \
    --ast plans/ast-ultimate.json \
    --project orchestrata-ultimate \
    --uri neo4j://10.222.9.188:7687 --user neo4j --password knowledge123

# Full sync for orchestrata (with package tier info from class-mapping.json)
python scripts/sync_ast_to_neo4j.py full \
    --ast plans/ast-orchestrata.json \
    --project orchestrata \
    --packages-from plans/class-mapping.json \
    --uri neo4j://10.222.9.188:7687 --user neo4j --password knowledge123

# Create cross-project MAPS_TO relationships
python scripts/sync_ast_to_neo4j.py map \
    --mapping plans/class-mapping.json \
    --uri neo4j://10.222.9.188:7687 --user neo4j --password knowledge123

# Incremental re-sync for specific modules
python scripts/sync_ast_to_neo4j.py incremental \
    --ast plans/ast-ultimate.json \
    --project orchestrata-ultimate \
    --modules kernel.event_bus,agents.base_agent \
    --uri neo4j://10.222.9.188:7687 --user neo4j --password knowledge123

# Print node/relationship counts
python scripts/sync_ast_to_neo4j.py stats \
    --uri neo4j://10.222.9.188:7687 --user neo4j --password knowledge123

Sync order: NODES FIRST, EDGES SECOND.

Each node carries its parent's identifier as a property (e.g., Method has class_qualified_name, Class has module_qualified_name). All nodes are created in Phase A with UNWIND batches. All edges are created in Phase B by matching on those parent IDs against indexed constraint properties. This is the fastest pattern — no nested queries during node creation.

Phase A — All nodes (independent, could be parallelized):

1. Ensure schema (constraints + indexes)
2. Clear existing data: MATCH (n) WHERE n.project = $project DETACH DELETE n
3. MERGE Root node: MERGE (r:Root {name: 'Projects'})
4. MERGE Project node (carries: name, root_path, totals)
5. UNWIND batch: Package nodes (carries: project name for parent lookup)
6. UNWIND batch: Module nodes (carries: package_name for parent lookup)
7. UNWIND batch: Class nodes (carries: module_qualified_name for parent lookup)
8. UNWIND batch: Method nodes (carries: class_qualified_name for parent lookup)
9. UNWIND batch: Function nodes (carries: module_qualified_name for parent lookup)
10. UNWIND batch: Parameter nodes (carries: owner_qualified_name for parent lookup)
11. UNWIND batch: ClassVar nodes (carries: owner_qualified_name for parent lookup)
12. UNWIND batch: ModuleVar nodes (carries: owner_qualified_name for parent lookup)
13. UNWIND batch: Statement nodes (carries: owner_qualified_name + parent_statement_id for nesting)
14. UNWIND batch: Call nodes (carries: owner_qualified_name for parent lookup)

Phase B — All edges (match on indexed properties, fast): 15. HAS_PROJECT: MATCH (r:Root), (p:Project {name: $name}) CREATE (r)-[:HAS_PROJECT]->(p) 16. HAS_PACKAGE: MATCH (p:Project {name: $proj}), (pkg:Package {project: $proj, name: $pkg_name}) CREATE (p)-[:HAS_PACKAGE]->(pkg) 17. HAS_MODULE: UNWIND $batch AS row MATCH (pkg:Package {project: row.project, name: row.pkg}), (m:Module {project: row.project, qualified_name: row.qname}) CREATE (pkg)-[:HAS_MODULE]->(m) 18. DEFINES_CLASS: UNWIND $batch MATCH (m:Module {project: row.project, qualified_name: row.module_qname}), (c:Class {project: row.project, qualified_name: row.class_qname}) CREATE (m)-[:DEFINES_CLASS]->(c) 19. DEFINES_FUNCTION: same pattern, Module -> Function 20. HAS_METHOD: MATCH (c:Class ...), (m:Method {class_qualified_name: c.qualified_name}) CREATE (c)-[:HAS_METHOD]->(m) 21. HAS_PARAMETER: MATCH (owner) WHERE owner.qualified_name = param.owner_qualified_name CREATE (owner)-[:HAS_PARAMETER {position: row.pos}]->(param) 22. HAS_VAR: Class -> ClassVar via owner_qualified_name 23. HAS_MODULE_VAR: Module -> ModuleVar 24. HAS_BODY: Method/Function/Module -> Statement (top-level body, position ordered) 25. HAS_BODY: Statement -> Statement (nested body — parent_statement_id match) 26. HAS_HANDLER: Statement(Try) -> Statement(ExceptHandler) 27. HAS_ORELSE: Statement(If/Try/For) -> Statement 28. HAS_FINALLY: Statement(Try) -> Statement 29. MAKES_CALL: Statement/Method/Function -> Call 30. INHERITS: UNWIND $pairs MATCH (child:Class ...), (parent:Class ...) CREATE (child)-[:INHERITS]->(parent) 31. IMPORTS: UNWIND $pairs MATCH (a:Module ...), (b:Module ...) CREATE (a)-[:IMPORTS {names: row.names}]->(b) 32. RESOLVES_TO: post-pass — match Call.func to Method/Function qualified_names

Parent ID properties on each node (for fast edge creation):

Statement nesting: Each Statement gets a statement_id (e.g., "{owner_qname}:stmt:{line}:{col_offset}:{depth}"). col_offset prevents collisions for same-line semicolons (e.g., x = 1; y = 2 — both line 3 but col_offset 4 vs 11). Nested statements carry parent_statement_id pointing to their containing statement. Top-level method/module body statements have parent_statement_id = null and edge comes from Method/Function/Module directly.

All batch operations use UNWIND for performance. Phase B edges are pure MATCH+CREATE on indexed properties — no scanning.

Incremental Update Strategy (when code changes)

When a harness step modifies agents/base_agent.py and merges, the graph is stale. The incremental sync must handle:

Problem: cross-module edges break on delete-recreate.

• INHERITS: AnalystAgent → BaseAgent. If BaseAgent's node is deleted and recreated, Neo4j assigns a new internal ID. The INHERITS edge from AnalystAgent is gone (DETACH DELETE).
• MAPS_TO: BaseVolk → BaseAgent. Same — edge lost.
• RESOLVES_TO: Call nodes in other modules that resolved to BaseAgent.process → edge lost.
• IMPORTS: Other modules that import from this module → edge lost on the Module node.

Solution: delete module nodes, recreate nodes, then repair ALL edges touching that module — both incoming and outgoing.

incremental subcommand steps:

1. Read fresh AST JSON for the specified modules
2. Delete ALL nodes belonging to the changed modules in one sweep:
   ```cypher
   UNWIND $module_prefixes AS prefix
   MATCH (n)
   WHERE n.project = $project AND (
       n.module_path STARTS WITH prefix
       OR n.owner_qualified_name STARTS WITH prefix
       OR n.qualified_name STARTS WITH prefix
       OR n.qualified_name = prefix
   )
   DETACH DELETE n

This catches Module, Class, Method, Function AND all children (Parameter, ClassVar, ModuleVar, Statement, Call) regardless of which property stores the module reference. DETACH DELETE removes all edges in both directions. 4. Phase A: Re-create nodes for the changed modules only (same UNWIND batch pattern) 5. Phase B-internal: Re-create edges WITHIN the changed modules (parent-child hierarchy) 6. Phase B-cross: REPAIR cross-module edges: a. INHERITS: re-match all Class nodes in the changed modules against all Class nodes in the project MATCH (c:Class {project: $proj}) WHERE c.module_path IN $changed UNWIND c.bases AS base_name MATCH (parent:Class {project: $proj, name: base_name}) MERGE (c)-[:INHERITS]->(parent) b. INHERITS-incoming: re-match all Class nodes in OTHER modules that inherit from classes in the changed modules MATCH (c:Class {project: $proj})-[:INHERITS]->(missing) WHERE missing IS NULL // broken edge — target was deleted ... (can't match deleted nodes)

BETTER APPROACH: After step 4, just re-run ALL INHERITS for the entire project. It's only ~2000 class pairs — takes <1 second with UNWIND.

c. IMPORTS: re-run ALL IMPORTS for the entire project (same reason — fast enough) d. RESOLVES_TO: re-run CALLS resolution for methods in the changed modules PLUS re-run for methods in OTHER modules whose Call nodes point at changed module targets e. MAPS_TO: re-run from class-mapping.json for classes in the changed modules 7. Verify: run stats, compare node counts against previous

Versioning properties on every node:

• synced_at: str — ISO timestamp of last sync
• ast_hash: str — hash of the node's source data (detect drift without re-syncing)

The ast_hash is computed from the serialized JSON of the source dataclass. If the hash matches, the node is current. If it differs, the node needs update. This allows the incremental command to optionally SKIP unchanged modules:

## Only re-sync modules whose AST hash changed since last sync
python scripts/sync_ast_to_neo4j.py incremental --ast plans/ast-ultimate.json --project orchestrata-ultimate --modules agents.base_agent --skip-unchanged

Why not diff-and-patch instead of delete-recreate? Diffing individual node properties and selectively updating is more complex, harder to verify, and barely faster for the node counts we have (~90K total). Delete-recreate is simple, correct, and completes in seconds. The only cost is re-creating cross-module edges, which we handle by re-running the full INHERITS/IMPORTS pass (~2 seconds for the whole project).

Estimated node counts:

• orchestrata: ~456 modules, ~947 classes, ~3,877 methods, ~10,000+ parameters, ~50,000+ statements, ~20,000+ calls
• ultimate: ~37 modules, ~57 classes, ~147 methods, ~500+ parameters, ~2,000+ statements, ~1,000+ calls
• Total: ~90,000 nodes. Neo4j handles this trivially.

Package detection logic:

• orchestrata: module path orchestrata.kernel.event_bus -> package orchestrata-kernel. Tier/loc/target from class-mapping.json tier_assignments.
• ultimate: module paths in the AST are already relative (e.g., kernel.event_bus, agents.base_agent). Take first segment as package name. Edge cases: paths __init__ and __main__ (root-level files) map to a root package named after the project. Guard: if path in ('__init__', '__main__') or path.startswith('__init__.'): package = project_name.

CALLS resolution logic (step 17): For each Method/Function node's calls_raw array:

1. Strip self., super()., cls. prefixes
2. If bare name after strip (e.g., process from self.process): a. Try {module_path}.{class_name}.{bare_name} as Method in same class b. If not found, walk inheritance chain (from inheritance_graph) checking each parent class c. If still not found, try {module_path}.{bare_name} as module-level Function
3. If call still has dots after strip (e.g., event_bus.publish from instance attribute):
   • Only apply dot-path lookup if prefix before first dot is a known module or class name
   • Otherwise mark as unresolved (do NOT fabricate a wrong qualified_name)
4. If call matches a known class name: try {class_name}.__init__ as constructor call
5. If resolved -> create CALLS relationship with resolved=true + RESOLVES_TO from Call node
6. If unresolved -> calls_raw stays on the Method node, searchable via string queries

Verification Checklist

After both steps are complete, run these queries to verify the graph:

## 1. Root hierarchy
python scripts/sync_ast_to_neo4j.py stats

## Expected output (approximate):
## Root: 1
## Project: 2
## Package: 43 (26 orchestrata + 17 ultimate)
## Module: 493 (456 + 37)
## Class: 1004 (947 + 57)
## Method: 4024 (3877 + 147)
## Function: ...
## Parameter: ~10500
## ClassVar: ...
## ModuleVar: ...
## Statement: ~52000
## Call: ~21000

-- 2. Verify root -> project hierarchy
MATCH (r:Root)-[:HAS_PROJECT]->(p:Project) RETURN r.name, p.name, p.total_loc

-- 3. Verify package hierarchy for orchestrata
MATCH (p:Project {name:'orchestrata'})-[:HAS_PACKAGE]->(pkg:Package)
RETURN pkg.name, pkg.tier, pkg.loc ORDER BY pkg.loc DESC LIMIT 10

-- 4. Verify module -> class -> method depth
MATCH (m:Module {project:'orchestrata', qualified_name:'orchestrata.kernel.event_bus'})
      -[:DEFINES_CLASS]->(c:Class)-[:HAS_METHOD]->(meth:Method)
RETURN c.name, collect(meth.name)

-- 5. Verify parameter nodes
MATCH (m:Method {name:'process', project:'orchestrata'})-[:HAS_PARAMETER]->(p:Parameter)
RETURN m.qualified_name, p.name, p.type_annotation LIMIT 20

-- 6. Verify statement-level depth
MATCH (m:Method {project:'orchestrata'})-[:HAS_BODY]->(s:Statement)
RETURN m.qualified_name, s.type, s.line LIMIT 20

-- 7. Verify nested statements (try/except inside a method)
MATCH (m:Method)-[:HAS_BODY*]->(t:Statement {type:'Try'})-[:HAS_HANDLER]->(h:Statement {type:'ExceptHandler'})
WHERE m.project = 'orchestrata'
RETURN m.qualified_name, t.line, h.exception LIMIT 10

-- 8. Verify MAPS_TO cross-project
MATCH (a:Class)-[:MAPS_TO]->(b:Class)
RETURN a.name AS orchestrata_class, a.project, b.name AS ultimate_class, b.project LIMIT 10

-- 9. Verify INHERITS
MATCH (c:Class)-[:INHERITS]->(base:Class)
WHERE c.project = 'orchestrata'
RETURN c.name, base.name LIMIT 10

-- 10. Verify IMPORTS
MATCH (a:Module)-[:IMPORTS]->(b:Module)
WHERE a.project = 'orchestrata' AND a.qualified_name = 'orchestrata.kernel.event_bus'
RETURN b.qualified_name, a.qualified_name

-- 11. Verify Call nodes with resolution
MATCH (c:Call)-[:RESOLVES_TO]->(target:Method)
WHERE c.project = 'orchestrata'
RETURN c.func, target.qualified_name LIMIT 10

-- 12. Statement-level query: find all try/except ImportError
MATCH (m:Method)-[:HAS_BODY*]->(t:Statement {type:'Try'})-[:HAS_HANDLER]->(h:Statement {type:'ExceptHandler', exception:'ImportError'})
WHERE m.project = 'orchestrata'
RETURN m.qualified_name, t.line LIMIT 10

-- 13. Stub detection: methods with only Pass in body
MATCH (m:Method {project:'orchestrata-ultimate'})-[:HAS_BODY]->(s:Statement {type:'Pass'})
WHERE size((m)-[:HAS_BODY]->()) = 1
RETURN m.qualified_name

ALL 13 queries must return results (except #13 which may return 0 if no stubs exist).

Files Summary

Total new code: ~980 LOC across 3 files.

No other files are modified in Phase 1.

Sandboxed Execution Strategy

Design Principle

Prevention over detection. Agents work in minimal sandbox directories containing ONLY files they are allowed to modify, plus read-only context extracted from the AST. They cannot access the rest of the project. The orchestrator shuttles files and test results between the sandbox and the real repo.

The agent either succeeds within its sandbox or fails and returns the reason. It cannot circumvent the design.

Sandbox Layout

Agent 1 (AST Serializer Extension):

/home/mohamed/workspace/orchestrata-ultimate-worktrees/step0-ast-serializer/
├── work/                                    ← WRITABLE (agent can only modify/create files here)
│   ├── serialize_codebase_ast.py            ← copied from repo, agent modifies this
│   └── AUDIT.md                             ← agent MUST create this (see Audit Document below)
├── context/                                 ← READ-ONLY (injected by orchestrator)
│   ├── ast-ultimate.json                    ← full AST of the codebase
│   ├── spec.md                              ← extracted Agent 1 scope from this subplan
│   ├── standards.md                         ← coding/testing/structure standards
│   ├── traceability.md                      ← REQ mapping for Agent 1
│   ├── existing_dataclasses.md              ← extracted from AST: all current dataclass fields/types
│   ├── existing_design.mmd                  ← Mermaid class diagram of CURRENT serialize_codebase_ast.py
│   └── proposed_design.mmd                  ← Mermaid class diagram of TARGET state after changes
└── verify/                                  ← orchestrator writes test results here
    └── (test_output.txt created by orchestrator after each verify cycle)

Agent 2 (Neo4j Sync Engine):

/home/mohamed/workspace/orchestrata-ultimate-worktrees/step0-neo4j-sync/
├── work/                                    ← WRITABLE
│   ├── sync.py                              ← agent creates this (knowledge_graph/sync.py)
│   ├── sync_ast_to_neo4j.py                ← agent creates this (CLI wrapper)
│   └── AUDIT.md                             ← agent MUST create this (see Audit Document below)
├── context/                                 ← READ-ONLY
│   ├── ast-ultimate.json                    ← full AST
│   ├── spec.md                              ← extracted Agent 2 scope from this subplan
│   ├── standards.md                         ← coding/testing/structure standards
│   ├── traceability.md                      ← REQ mapping for Agent 2
│   ├── graph.py                             ← existing knowledge_graph/graph.py (for patterns)
│   ├── __init__.py                          ← existing knowledge_graph/__init__.py
│   ├── traceability_utils.py                ← @traces decorator source
│   ├── env_vars.md                          ← NEO4J_URI, NEO4J_USERNAME, NEO4J_PASSWORD docs
│   ├── existing_design.mmd                  ← Mermaid diagram of CURRENT knowledge_graph/ module
│   └── proposed_design.mmd                  ← Mermaid diagram of TARGET state (Neo4jSyncer + CLI)
└── verify/                                  ← orchestrator writes test/sync results here
    └── (test_output.txt, cypher_results.txt created by orchestrator)

Design Diagrams (generated by orchestrator before dispatch)

Each agent receives two Mermaid class diagrams in context/:

existing_design.mmd — the CURRENT state, generated from the AST using diagram_generator.py:

## Agent 1: current serialize_codebase_ast.py classes
python scripts/diagram_generator.py code2diagram \
    --ast plans/ast-ultimate.json \
    --filter scripts/serialize_codebase_ast \
    --output sandbox/context/existing_design.mmd

## Agent 2: current knowledge_graph/ module
python scripts/diagram_generator.py code2diagram \
    --ast plans/ast-ultimate.json \
    --filter knowledge_graph \
    --output sandbox/context/existing_design.mmd

proposed_design.mmd — the TARGET state after changes, hand-written by orchestrator from the spec. Shows the new classes, methods, and relationships the agent must implement.

Agent 1 proposed design includes:

• StatementInfo dataclass with all fields
• body_statements field on FunctionInfo and ModuleInfo
• _extract_statements() and _statement_to_info() on CodebaseAnalyzer
• Relationship: CodebaseAnalyzer → StatementInfo

Agent 2 proposed design includes:

• Neo4jSyncer class with all public methods
• flatten_statements() helper
• CLI wrapper structure
• Relationship: Neo4jSyncer → Neo4j driver, Neo4jSyncer reads CodebaseAST

The agent uses these diagrams to:

1. Understand the current structure before making changes
2. Know exactly what the target state looks like
3. Verify its implementation matches the proposed design

Audit Document (mandatory deliverable)

Every agent MUST create work/AUDIT.md as a deliverable. This is the agent's complete audit trail. The task is NOT complete without it.

AUDIT.md structure:

## Audit Document — [Agent Name]

### 1. Task Summary
- Task type: Extension / Creation
- Files modified/created: [list]
- REQs implemented: [list]

### 2. Design Comparison

#### Existing Design (from context/existing_design.mmd)
[Agent reproduces or references the existing design diagram]

#### Proposed Design (from context/proposed_design.mmd)
[Agent reproduces or references the proposed design diagram]

#### Implementation vs Proposed Design
| Proposed Element | Implemented? | Location (file:line) | Notes |
|-----------------|-------------|---------------------|-------|
| StatementInfo dataclass | YES | serialize_codebase_ast.py:111 | All 23 stmt types |
| _extract_statements() | YES | serialize_codebase_ast.py:200 | Recursive walker |
| ... | ... | ... | ... |

#### Deviations from Proposed Design
List any differences between proposed_design.mmd and actual implementation.
If none: "None — implementation matches proposed design exactly."

### 3. Changeset Report
[Full changeset report as specified above]

### 4. Traceability Matrix Coverage
| REQ | Status | Code Location | Evidence |
|-----|--------|--------------|----------|
| REQ-008-103 | IMPLEMENTED | serialize_codebase_ast.py:111-200 | StatementInfo + _extract_statements |
| REQ-008-104 | IMPLEMENTED | serialize_codebase_ast.py:450 | _analyze_file wiring |
| ... | ... | ... | ... |

### 5. Self-Audit Report
[16-row self-audit table as specified above]

### 6. Verification Results
[Test output from verify/ cycles — what passed, what failed, what was fixed]

The orchestrator reviews AUDIT.md alongside the actual code diff. Both must be consistent.

Orchestrator-Agent Protocol

The agent has NO shell access to the real repo. All execution happens through the orchestrator.

┌─────────────────────────────────────────────────────────┐
│                    ORCHESTRATOR (me)                      │
│                                                          │
│  1. Create sandbox dirs                                  │
│  2. Copy writable files to sandbox/work/                 │
│  3. Generate context/ from AST + subplan                 │
│  4. Dispatch agent into sandbox                          │
│  5. Agent writes code in sandbox/work/                   │
│  6. Agent signals "READY TO VERIFY"                      │
│  7. Orchestrator copies work/ files → real repo branch   │
│  8. Orchestrator runs tests in real repo                 │
│  9. Orchestrator copies test output → sandbox/verify/    │
│  10. If FAIL → agent reads verify/, fixes, goto 6        │
│  11. If PASS → agent produces changeset report + audit   │
│  12. Orchestrator diffs sandbox/work/ against originals  │
│  13. Human reviews diff                                  │
│  14. If approved → orchestrator commits to branch        │
└─────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────┐
│                    AGENT (sandboxed)                      │
│                                                          │
│  CAN:                                                    │
│  - Read context/* files                                  │
│  - Read/write work/* files                               │
│  - Read verify/* files (test results from orchestrator)  │
│  - Signal "READY TO VERIFY"                              │
│  - Produce changeset report + self-audit                 │
│                                                          │
│  CANNOT:                                                 │
│  - Access the real repo                                  │
│  - Run tests directly                                    │
│  - Run git commands                                      │
│  - Push to GitLab                                        │
│  - Modify files outside work/                            │
│  - Access the network                                    │
└─────────────────────────────────────────────────────────┘

Verify Loop (max 3 iterations)

Agent writes/modifies code in work/
    ↓
Agent signals "READY TO VERIFY"
    ↓
Orchestrator copies work/ files → real repo on branch
    ↓
Orchestrator runs:
  - uv run pytest tests/test_serialize_codebase_ast.py (WT1)
  - python scripts/serialize_codebase_ast.py serialize ... (WT1)
  - python scripts/sync_ast_to_neo4j.py schema/full/stats (WT2)
    ↓
Pass? → Agent produces changeset report + self-audit → Done
Fail? → Orchestrator writes test output to sandbox/verify/test_output.txt
    ↓
Agent reads failure output, diagnoses, fixes code in work/
    ↓
Agent signals "READY TO VERIFY" again (iteration 2)
    ↓
(max 3 iterations — if still failing, agent returns failure reason)

Post-Agent Verification (by orchestrator)

After agent signals completion with changeset report + self-audit:

## 1. Diff the work/ files against the originals
diff -u original/serialize_codebase_ast.py sandbox/work/serialize_codebase_ast.py

## 2. Verify ONLY expected files exist in work/
ls sandbox/work/  # must match allowlist exactly

## 3. Cross-reference changeset report against actual diff
## Any discrepancy = rejection

## 4. Copy to real repo, run full test suite one final time
## This is the orchestrator's own verification, independent of agent

What the AST Context Provides (replacing source file access)

The agent doesn't need to read source files because the AST JSON contains:

For WT1 (extension), the agent also gets the actual file in work/ — it reads the real code to understand patterns, then modifies it. The AST context supplements this with project-wide understanding.

For WT2 (creation), the agent builds from scratch using the spec + AST context. It never needs to read source files because the spec fully defines the schema.

Git Branch Strategy

Even though agents don't use git, the orchestrator still uses branches:

## Before dispatching agents
git checkout -b step0/ast-serializer
git checkout -b step0/neo4j-sync

## After agent 1 completes and is verified
git checkout step0/ast-serializer
cp sandbox-wt1/work/serialize_codebase_ast.py scripts/
git add scripts/serialize_codebase_ast.py
git commit -m "feat(ast): add statement-level AST capture to serializer"

## After agent 2 completes and is verified
git checkout step0/neo4j-sync
cp sandbox-wt2/work/sync.py src/orchestrata_ultimate/knowledge_graph/
cp sandbox-wt2/work/sync_ast_to_neo4j.py scripts/
git add src/orchestrata_ultimate/knowledge_graph/sync.py scripts/sync_ast_to_neo4j.py
git commit -m "feat(neo4j): add Neo4jSyncer engine with full AST-to-graph sync"

## Merge to master
git checkout master
git merge step0/ast-serializer --no-ff
git merge step0/neo4j-sync --no-ff

## DO NOT push — human audit required first

Prerequisite: Interface Tests on Master (before branching)

Before creating any worktrees, generate and commit interface tests for scripts/serialize_codebase_ast.py on master. These tests lock down the existing behavior so WT1 can prove it didn't break anything.

File: tests/test_serialize_codebase_ast.py

What to test (existing public API):

1. Dataclass instantiation & serialization (7 dataclasses):

   • ParamInfo, FunctionInfo, ClassVarInfo, ClassInfo, ImportInfo, ModuleInfo, CodebaseAST
   • Verify field defaults, JSON round-trip via serialize() function

2. CodebaseAnalyzer.analyze() — end-to-end:

   • Analyze src/orchestrata_ultimate (the real codebase, no mocking)
   • Verify returned CodebaseAST has modules, classes, methods, imports populated
   • Verify total_classes > 0, total_loc > 0, inheritance_graph non-empty

3. Standalone functions (key ones):

   • index_codebase() — verify lookup dicts populated from real AST JSON
   • classify_readiness() — test all 6 return values with synthetic pairs
   • pair_classes() — verify 3-pass pairing with plans/class-mapping.json
   • compare_methods_deep() — verify shared/only-in-a/only-in-b counts
   • audit_imports() — verify translation count > 0
   • serialize() — verify dataclass → dict conversion

4. CLI smoke tests (subprocess):

   • serialize subcommand with --source src/orchestrata_ultimate → valid JSON output
   • compare subcommand with two AST files → valid JSON output
   • deep-compare subcommand with mapping → valid JSON output

Standards: No mocking. Real codebase. Real files. pytest function-scope. Tests must pass before branching.

Commit: test(ast): add interface tests for serialize_codebase_ast.py

Prerequisite 2: Refactor Standalone Functions into Class (before branching)

After interface tests pass, refactor the 14 standalone comparison/analysis functions into a CodebaseComparator class. The interface tests are the safety net — they must all pass after the refactor.

Task type: Refactor (no new functionality, no behavior change)

Current state: 14 module-level functions passing dicts around:

• compare_codebases(), index_codebase(), resolve_chain(), summarize_method()
• compare_methods_deep(), classify_readiness(), pair_classes(), analyze_pair()
• path_to_package(), analyze_subsystems(), coverage_gaps_by_tier()
• audit_imports(), compare_dep_graphs(), complexity_metrics()
• deep_compare() (main entry point for deep-compare subcommand)

Target state: New class CodebaseComparator:

class CodebaseComparator:
    """Semantic comparison engine for two codebase ASTs."""

    def __init__(self, ast_a: dict, ast_b: dict, mapping: dict | None = None) -> None:
        self.ast_a = ast_a
        self.ast_b = ast_b
        self.mapping = mapping or {}
        self.idx_a = self._index_codebase(ast_a)
        self.idx_b = self._index_codebase(ast_b)

    # Indexing
    def _index_codebase(self, ast_data: dict) -> dict: ...
    def _resolve_chain(self, qualified: str, graph: dict, visited: set | None = None) -> list[str]: ...

    # Comparison
    def compare_shallow(self) -> dict: ...              # was compare_codebases()
    def compare_deep(self) -> dict: ...                 # was deep_compare() — main entry point
    def _compare_methods_deep(self, a_methods: list, b_methods: list) -> dict: ...
    def _classify_readiness(self, a_cls: dict | None, b_cls: dict | None) -> str: ...
    def _pair_classes(self) -> tuple[list[dict], set[str], set[str]]: ...
    def _analyze_pair(self, pair: dict) -> dict: ...
    def _summarize_method(self, m: dict) -> dict: ...

    # Analysis
    def _analyze_subsystems(self, tier_data: dict, mod_map: dict) -> list[dict]: ...
    def _coverage_gaps_by_tier(self, subsystems: list[dict]) -> dict: ...
    def _audit_imports(self, mod_map: dict) -> dict: ...
    def _compare_dep_graphs(self) -> dict: ...
    def _complexity_metrics(self) -> dict: ...
    def _path_to_package(self, path: str) -> str: ...

What stays module-level:

• serialize() — utility function, not comparison logic
• CodebaseAnalyzer — unchanged, handles parsing
• CLI entry points — updated to instantiate CodebaseComparator

CLI wiring change:

## Before (in deep-compare subcommand):
result = deep_compare(ast_a, ast_b, mapping)

## After:
comparator = CodebaseComparator(ast_a, ast_b, mapping)
result = comparator.compare_deep()

Verification: All interface tests from Prerequisite 1 must pass. Run the full serialize → compare → deep-compare pipeline and diff the output against a baseline snapshot.

Commit: refactor(ast): extract comparison functions into CodebaseComparator class

Agent 1 Scope: AST Serializer Extension

Task type: Extension — agent receives the actual file in work/, modifies it.

• Add StatementInfo dataclass (all 23 stmt types + ExceptHandler)
• Add _extract_statements() recursive walker
• Add _statement_to_info() handler (broadened: ast.stmt | ast.excepthandler)
• Add body_statements field to FunctionInfo and ModuleInfo
• Wire into _analyze_function() and _analyze_file()

Orchestrator runs verification (agent cannot):

## Copy modified file to real repo
cp sandbox-wt1/work/serialize_codebase_ast.py scripts/

## Run interface tests
uv run pytest tests/test_serialize_codebase_ast.py -v

## Run serialization
python scripts/serialize_codebase_ast.py serialize \
    --source src/orchestrata_ultimate \
    --name orchestrata-ultimate \
    --output /tmp/test-ast-statements.json

## Verify body_statements populated
python -c "
import json
data = json.load(open('/tmp/test-ast-statements.json'))
total = 0
for mod in data['modules']:
    total += len(mod.get('body_statements', []))
    for cls in mod.get('classes', []):
        for m in cls.get('methods', []):
            total += len(m.get('body_statements', []))
    for f in mod.get('functions', []):
        total += len(f.get('body_statements', []))
print(f'Total top-level statements: {total}')
assert total > 0, 'No statements captured!'
print('PASS')
"

Merge gate: interface tests pass + body_statements populated in JSON for methods and module-level.

Agent 2 Scope: Neo4j Sync Engine

Task type: Creation — agent creates new files in work/.

Architecture: Sync logic in sync.py (SPEC-008 module). CLI wrapper in sync_ast_to_neo4j.py.

Sync engine scope (sync.py):

• Neo4jSyncer class with methods for: create_schema, full_sync, map_classes, incremental_sync, get_stats, verify
• Nodes-first-edges-second sync pattern
• 12 node types, 18 relationship types
• Statement nodes with nesting (parent_statement_id)
• Call nodes with resolution
• Incremental update with cross-module edge repair
• flatten_statements() helper

Environment variables — NEVER hardcode connection strings:

class Neo4jSyncer:
    def __init__(
        self,
        uri: str | None = None,
        username: str | None = None,
        password: str | None = None,
    ) -> None:
        self.uri = uri or os.environ.get("NEO4J_URI", "bolt://localhost:7687")
        self.username = username or os.environ.get("NEO4J_USERNAME", "neo4j")
        self.password = password or os.environ.get("NEO4J_PASSWORD", "")

CLI wrapper provides --uri, --user, --password as optional overrides.

Agent 2 depends on Agent 1 merged — needs enriched AST JSON with body_statements.

Orchestrator runs verification:

## Copy files to real repo
cp sandbox-wt2/work/sync.py src/orchestrata_ultimate/knowledge_graph/
cp sandbox-wt2/work/sync_ast_to_neo4j.py scripts/

## Run sync pipeline
python scripts/sync_ast_to_neo4j.py schema
python scripts/sync_ast_to_neo4j.py full \
    --ast /tmp/test-ast-statements.json \
    --project orchestrata-ultimate
python scripts/sync_ast_to_neo4j.py stats

Then run all 14 verification Cypher queries from the Verification Checklist section above.

Merge gate: all 14 verification queries return results, stats show expected node counts.

Execution Order

master ── (tests) ── (refactor) ── (commit) ─────────> (merge Agent1) ──> (merge Agent2) ──>
                                       │                    ↑                   ↑
                                       │   ┌────────────────┘                   │
                                       │   │ orchestrator copies                │
                                       │   │ + verifies + commits              │
                                       ▼   │                                    │
                                    sandbox-wt1 (Agent 1)           sandbox-wt2 (Agent 2)
                                    work/ ← only writable           work/ ← only writable
                                    context/ ← read-only            context/ ← read-only

1. Commit pending session context changes to master
2. Prerequisite 1: Generate and commit interface tests for serializer on master
3. Run interface tests — all must pass
4. Prerequisite 2: Refactor standalone functions into CodebaseComparator class on master
5. Run interface tests again — all must still pass (proves refactor is behavior-preserving)
6. Commit refactor to master
7. Create sandbox directories, populate work/ and context/
8. Dispatch Agent 1 — modifies serialize_codebase_ast.py in sandbox
9. Agent 1 signals READY TO VERIFY → orchestrator copies to repo, runs tests
10. Verify loop (max 3 iterations) until tests pass
11. Agent 1 produces changeset report + self-audit
12. Orchestrator diffs sandbox/work/ against original, human reviews
13. If approved → orchestrator commits to step0/ast-serializer branch, merges to master
14. Dispatch Agent 2 — creates sync.py + CLI wrapper in sandbox
15. Agent 2 signals READY TO VERIFY → orchestrator copies to repo, runs sync + queries
16. Verify loop (max 3 iterations) until all 14 queries pass
17. Agent 2 produces changeset report + self-audit
18. Orchestrator diffs sandbox/work/ against empty (creation), human reviews
19. If approved → orchestrator commits to step0/neo4j-sync branch, merges to master
20. DO NOT push to GitLab — full human audit of all changes required first

Task Type Classification

Each agent prompt MUST declare the task type upfront. This changes how the agent approaches the work.

The self-audit report must include the task type and evidence that the agent followed the corresponding approach.

Mandatory Changeset Report (audit artifact)

Every agent MUST produce a changeset report listing EVERY change made, with justification. The orchestrator independently verifies this against the actual diff of work/ files.

### Changeset Report

#### Files in work/

| # | File | Action | Lines Changed | Justification (REQ ID) |
|---|------|--------|---------------|----------------------|
| 1 | serialize_codebase_ast.py | Modified | +250 -0 | REQ-008-103: StatementInfo, _extract_statements, body_statements |

#### Detailed Change Summary

For EACH file, list every logical change grouped by intent:

| Change | Lines | REQ | Notes |
|--------|-------|-----|-------|
| Added StatementInfo dataclass | 111-145 | REQ-008-103 | 23 stmt types + ExceptHandler |
| Added body_statements to FunctionInfo | 137 | REQ-008-103 | New field |
| ... | ... | ... | ... |

#### Out-of-Scope Changes

"None" or list with justification. **Cosmetic changes to existing code are PROHIBITED.**

The orchestrator verifies by diffing work/ files against originals. Any discrepancy between the report and the actual diff = rejection.

Coding & Testing Standards (injected into every agent)

Every agent MUST follow these standards. Violation of any standard means the task is NOT complete.

Python Standards:

Testing Standards:

Code Structure Standards:

Code Quality Standards:

Mandatory Self-Audit (task completion gate)

The task is NOT complete until the agent produces this self-audit table.

The agent must audit every file it created or modified against the standards above and produce a table in this exact format:

### Self-Audit Report

| # | Standard | Status | Evidence |
|---|----------|--------|----------|
| 1 | Task type followed | PASS/FAIL | Extension: read existing file first, ran before/after diff, no regressions. Creation: built from spec, no stale assumptions. |
| 2 | Python >=3.14 compatible | PASS/FAIL | [specific evidence] |
| 3 | Absolute imports only | PASS/FAIL | [specific evidence] |
| 4 | Type annotations on public API | PASS/FAIL | [specific evidence] |
| 5 | Docstrings on public classes/functions | PASS/FAIL | [specific evidence] |
| 6 | Business logic in classes | PASS/FAIL | [list every module-level function, justify why each is standalone or explain which class it belongs to] |
| 7 | Line length <=88 | PASS/FAIL | [ran black --check] |
| 8 | No mocking | PASS/FAIL | [grep for mock/patch] |
| 9 | No relative imports | PASS/FAIL | [grep for "from \."] |
| 10 | No speculative language in comments | PASS/FAIL | [grep for should/might/probably] |
| 11 | No AI attribution | PASS/FAIL | [checked git log] |
| 12 | Conventional commits | PASS/FAIL | [commit messages listed] |
| 13 | Verification executed | PASS/FAIL | [test output shown] |
| 14 | No unnecessary features | PASS/FAIL | [only spec requirements built] |
| 15 | Existing functionality preserved (Extension only) | PASS/FAIL/N/A | [ran existing commands before and after, output compared] |
| 16 | @traces decorators on all classes/public functions | PASS/FAIL | [list every class/function with its @traces REQ IDs, cross-reference traceability matrix] |

If any row is FAIL, the agent must fix it before reporting completion.

Traceability Matrix

All deliverables trace to SPEC-008 (Knowledge Graph) requirements in specs/008-knowledge-graph/checklists/requirements.md.

WT1: AST Serializer Extension

WT2: Neo4j Sync Script

Prerequisite: Refactor

@traces Decorator Requirement

Agents MUST add @traces decorators to all classes and public functions they create or modify. The decorator maps code to SPEC-008 requirements for RTM generation.

Import:

from orchestrata_ultimate.core.traceability.traceability_utils import traces

Usage examples:

@traces(spec="008", reqs=["REQ-008-103", "REQ-008-105"])
@dataclass
class StatementInfo:
    """AST statement node for knowledge graph sync."""
    ...

@traces(spec="008", reqs=["REQ-008-103", "REQ-008-106"])
def _extract_statements(self, body: list[ast.stmt], depth: int = 0) -> list[StatementInfo]:
    """Recursively extract all statements from a body list."""
    ...

@traces(spec="008", reqs=["REQ-008-108", "REQ-008-110", "REQ-008-113"])
class Neo4jSyncer:
    """Full AST-to-Neo4j sync engine."""
    ...

@traces(spec="008", reqs=["REQ-008-109"])
def create_schema(self) -> None:
    """Create constraints and indexes."""
    ...

Rules:

• Every class gets @traces with ALL REQ IDs it implements (from the traceability matrix above)
• Every public method/function that directly implements a requirement gets @traces
• Private helper methods that don't directly map to a requirement do NOT need @traces
• The self-audit must cross-reference every @traces decorator against the traceability matrix — no orphan traces, no missing traces

Phase 2 (Deferred — after Phase 1 verified)

After the graph is populated and all 13 structural verification queries pass:

1. Build the Scope class — loads a class's full subgraph from Neo4j in one query into an in-memory object with traversal methods (.services(), .constructor(), .validators(), .raises(), .stub_methods(), .env_var_accesses(), .ultimate_pair(), etc.). This is the interface the harness uses. All questions answered from memory, zero additional Neo4j calls per step.
2. Build graph_client.py (~60 LOC) — thin Neo4j driver wrapper that Scope uses internally for the initial load query.
3. Build context_builder.py — calls graph.load(project, class_name) to get a Scope, then formats its data for prompt injection. Replaces all heuristic grep.
4. Rewrite harness prompt templates to inject Scope-derived context.
5. Wire into harness cli.py + add incremental re-sync after each step merge.

Related

• Plans
• Orchestrata Ultimate — project index
• Workflow
• Tasks