Orchestrata Ultimate › Plans › codemod-engine-briefs › BRIEF-W1

BRIEF-W1: Forest + Sentinel (Qwen Max + Sonnet, Parallel)

Source: plans/codemod-engine-briefs/BRIEF-W1.md at 8a2cb42ad · last changed 2026-07-27 · first committed 2026-07-27 · 678 lines.

Prose pending.

Plan text

BRIEF-W1: Forest + Sentinel (Qwen Max + Sonnet, Parallel)

Wave Overview

Wave: W1 (Forest + Sentinel)
Agents:

• FOREST (Qwen Max): workspace.py, _csr_index.py, _cig.py, _cig_builder.py
• SENTINEL (Sonnet): txn.py
  Duration: ~20 minutes
  Dependencies: W0 (skeleton code)
  Parallel Execution: Yes (FOREST and SENTINEL work in parallel)

Objective

Implement the core data structures and algorithms for the codemod engine:

• FOREST: CST forest loading, Code Intelligence Graph (CIG), and CSR adjacency arrays
• SENTINEL: Transactional write-back with atomic commits and rollback

Scope

FOREST Agent (Qwen Max)

Files to Implement

1. src/orchestrata_ultimate/refactor/codemod/workspace.py (~400 LOC)

   • CodemodWorkspace class
   • load() classmethod (eager parse, parallel on 12 cores)
   • run() method (execute transform passes)
   • commit() method (review + write-back)
   • forest property (dict of path → libcst.Module)
   • unparsable property (list of failed files)
   • stats() method (WorkspaceStats)

2. src/orchestrata_ultimate/refactor/codemod/_csr_index.py (~200 LOC)

   • _CSRIndex class
   • __init__() (build CSR from edge list)
   • dependencies() (forward neighbors)
   • dependents() (reverse neighbors)
   • transitive_closure() (BFS with depth limit)

