CLAUDE.md

1# Code Confluence Flow Bridge - Claude Code Integration Guide
2 
3## Agent Context
4 
5Also read `AGENTS.md` in this directory for auto-generated dependency guide, business logic domain map, and app interface inventory. Companion files `business_logic_references.md` and `app_interfaces.md` provide detailed module and endpoint references.
6 
7## Project Overview
8 
9Code Confluence Flow Bridge is the backend ingestion service for the Unoplat Code Confluence system. It orchestrates repository processing workflows using Temporal, parses code structures using Tree-sitter, and stores results in Neo4j graph and PostgreSQL relational databases.
10 
11### Key Responsibilities
12- **Workflow Orchestration**: Temporal-based workflows for repository and codebase processing
13- **Code Parsing**: Language-agnostic parsing using Tree-sitter (Python and TypeScript support)
14- **Graph Persistence**: Neo4j storage for code structure and relationships
15- **Metadata Management**: PostgreSQL for workflow state, credentials, and configuration
16- **Multi-language Support**: Framework detection and package manager analysis
17 
18## Essential Development Commands
19 
20### Dependency Management
21```bash
22# Sync dependencies (required before running)
23task sync
24 
25# Update a specific package
26task update-package PACKAGE=package_name
27 
28# Update all packages
29task update-all-packages
30```
31 
32### Development Workflow
33```bash
34# Start all dependencies (PostgreSQL, Neo4j, Temporal, Elasticsearch)
35task start-dependencies
36 
37# Start only core dependencies (without Signoz/OpenTelemetry)
38task start-core-dependencies
39 
40# Run FastAPI server in development mode
41task run-dev        # Requires dependencies already running
42task dev            # Starts dependencies then runs FastAPI
43 
44# Stop all dependencies
45task stop-dependencies
46task stop-core-dependencies
47```
48 
49### Testing & Code Quality
50```bash
51# Run tests with coverage
52task test
53 
54# Linting
55task lint           # Check with ruff
56task lint-fix       # Auto-fix issues
57 
58# Type checking
59task typecheck      # Full type check
60task typecheck-file FILE=path/to/file.py
61 
62# Code formatting
63task format         # Format with ruff
64 
65# Framework definition validation
66task validate-framework-definitions
67task validate-single-framework FILE=framework-definitions/python/fastapi.json
68```
69 
70### GitHub Integration
71```bash
72# Submit test job to flow bridge
73task run-client    # Uses CLI config to test ingestion
74 
75# Run GitHub Actions locally
76task run-github-action-locally
77```
78 
79## Architecture Overview
80 
81### Service Architecture
82 
83```
84┌─────────────────────────────────────────────────────────────┐
85│                 FastAPI Application (main.py)                │
86│  - Token management endpoints                                │
87│  - Repository discovery (GitHub GraphQL)                    │
88│  - Ingestion orchestration                                  │
89│  - Status tracking and deletion                             │
90└──────────────────────┬──────────────────────────────────────┘
91                       │
92        ┌──────────────┼──────────────┐
93        │              │              │
94        ▼              ▼              ▼
95   Temporal      PostgreSQL        Neo4j
96   Workflows     (Metadata)      (Graph DB)
97        │
98   ┌────┴─────┬──────────┬──────────────┐
99   │           │          │              │
100   ▼           ▼          ▼              ▼
101RepoWorkflow  Git      Package      Codebase
102             Activity  Metadata    Processing
103                      Activity     Activity
104```
105 
106### Workflow Orchestration (Temporal)
107 
108**Parent Workflow**: `RepoWorkflow` (class-based)
109- Orchestrates entire repository ingestion process
110- Manages child workflows for each codebase
111- Task Queue: `unoplat-code-confluence-repository-context-ingestion`
112 
113**Key Activities** (executed by worker pool):
1141. **GitActivity** - Clones repository, extracts metadata
1152. **ConfluenceGitGraph** - Inserts repository structure into Neo4j
1163. **PackageMetadataActivity** - Extracts dependency information
1174. **PackageManagerMetadataIngestion** - Stores package data in Neo4j
1185. **GenericCodebaseProcessingActivity** - Parses code and creates graph nodes
1196. **ChildWorkflowDbActivity** - Updates codebase workflow status in PostgreSQL
1207. **ParentWorkflowDbActivity** - Updates repository workflow status in PostgreSQL
121 
122**Child Workflow**: `CodebaseChildWorkflow`
123- Processes individual codebases detected in repository
124- Handles language-specific parsing and metadata extraction
125 
126### Data Flow
127 
1281. **Repository Discovery**
129   - Frontend requests repository list via GraphQL
130   - Backend fetches from GitHub (GITHUB_OPEN or GITHUB_ENTERPRISE)
131   - Credentials encrypted/stored in PostgreSQL
132 
1332. **Codebase Detection**
134   - Auto-detection using multi-language ripgrep detectors
135   - Supported: Python (PythonRipgrepDetector), TypeScript (TypeScriptRipgrepDetector)
136   - Returns: List of CodebaseConfig with metadata
137 
1383. **Ingestion Workflow**
139   - Temporal workflow clones repository to `/opt/unoplat/repositories`
140   - GitActivity creates UnoplatGitRepository structure
141   - ConfluenceGitGraph inserts repository node into Neo4j
142   - Child workflows spawn for each detected codebase
143 
1444. **Code Parsing**
145   - GenericCodebaseParser uses language processors
146   - Tree-sitter extracts structural signatures
147   - Framework detection (Python: FastAPI, Django, Flask, etc.)
148   - Package manager parsing (pip, uv, poetry, npm, yarn, etc.)
149 
1505. **Persistence**
151   - PostgreSQL: Workflow runs, credentials, framework definitions
152   - Neo4j: Repository, File, Class, Function, Package nodes with relationships
153 
154### Database Configuration
155 
156**PostgreSQL** (via SQLAlchemy async + asyncpg)
157- Connection URL: `postgresql+asyncpg://user:password@host:port/db`
158- Session management: Per-loop async engine pattern (see `db.py`)
159- Models: SQLBase from commons (Repository, CodebaseConfig, RepositoryWorkflowRun, CodebaseWorkflowRun)
160 
161**Neo4j** (via neomodel + neo4j driver)
162- Connection URL: `bolt://user:password@host:port`
163- Uses async driver for non-blocking operations
164- Global connection via `CodeConfluenceGraph` singleton
165- Schema: Installed at startup via `adb.install_all_labels()`
166 
167**Environment Variables** (from `EnvironmentSettings`):
168```python
169# PostgreSQL
170DB_HOST, DB_PORT, DB_USER, DB_PASSWORD, DB_NAME
171 
172# Neo4j
173NEO4J_HOST, NEO4J_PORT, NEO4J_USERNAME, NEO4J_PASSWORD
174NEO4J_MAX_CONNECTION_LIFETIME, NEO4J_MAX_CONNECTION_POOL_SIZE
175NEO4J_CONNECTION_ACQUISITION_TIMEOUT
176 
177# Temporal
178TEMPORAL_SERVER_ADDRESS (default: localhost:7233)
179TEMPORAL_MAX_CONCURRENT_ACTIVITIES (default: 4)
180TEMPORAL_ENABLE_POLLER_AUTOSCALING (default: false)
181 
182# Repositories
183REPOSITORIES_BASE_PATH (default: ~/.unoplat/repositories)
184 
185# Framework definitions
186FRAMEWORK_DEFINITIONS_PATH (default: /framework-definitions)
187```
188 
189## Important Architecture Patterns
190 
191### 1. Envelope Pattern (Temporal Parameters)
192 
193All Temporal workflow/activity parameters are wrapped in envelope models to handle extra fields:
194 
195```python
196# Example: RepoWorkflowRunEnvelope
197@dataclass
198class RepoWorkflowRunEnvelope(BaseModel):
199    repo_request: RepositoryRequestConfiguration  # Main payload
200    github_token: str
201    trace_id: str
202    model_config = ConfigDict(extra="allow")  # Allow extra fields
203    
204    @property
205    def extras(self) -> dict[str, Any]:
206        return dict(self.model_extra or {})
207```
208 
209All envelope models support `extra="allow"` to handle Pydantic data converter requirements.
210 
211### 2. Async Database Session Management
212 
213Critical pattern for handling multiple event loops:
214 
215```python
216# Gets or creates AsyncEngine per event loop
217engine, session_factory = await get_engine_for_loop()
218 
219# Yields session for dependency injection
220async def get_session() -> AsyncGenerator[AsyncSession, None]:
221    async with session_factory() as session:
222        async with session.begin():  # Explicit transaction
223            yield session
224            # Commit on __exit__, rollback on exception
225 
226# For context managers (activities)
227async with get_session_cm() as session:
228    await session.execute(...)  # Auto-committed on context exit
229```
230 
231**Key Points**:
232- Each event loop gets its own AsyncEngine (fixes Temporal activity issues)
233- `expire_on_commit=False` critical for async (prevents implicit I/O)
234- Always use `async with session.begin()` for explicit transactions
235- Activities in Temporal need context manager pattern for cleanup
236 
237### 3. Temporal Worker Configuration
238 
239Worker handles both parent and child workflows with interceptors:
240 
241```python
242worker = Worker(
243    client,
244    task_queue="unoplat-code-confluence-repository-context-ingestion",
245    workflows=[RepoWorkflow, CodebaseChildWorkflow],
246    activities=activities,
247    activity_executor=ThreadPoolExecutor(size=max_concurrent + 4),
248    interceptors=[
249        ParentWorkflowStatusInterceptor(),  # Updates parent workflow status
250        ActivityStatusInterceptor(),        # Updates activity status
251    ],
252    max_concurrent_activities=env.temporal_max_concurrent_activities,
253)
254```
255 
256### 4. Parser Factory Pattern
257 
258Language-specific parsing via strategy pattern:
259 
260```python
261# Generic parser delegates to language-specific processors
262parser = GenericCodebaseParser(
263    codebase_name="my-app",
264    codebase_path="/repo/src",
265    language_metadata=ProgrammingLanguageMetadata(...),
266)
267 
268# Internally uses:
269# - PythonLanguageProcessor (wraps TreeSitterPythonStructuralSignatureExtractor)
270# - TypeScriptLanguageProcessor
271# - Custom processors for other languages
272```
273 
274### 5. Package Manager Strategy
275 
276Different strategies for different package managers:
277 
278```python
279# Strategy pattern for package manager processing
280class PackageManagerStrategy(ABC):
281    @abstractmethod
282    def process_metadata(
283        self, 
284        local_workspace_path: str, 
285        metadata: ProgrammingLanguageMetadata
286    ) -> UnoplatPackageManagerMetadata:
287        pass
288```
289 
290Implementations available for: pip, uv, poetry, npm, yarn, maven, etc.
291 
292### 6. Neo4j Session Management
293 
294Uses neomodel's managed transactions pattern:
295 
296```python
297# Get session from global connection
298async with code_confluence_graph.get_session() as neo4j_session:
299    # Use session.execute_write() or execute_read() for managed transactions
300    result = await neo4j_session.execute_write(some_function, arg1, arg2)
301```
302 
303Automatic retry and transaction management via Neo4j driver.
304 
305### 7. Logging & Tracing
306 
307Distributed tracing with Loguru and OpenTelemetry:
308 
309```python
310# ContextVar-based trace propagation
311from code_confluence_flow_bridge.logging.trace_utils import (
312    trace_id_var,
313    workflow_id_var,
314    activity_name_var,
315)
316 
317# Bind trace context to logger
318log = seed_and_bind_logger_from_trace_id(
319    trace_id=trace_id,
320    workflow_id=workflow_id,
321    workflow_run_id=run_id
322)
323 
324# Logs include trace context automatically
325log.info("Processing started", extra={"activity": "git_clone"})
326```
327 
328Optional OTLP export to SigNoz when `OTEL_EXPORTER_OTLP_ENDPOINT` is set.
329 
330### 8. Error Handling Pattern
331 
332Standardized error context for debugging:
333 
334```python
335try:
336    # Operation
337except Exception as e:
338    error_context = {
339        "workflow_id": workflow_id,
340        "activity_name": "process_git",
341        "error_details": str(e),
342        "traceback": traceback.format_exc(),
343    }
344    logger.error("Operation failed: {}", str(e), extra={"error_context": error_context})
345    # Re-raise as ApplicationError for Temporal
346    raise ApplicationError(str(e), type="CUSTOM_ERROR_TYPE") from e
347```
348 
349## Key Directory Structure
350 
351```
352src/code_confluence_flow_bridge/
353├── main.py                          # FastAPI app, lifespan, endpoints
354├── logging/
355│   ├── log_config.py               # Loguru + OTLP setup
356│   └── trace_utils.py              # ContextVar-based trace propagation
357├── models/
358│   ├── configuration/
359│   │   └── settings.py             # EnvironmentSettings (all env vars)
360│   ├── github/
361│   │   └── github_repo.py          # Request/response models
362│   ├── workflow/
363│   │   └── repo_workflow_base.py   # Envelope models
364│   └── code_confluence_parsing_models/
365│       ├── unoplat_git_repository.py
366│       ├── unoplat_package_manager_metadata.py
367│       └── unoplat_file.py
368├── processor/                      # Temporal workflows & activities
369│   ├── repo_workflow.py            # Parent workflow orchestration
370│   ├── codebase_child_workflow.py  # Child workflow for each codebase
371│   ├── db/
372│   │   ├── postgres/
373│   │   │   ├── db.py              # AsyncEngine & session management
374│   │   │   ├── parent_workflow_db_activity.py
375│   │   │   ├── child_workflow_db_activity.py
376│   │   │   └── framework_loader.py
377│   │   └── graph_db/
378│   │       ├── code_confluence_graph.py          # Neo4j connection
379│   │       ├── code_confluence_graph_ingestion.py
380│   │       └── code_confluence_graph_deletion.py
381│   ├── git_activity/
382│   │   ├── confluence_git_activity.py   # Clones repo activity
383│   │   └── confluence_git_graph.py      # Inserts into Neo4j
384│   ├── package_metadata_activity/       # Package extraction
385│   └── activity_*.py                    # Interceptors for status updates
386├── parser/
387│   ├── generic_codebase_parser.py  # Main parser (delegates to processors)
388│   ├── tree_sitter_*.py           # Tree-sitter utilities
389│   ├── language_processors/
390│   │   ├── base.py                # Abstract processor
391│   │   ├── python_processor.py
392│   │   └── typescript_processor.py
393│   └── package_manager/
394│       ├── package_manager_strategy.py
395│       ├── package_manager_factory.py
396│       └── detectors/              # Codebase detection
397├── detector/
398│   ├── base_detector.py
399│   └── ripgrep_*.py               # Ripgrep-based detection
400├── engine/
401│   ├── framework_detection_service.py
402│   ├── python/
403│   │   └── python_framework_detection_service.py
404│   └── generic_filters.py
405└── github_app/
406    └── router.py                   # GitHub App webhook endpoints
407```
408 
409## Common Development Workflows
410 
411### Adding a New Framework
412 
4131. Add framework definition JSON to `framework-definitions/{language}/framework-name.json`
4142. Validate: `task validate-single-framework FILE=...`
4153. Framework loaded automatically at startup via `FrameworkDefinitionLoader`
416 
417### Implementing Language Support
418 
4191. Create `LanguageCodebaseProcessor` subclass in `parser/language_processors/`
4202. Implement Tree-sitter extraction via `TreeSitterExtractorBase`
4213. Register in `GenericCodebaseParser.LANGUAGE_PROCESSORS` mapping
4224. Add detector (ripgrep-based) in `detector/`
4235. Add to `main.py` lifespan detectors registration
424 
425### Testing Ingestion Locally
426 
427```bash
428# Terminal 1: Start dependencies
429task start-dependencies
430 
431# Terminal 2: Run development server
432task run-dev
433 
434# Terminal 3: Submit test job
435task run-client
436 
437# Monitor: Check Temporal UI at http://localhost:8081
438```
439 
440### Debugging Workflows
441 
4421. Check Temporal UI for workflow execution history: http://localhost:8081
4432. View Neo4j graph: http://localhost:7474 (default: neo4j/password)
4443. Check PostgreSQL: `psql -U postgres -h localhost -d code_confluence`
4454. Check logs with trace ID: Search for `trace_id` in logs
446 
447## Critical Implementation Notes
448 
449### Async/Await Best Practices
450 
451- Always use `async with` for database sessions
452- Never use `.result()` on coroutines in sync context
453- Use `asyncio.create_task()` for fire-and-forget background work (e.g., `monitor_workflow()`)
454- ThreadPoolExecutor required for Temporal activities (can't run pure async code)
455 
456### Type Safety
457 
458- Use `type: ignore` for dynamic models from commons (Credentials, Repository, etc.)
459- Enable strict type checking: `pyrefly` with `preset = "strict"`
460- Always define return types for activities (required by Temporal)
461 
462### Neo4j Transactions
463 
464- Always use `session.execute_write()` or `execute_read()` for managed transactions
465- Don't use raw `session.run()` in activities (no retry)
466- Batch operations with UNWIND for performance (see `codebase_parser_*_batch_size`)
467 
468### PostgreSQL Sessions
469 
470- Never commit manually in activity - use context manager
471- Use `session.begin()` explicitly for transaction control
472- Cascade deletes work via SQLAlchemy relationships
473 
474### Performance Optimization
475 
476- File batch processing: `CODEBASE_PARSER_FILE_BATCH_SIZE` (default 1000)
477- Package batch processing: `CODEBASE_PARSER_PACKAGE_BATCH_SIZE` (default 500)
478- Concurrent file parsing: `CODEBASE_PARSER_FILE_PROCESSING_CONCURRENCY` (default 3)
479- Activity concurrency: `TEMPORAL_MAX_CONCURRENT_ACTIVITIES` (default 4)
480 
481## Environment Variables Summary
482 
483### Required at Startup
484- `DB_HOST`, `DB_PORT`, `DB_USER`, `DB_PASSWORD`, `DB_NAME`
485- `NEO4J_HOST`, `NEO4J_PORT`, `NEO4J_USERNAME`, `NEO4J_PASSWORD`
486- `TEMPORAL_SERVER_ADDRESS`
487 
488### Optional with Defaults
489- `REPOSITORIES_BASE_PATH` → `~/.unoplat/repositories`
490- `FRAMEWORK_DEFINITIONS_PATH` → `/framework-definitions`
491- `LOG_LEVEL` → `DEBUG`
492- `ALLOWED_ORIGINS` → `http://localhost:5173` (CORS)
493 
494### Feature Flags
495- `OTEL_EXPORTER_OTLP_ENDPOINT` → Enable OpenTelemetry export
496- `LOAD_FRAMEWORK_DEFINITIONS` → Load framework definitions at startup (default: true)
497- `FRAMEWORK_DEFINITIONS_REQUIRED` → Fail startup if loading fails (default: false)
498- `DB_ECHO` → Log all SQL (default: false)
499 
500## Integration with Commons
501 
502The project uses `unoplat-code-confluence-commons` for shared models:
503- `Credentials` - Encrypted token storage
504- `Repository`, `CodebaseConfig` - Repository metadata
505- `RepositoryWorkflowRun`, `CodebaseWorkflowRun` - Workflow execution records
506- `ProgrammingLanguageMetadata` - Language-specific config
507- `Flag` - Feature flags
508 
509Reference via absolute imports: `from unoplat_code_confluence_commons.base_models import ...`
510 
511---
512 
513**Last Updated**: 2025-11-14
514**Python Version**: 3.13+
515**Package Manager**: uv
516 
517 
518<CRITICAL_INSTRUCTION>
519 
520## Backlog Workflow
521 
522This project uses Backlog.md MCP for all task and project management. **Before creating tasks or tracking work, read [`backlog_instructions.md`](./backlog_instructions.md)** for the complete workflow guidance.
523 
524</CRITICAL_INSTRUCTION>
525