3. src/orchestrata_ultimate/refactor/codemod/_cig.py (~500 LOC)

   • CodeIntelligenceGraph class
   • dependencies(), dependents() (O(1) via CSR)
   • blast_radius() (BFS transitive closure)
   • breaking_changes() (signature diff + downstream consumers)
   • cycles() (Tarjan's SCC algorithm)
   • topological_order() (Kahn's algorithm)
   • affected_tests() (reverse TESTED_BY edges)
   • coupling_score() (efferent/afferent + instability)
   • abstraction_level() (Protocol/ABC ratio)
   • centrality() (betweenness centrality)
   • impact_score() (PageRank-style)

4. src/orchestrata_ultimate/refactor/codemod/_cig_builder.py (~300 LOC)

   • build_cig() function (extract nodes and edges from CST forest)
   • Node extraction (Module, Class, Function, Method, Parameter, Test, Import)
   • Edge extraction (IMPORTS, CALLS, INHERITS, DEFINES, TESTED_BY, OVERRIDES)

Files to Test

1. tests/refactor/codemod/test_workspace_load.py (~300 LOC)

   • 7 BDD scenarios from workspace_load.feature
   • Test eager parse (serial and parallel)
   • Test lossless round-trip
   • Test unparsable files
   • Test RSS delta

2. tests/refactor/codemod/test_workspace_index.py (~200 LOC)

   • 6 BDD scenarios from workspace_index.feature
   • Test modules() mapping
   • Test dependencies() and dependents()
   • Test symbol_table()

3. tests/refactor/codemod/test_code_intelligence_graph.py (~400 LOC)

   • 10 BDD scenarios from code_intelligence_graph.feature
   • Test all CIG query methods
   • Test performance (<2ms per query)

4. tests/refactor/codemod/test_csr_index.py (~200 LOC)

   • Plain pytest tests (not BDD)
   • Test CSR construction
   • Test forward/reverse lookups
   • Test transitive closure
   • Test edge cases (empty graph, self-loops, disconnected components)

SENTINEL Agent (Sonnet)

Files to Implement

1. src/orchestrata_ultimate/refactor/codemod/txn.py (~250 LOC)
   • commit_changes() function
   • _snapshot_files() (capture pre-commit bytes)
   • _write_atomic() (tempfile + fsync + rename)
   • _batch_fsync() (fsync directory fds)
   • _verify_post_write() (re-parse written files)
   • _rollback() (restore from snapshot)
   • _check_concurrent_edits() (mtime + hash + O_NOFOLLOW + flock)

Files to Test

1. tests/refactor/codemod/test_transactional_writeback.py (~300 LOC)
   • 6 BDD scenarios from transactional_writeback.feature
   • Test atomic write (tempfile + fsync + rename)
   • Test batch fsync (5 dirs instead of 100 files)
   • Test post-write verification
   • Test rollback on failure
   • Test concurrent-edit guard (mtime + hash)
   • Test symlink attack prevention (O_NOFOLLOW)
   • Test advisory locking (fcntl.flock)

Instructions for FOREST Agent

Step 1: Implement _csr_index.py

"""Compressed Sparse Row (CSR) adjacency arrays for O(1) graph queries."""

from __future__ import annotations
from dataclasses import dataclass
from typing import Sequence
import array


@dataclass(frozen=True, slots=True)
class _CSRIndex:
    """CSR adjacency arrays for forward and reverse lookups."""
    
    _offsets: array.array  # array('i', [...])
    _indices: array.array  # array('i', [...])
    _reverse_offsets: array.array
    _reverse_indices: array.array
    
    @classmethod
    def from_edges(cls, node_count: int, edges: Sequence[tuple[int, int]]) -> _CSRIndex:
        """Build CSR from edge list."""
        # Count out-degree and in-degree for each node
        out_degree = [0] * node_count
        in_degree = [0] * node_count
        for src, dst in edges:
            out_degree[src] += 1
            in_degree[dst] += 1
        
        # Build forward CSR (outgoing edges)
        offsets = array.array('i', [0] * (node_count + 1))
        for i in range(node_count):
            offsets[i + 1] = offsets[i] + out_degree[i]
        
        indices = array.array('i', [0] * len(edges))
        cursor = list(offsets)
        for src, dst in edges:
            indices[cursor[src]] = dst
            cursor[src] += 1
        
        # Build reverse CSR (incoming edges)
        reverse_offsets = array.array('i', [0] * (node_count + 1))
        for i in range(node_count):
            reverse_offsets[i + 1] = reverse_offsets[i] + in_degree[i]
        
        reverse_indices = array.array('i', [0] * len(edges))
        cursor = list(reverse_offsets)
        for src, dst in edges:
            reverse_indices[cursor[dst]] = src
            cursor[dst] += 1
        
        return cls(offsets, indices, reverse_offsets, reverse_indices)
    
    def dependencies(self, node_id: int) -> list[int]:
        """Get outgoing neighbors (dependencies)."""
        start = self._offsets[node_id]
        end = self._offsets[node_id + 1]
        return list(self._indices[start:end])
    
    def dependents(self, node_id: int) -> list[int]:
        """Get incoming neighbors (dependents)."""
        start = self._reverse_offsets[node_id]
        end = self._reverse_offsets[node_id + 1]
        return list(self._reverse_indices[start:end])
    
    def transitive_closure(self, node_id: int, max_depth: int = -1) -> set[int]:
        """BFS transitive closure with optional depth limit."""
        visited = set()
        queue = [(node_id, 0)]
        
        while queue:
            current, depth = queue.pop(0)
            if current in visited:
                continue
            visited.add(current)
            
            if max_depth >= 0 and depth >= max_depth:
                continue
            
            for neighbor in self.dependencies(current):
                if neighbor not in visited:
                    queue.append((neighbor, depth + 1))
        
        return visited

Step 2: Implement _cig_builder.py

"""Build Code Intelligence Graph from CST forest."""

from __future__ import annotations
from pathlib import Path
from typing import Dict, List, Tuple
import libcst as cst
from orchestrata_ultimate.refactor.codemod._cig import CodeIntelligenceGraph, CIGNode, CIGEdge, CIGEdgeType


def build_cig(forest: Dict[str, cst.Module]) -> CodeIntelligenceGraph:
    """Extract nodes and edges from CST forest."""
    nodes: Dict[str, CIGNode] = {}
    edges: List[CIGEdge] = []
    
    # Assign node IDs
    node_id_counter = 0
    
    for path, module in forest.items():
        # Create Module node
        module_id = node_id_counter
        node_id_counter += 1
        nodes[f"module:{path}"] = CIGNode(
            node_id=module_id,
            node_type="Module",
            name=path,
            qualified_name=path,
            path=path,
            line=1,
            properties={}
        )
        
        # Extract classes, functions, imports
        visitor = _CIGVisitor(module_id, path, nodes, edges, node_id_counter)
        module.visit(visitor)
        node_id_counter = visitor.node_id_counter
    
    # Build CSR index
    edge_tuples = [(e.source_id, e.target_id) for e in edges]
    csr = _CSRIndex.from_edges(len(nodes), edge_tuples)
    
    return CodeIntelligenceGraph(nodes, edges, csr)


class _CIGVisitor(cst.CSTVisitor):
    """Extract nodes and edges from CST."""
    
    def __init__(self, module_id: int, path: str, nodes: Dict, edges: List, node_id_counter: int):
        self.module_id = module_id
        self.path = path
        self.nodes = nodes
        self.edges = edges
        self.node_id_counter = node_id_counter
        self.current_class = None
    
    def visit_ClassDef(self, node: cst.ClassDef) -> None:
        """Extract Class node and DEFINES edge."""
        class_name = node.name.value
        qualified_name = f"{self.path}:{class_name}"
        
        class_id = self.node_id_counter
        self.node_id_counter += 1
        
        self.nodes[f"class:{qualified_name}"] = CIGNode(
            node_id=class_id,
            node_type="Class",
            name=class_name,
            qualified_name=qualified_name,
            path=self.path,
            line=node.lineno,
            properties={}
        )
        
        # DEFINES edge: Module → Class
        self.edges.append(CIGEdge(
            source_id=self.module_id,
            target_id=class_id,
            edge_type=CIGEdgeType.DEFINES
        ))
        
        self.current_class = class_id
    
    def leave_ClassDef(self, node: cst.ClassDef) -> None:
        self.current_class = None
    
    def visit_FunctionDef(self, node: cst.FunctionDef) -> None:
        """Extract Function/Method node and DEFINES edge."""
        func_name = node.name.value
        qualified_name = f"{self.path}:{func_name}"
        
        func_id = self.node_id_counter
        self.node_id_counter += 1
        
        node_type = "Method" if self.current_class else "Function"
        
        self.nodes[f"{node_type.lower()}:{qualified_name}"] = CIGNode(
            node_id=func_id,
            node_type=node_type,
            name=func_name,
            qualified_name=qualified_name,
            path=self.path,
            line=node.lineno,
            properties={}
        )
        
        # DEFINES edge: Module → Function or Class → Method
        parent_id = self.current_class if self.current_class else self.module_id
        self.edges.append(CIGEdge(
            source_id=parent_id,
            target_id=func_id,
            edge_type=CIGEdgeType.DEFINES
        ))
    
    def visit_Import(self, node: cst.Import) -> None:
        """Extract IMPORTS edge."""
        # TODO: Parse import statement and create IMPORTS edge
        pass
    
    def visit_ImportFrom(self, node: cst.ImportFrom) -> None:
        """Extract IMPORTS edge."""
        # TODO: Parse import statement and create IMPORTS edge
        pass
    
    def visit_Call(self, node: cst.Call) -> None:
        """Extract CALLS edge."""
        # TODO: Resolve call target and create CALLS edge
        pass

Step 3: Implement _cig.py

Implement all CIG query methods. Key algorithms:

• Tarjan's SCC for cycle detection
• Kahn's algorithm for topological sort
• BFS for blast radius and affected tests
• Signature diff for breaking changes
• Coupling metrics (efferent/afferent/instability)
• Betweenness centrality (Brandes algorithm)
• PageRank for impact score

Step 4: Implement workspace.py

"""CodemodWorkspace: Main entry point for loading and transforming code."""

from __future__ import annotations
from pathlib import Path
from typing import Sequence, Optional
from concurrent.futures import ThreadPoolExecutor
import time
import psutil
import libcst as cst
from orchestrata_ultimate.refactor.codemod.report import (
    ChangeSet, CommitReport, WorkspaceStats, CodemodEvent, CodemodEventType
)
from orchestrata_ultimate.refactor.codemod._cig import CodeIntelligenceGraph
from orchestrata_ultimate.refactor.codemod._cig_builder import build_cig
from orchestrata_ultimate.refactor.codemod.passes import TransformPass
from orchestrata_ultimate.refactor.codemod.review import ReviewPass
from orchestrata_ultimate.refactor.codemod.txn import commit_changes


class CodemodWorkspace:
    """In-memory CST forest with Code Intelligence Graph."""
    
    def __init__(self, forest: dict[str, cst.Module], unparsable: list[tuple[str, str]], cig: CodeIntelligenceGraph, stats: WorkspaceStats):
        self._forest = forest
        self._unparsable = unparsable
        self._cig = cig
        self._stats = stats
    
    @classmethod
    def load(cls, root: Path, include: tuple[str, ...] = ("src", "scripts", "tests"), parallel: bool = True) -> CodemodWorkspace:
        """Load codebase into memory (eager parse)."""
        # Measure RSS before
        process = psutil.Process()
        rss_before = process.memory_info().rss
        
        # Discover files
        files = []
        for subdir in include:
            subdir_path = root / subdir
            if subdir_path.exists():
                files.extend(subdir_path.rglob("*.py"))
        
        # Parse files (parallel or serial)
        start_time = time.time()
        
        if parallel:
            with ThreadPoolExecutor(max_workers=12) as executor:
                results = list(executor.map(_parse_file, files))
        else:
            results = [_parse_file(f) for f in files]
        
        parse_time = time.time() - start_time
        
        # Separate successful and failed parses
        forest = {}
        unparsable = []
        for path, result in zip(files, results):
            if isinstance(result, cst.Module):
                forest[str(path.relative_to(root))] = result
            else:
                unparsable.append((str(path.relative_to(root)), str(result)))
        
        # Build CIG
        cig = build_cig(forest)
        
        # Measure RSS after
        rss_after = process.memory_info().rss
        rss_delta = rss_after - rss_before
        
        stats = WorkspaceStats(
            file_count=len(forest),
            total_bytes=sum(len(m.code.encode('utf-8')) for m in forest.values()),
            rss_delta_bytes=rss_delta,
            parse_time_seconds=parse_time,
            unparsable_count=len(unparsable)
        )
        
        return cls(forest, unparsable, cig, stats)
    
    @property
    def forest(self) -> dict[str, cst.Module]:
        """Get the CST forest."""
        return self._forest
    
    @property
    def unparsable(self) -> list[tuple[str, str]]:
        """Get list of unparsable files."""
        return self._unparsable
    
    @property
    def cig(self) -> CodeIntelligenceGraph:
        """Get the Code Intelligence Graph."""
        return self._cig
    
    def stats(self) -> WorkspaceStats:
        """Get workspace statistics."""
        return self._stats
    
    def run(self, passes: Sequence[TransformPass]) -> ChangeSet:
        """Execute transform passes sequentially."""
        # TODO: Implement in W2
        raise NotImplementedError()
    
    def commit(self, changes: ChangeSet, reviews: Sequence[ReviewPass] | None = None, dry_run: bool = True) -> CommitReport:
        """Review and commit changes."""
        # TODO: Implement in W2 (review) and W1 (txn)
        raise NotImplementedError()


def _parse_file(path: Path) -> cst.Module | Exception:
    """Parse a single Python file."""
    try:
        code = path.read_text(encoding='utf-8')
        return cst.parse_module(code)
    except Exception as e:
        return e

Instructions for SENTINEL Agent

Step 1: Implement txn.py

"""Transactional write-back with atomic commits and rollback."""

from __future__ import annotations
from pathlib import Path
from typing import Sequence
import os
import tempfile
import fcntl
import xxhash
import libcst as cst
from orchestrata_ultimate.refactor.codemod.report import ChangeSet, ChangeEntry


def commit_changes(
    root: Path,
    changes: ChangeSet,
    dry_run: bool = True
) -> tuple[list[str], list[str]]:
    """
    Commit changes to disk with atomic writes and rollback.
    
    Returns:
        (files_written, files_rolled_back)
    """
    if dry_run:
        return ([], [])
    
    # Snapshot pre-commit bytes
    snapshot = _snapshot_files(root, changes)
    
    # Check concurrent edits
    _check_concurrent_edits(root, snapshot)
    
    # Write files atomically
    files_written = []
    try:
        for entry in changes.entries:
            _write_atomic(root, entry)
            files_written.append(entry.path)
        
        # Batch fsync directories
        _batch_fsync(root, files_written)
        
        # Verify post-write
        _verify_post_write(root, files_written)
        
        return (files_written, [])
    
    except Exception as e:
        # Rollback on failure
        files_rolled_back = _rollback(root, snapshot)
        raise RuntimeError(f"Commit failed, rolled back {len(files_rolled_back)} files: {e}")


def _snapshot_files(root: Path, changes: ChangeSet) -> dict[str, bytes]:
    """Capture pre-commit bytes for rollback."""
    snapshot = {}
    for entry in changes.entries:
        path = root / entry.path
        if path.exists():
            snapshot[entry.path] = path.read_bytes()
    return snapshot


def _check_concurrent_edits(root: Path, snapshot: dict[str, bytes]) -> None:
    """Check if files were modified since snapshot."""
    for path_str, original_bytes in snapshot.items():
        path = root / path_str
        
        # Open with O_NOFOLLOW to prevent symlink attacks
        fd = os.open(str(path), os.O_RDONLY | os.O_NOFOLLOW)
        try:
            # Acquire advisory lock
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
            
            # Read current bytes
            current_bytes = os.read(fd, os.path.getsize(fd))
            
            # Check if modified
            if current_bytes != original_bytes:
                raise RuntimeError(f"File {path_str} was modified concurrently")
            
            # Release lock
            fcntl.flock(fd, fcntl.LOCK_UN)
        finally:
            os.close(fd)


def _write_atomic(root: Path, entry: ChangeEntry) -> None:
    """Write file atomically using tempfile + fsync + rename."""
    path = root / entry.path
    new_bytes = entry.mutated_tree.code.encode('utf-8')
    
    # Write to tempfile in same directory
    dir_path = path.parent
    with tempfile.NamedTemporaryFile(mode='wb', dir=dir_path, delete=False) as tmp:
        tmp.write(new_bytes)
        tmp.flush()
        os.fsync(tmp.fileno())
        tmp_path = tmp.name
    
    # Atomic rename
    os.rename(tmp_path, str(path))


def _batch_fsync(root: Path, files_written: list[str]) -> None:
    """Fsync directory file descriptors (not per-file)."""
    dirs_to_sync = set()
    for path_str in files_written:
        dir_path = (root / path_str).parent
        dirs_to_sync.add(dir_path)
    
    for dir_path in dirs_to_sync:
        fd = os.open(str(dir_path), os.O_RDONLY)
        try:
            os.fsync(fd)
        finally:
            os.close(fd)


def _verify_post_write(root: Path, files_written: list[str]) -> None:
    """Re-parse written files to verify correctness."""
    for path_str in files_written:
        path = root / path_str
        code = path.read_text(encoding='utf-8')
        try:
            cst.parse_module(code)
        except Exception as e:
            raise RuntimeError(f"Post-write verification failed for {path_str}: {e}")


def _rollback(root: Path, snapshot: dict[str, bytes]) -> list[str]:
    """Restore files from snapshot."""
    files_rolled_back = []
    for path_str, original_bytes in snapshot.items():
        path = root / path_str
        path.write_bytes(original_bytes)
        files_rolled_back.append(path_str)
    return files_rolled_back

Step 2: Implement test_transactional_writeback.py

Write BDD scenarios from transactional_writeback.feature. Key tests:

• Atomic write (tempfile + fsync + rename)
• Batch fsync (5 dirs instead of 100 files)
• Post-write verification
• Rollback on failure
• Concurrent-edit guard (mtime + hash)
• Symlink attack prevention (O_NOFOLLOW)
• Advisory locking (fcntl.flock)

Success Criteria

FOREST Agent

• [ ] _csr_index.py implemented with O(1) lookups
• [ ] _cig_builder.py extracts all node and edge types
• [ ] _cig.py implements all 11 query methods
• [ ] workspace.py loads codebase in <120s (parallel)
• [ ] All BDD scenarios pass (workspace_load, workspace_index, code_intelligence_graph)
• [ ] CSR index tests pass (plain pytest)
• [ ] Performance targets met (<2ms per CIG query)

SENTINEL Agent

• [ ] txn.py implements atomic writes with rollback
• [ ] Batch fsync reduces I/O overhead (5 dirs vs 100 files)
• [ ] Concurrent-edit guard detects modifications
• [ ] Symlink attacks prevented (O_NOFOLLOW)
• [ ] Advisory locking prevents concurrent writes
• [ ] All BDD scenarios pass (transactional_writeback)

Deliverables

FOREST Agent

1. src/orchestrata_ultimate/refactor/codemod/workspace.py
2. src/orchestrata_ultimate/refactor/codemod/_csr_index.py
3. src/orchestrata_ultimate/refactor/codemod/_cig.py
4. src/orchestrata_ultimate/refactor/codemod/_cig_builder.py
5. tests/refactor/codemod/test_workspace_load.py
6. tests/refactor/codemod/test_workspace_index.py
7. tests/refactor/codemod/test_code_intelligence_graph.py
8. tests/refactor/codemod/test_csr_index.py

SENTINEL Agent

1. src/orchestrata_ultimate/refactor/codemod/txn.py
2. tests/refactor/codemod/test_transactional_writeback.py

Handoff to W2

W2 (Architect + Reviewer) will implement:

• passes.py (5 built-in transform passes)
• review.py (40 in-memory review checks)
• analysis/*.py (7 analysis modules)
• observability/*.py (5 observer classes)

W2 agents will use the workspace, CIG, and txn from W1 to implement passes and reviews.

Notes

• FOREST and SENTINEL work in parallel (no dependencies between them)
• FOREST agent should use Qwen Max (strong at algorithms and graph theory)
• SENTINEL agent should use Sonnet (strong at systems programming and error handling)
• Both agents must use the skeleton data structures from W0 (ChangeEntry, ChangeSet, etc.)
• All code must have @traces decorators with correct spec/story/reqs
• All tests must use BDD scenarios from feature files (except CSR index tests)

Related

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