Orchestrata Ultimate › Plans › archive › semantic_graph_qa_results
Semantic Graph Q&A Results
Source: plans/archive/semantic_graph_qa_results.md at 8a2cb42ad · last changed 2026-08-19 · first committed 2026-07-08 · 703 lines.
Prose pending.
Plan text
Semantic Graph Q&A Results
Model: mistralai/codestral-embed-2505 | Reranker: cohere/rerank-4-pro | Project: orchestrata-ultimate
Q1: Why must the Gateway never serve HTML, JS, or CSS?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/http_surface.py:1-50— """Gateway HTTP surface —/health,/ready,/metrics. The gateway is a WebSocket server; SPEC-014 FR-021 additionally requires a small HTTP surface for liveness/readiness probes (k8s/load-balancer friendly) and Prometheus metrics scraping. This module builds that FastAPI app and serves it alongside the WS server. It reuses the existingpostman.healthHealthChecker (real Redis/Ne.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/cli.py:46-95— Returns: Configured ArgumentParser with serve and version subcommands. """ parser = argparse.ArgumentParser( prog="orchestrata-gateway", description="SPEC-014 WebSocket Gateway", ) sub = parser.add_subparsers(dest="command") sub.add_parser("serve", help="Start the gateway") sub.add_parser("version", help="Show version") return parser .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/http_surface.py:46-84— async def health() -> dict[str, Any]: return {"status": "alive"} @app.get("/ready") @traces(spec="014", story="US-009", reqs=["REQ-014-140"]) async def ready(response: Response) -> dict[str, Any]: # Gateway-level readiness first — if the WS server has not started, # report not-ready without touching infrastructure. if readiness_probe is not None and ...
Q2: How does VizInt UI communicate with the Gateway if they run on different ports?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/orchestrator/pipeline_kernel.py:1-50— """Pipeline Kernel Client — bridges Gateway → PipelineRunner. Implements the samesend_packet(CXPPacket)interface asGRPCClientso the gateway can dispatch intents from the VizInt UI to the local PipelineRunner instead of requiring an out-of-process gRPC kernel. When the gateway is running without a gRPC endpoint, it falls back to this in-process client which runs the full agent pipe.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/handlers/ultimatetui.py:1-50— """UltimateTUI gateway handler — bridges VizInt UI to CheetahClaws agents. Exposes CC slash commands as JSON-RPC 2.0 endpoints so the VizInt browser can callUltimateTUI.handle_edit(),handle_intent(), etc. through the WebSocket gateway. RPC Methods: tui.edit { file_path, old_string, new_string, strict? } → EditResult tui.intent { text } .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/server.py:721-763— @traces(spec="014", story="US-001", reqs=["REQ-014-136", "REQ-004-US-003-022"]) async def push_to_client( self, connection_id: str, method: str, params: dict[str, Any], ) -> bool: """Send a JSON-RPC notification to a specific client. Args: connection_id: Target client connection ID. method: Notification method name....
Q3: What is the exact port and protocol for each service: VizInt UI, Gateway, Kernel, Redis, Neo4j?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/config.py:91-140— All connection parameters read from environment variables with sensible localhost defaults. """ # Optional explicit environment mapping for testing. # When provided, post_init overrides all fields from this dict. environ: dict[str, str] | None = None # Server host: str = field( default_factory=lambda: os.environ.get("GATEWAY_HOST", "localhost") ) .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/infra_config.py:46-95— class InfraConfig(BaseSettings): """Global infrastructure configuration. All fields load from environment variables with sensible defaults. """ model_config = SettingsConfigDict( env_prefix="ORCHESTRATA_", env_file=".env", env_file_encoding="utf-8", extra="ignore", ) # ------------------------------------------------------------------ .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/server.py:226-275— self._kernel_client: Any = None self._reply_task: Any = None # Pipeline-event forwarder task (kernel PipelineEventBus → UI topics). self._pipeline_task: Any = None # Gateway-owned DI container + the singleton memory engine it hosts # (wired in start()). One engine is shared by the memory RPC handler and # the session-close cold-migration path (SPE...
Q4: Why is the gRPC kernel required to be out-of-process from the Gateway?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/orchestrator/pipeline_kernel.py:1-50— """Pipeline Kernel Client — bridges Gateway → PipelineRunner. Implements the samesend_packet(CXPPacket)interface asGRPCClientso the gateway can dispatch intents from the VizInt UI to the local PipelineRunner instead of requiring an out-of-process gRPC kernel. When the gateway is running without a gRPC endpoint, it falls back to this in-process client which runs the full agent pipe.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/server.py:316-365— return PolicyEngine(prolog_client=GovernancePrologClient(bridge)) self._rpc.set_policy_engine_factory(_make_policy_engine) logger.info("Governance PolicyEngine factory registered") # Admin token gating volk.register/deregister (unset -> admin ops denied). self._rpc.set_admin_token(os.environ.get("GATEWAY_ADMIN_TOKEN")) # Wire the live CXP volk kernel (ou.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/kernel_handler.py:1-50— """Packet handler that drives the volk engine for the live CXP gRPC kernel.agents/runtime_main.pystarts a :class:CXPServerfor the out-of-process kernel (theagent-runtimeservice). This module builds thepacket_handlerthat the server invokes for each received packet: it boots a :class:~orchestrata_ultimate.kernel.pipeline.PipelineRunner(all 9 volks on a realPostman) *...
Q5: What is the fallback when the gRPC kernel is unreachable?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/server.py:361-410— # in-process PipelineKernelClient below instead of accepting # intents that would only fail at send_packet time. ready_timeout = float( os.environ.get("GATEWAY_GRPC_READY_TIMEOUT_S", "3.0") ) await asyncio.wait_for( kernel_client.health_check(), timeout=ready_timeout .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/orchestrator/pipeline_kernel.py:1-50— """Pipeline Kernel Client — bridges Gateway → PipelineRunner. Implements the samesend_packet(CXPPacket)interface asGRPCClientso the gateway can dispatch intents from the VizInt UI to the local PipelineRunner instead of requiring an out-of-process gRPC kernel. When the gateway is running without a gRPC endpoint, it falls back to this in-process client which runs the full agent pipe.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/server.py:316-365— return PolicyEngine(prolog_client=GovernancePrologClient(bridge)) self._rpc.set_policy_engine_factory(_make_policy_engine) logger.info("Governance PolicyEngine factory registered") # Admin token gating volk.register/deregister (unset -> admin ops denied). self._rpc.set_admin_token(os.environ.get("GATEWAY_ADMIN_TOKEN")) # Wire the live CXP volk kernel (ou...
Q6: Why are hardcoded service IPs and ports forbidden in this codebase?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/endpoint_config.py:1-50— """Single source of truth for CXP gRPC endpoint configuration. Ports are read from the environment (ultimately from.env) at call time, with no hardcoded numeric default anywhere — a missing port raises rather than silently falling back, so the port lives in exactly one place and cannot drift between the kernel, the dispatcher, the harness, and the deployment manifests. Host values ar.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/runtime_main.py:1-50— """Agent Runtime gRPC server entry point. Provides a gRPC-based agent runtime service using the CXP protocol. Used by Docker container orchestrata-agent-runtime. """ import asyncio import logging import signal import sys from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.cxp.endpoint_config import cxp_grpc_port from orchestrata_ultimate.cxp.grpc_server import .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/config.py:91-140— All connection parameters read from environment variables with sensible localhost defaults. """ # Optional explicit environment mapping for testing. # When provided, post_init overrides all fields from this dict. environ: dict[str, str] | None = None # Server host: str = field( default_factory=lambda: os.environ.get("GATEWAY_HOST", "localhost") ) ...
Q7: How does the project enforce the rule against except: pass and except Exception: pass?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/codedirector/review_checks.py:676-725— (r"AsyncMock\s*(", "AsyncMock() instantiation"), (r"@\s*(mock.)?patch\s*(", "@patch decorator"), (r"mocker.(patch|spy|stub|mock)\s*(", "pytest-mock usage"), (r"create_autospec\s*(", "create_autospec()"), (r"patch.object\s*(", "patch.object()"), (r"patch.dict\s*(", "patch.dict()"), (r"patch.multiple\s*(", "patch.multiple()"), ( r"class\s+(Mock|Fake.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/codedirector/review_checks.py:1486-1535— "invalid Python, and rewrite it as valid Python code. " "Common causes: prose text instead of code, unclosed " "brackets, invalid indentation." ), "LINT": ( "FIX: Run the linter message through the code. Fix the " "exact line and column referenced. Common fixes: remove " "unused imports, fix undefined names, add missing colons." ), "IMPORT.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/volk.py:46-95— ISSUES: -HANDOFF: On PASS, results route to U:Cp (compliance_check); on FAIL, issues route back to U:Cd (fix_issues). RULES: - pytest only (no unittest) - ZERO mocking: no unittest.mock, MagicMock, patch, AsyncMock, Mock, no custom Mock/Fake/Stub/Dummy classes. Test REAL behavior only. - Python 3.14+: never use asyncio.Awaitable/asyncio.Coroutine — use collections.abc.Awai...
Q8: What is the three-tier gRPC logging policy for the Gateway?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/server.py:316-365— return PolicyEngine(prolog_client=GovernancePrologClient(bridge)) self._rpc.set_policy_engine_factory(_make_policy_engine) logger.info("Governance PolicyEngine factory registered") # Admin token gating volk.register/deregister (unset -> admin ops denied). self._rpc.set_admin_token(os.environ.get("GATEWAY_ADMIN_TOKEN")) # Wire the live CXP volk kernel (ou.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/audit/logger.py:316-365— correlation_id=correlation_id, outcome=AuditOutcome.SUCCESS, data={ "message_id": message_id, "from_volk": from_volk, "to_volk": to_volk, "route_decision": route_decision, }, duration_ms=duration_ms, ) await self.log(entry) @traces(spec="001", story="US-008", reqs.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/cli.py:46-95— Returns: Configured ArgumentParser with serve and version subcommands. """ parser = argparse.ArgumentParser( prog="orchestrata-gateway", description="SPEC-014 WebSocket Gateway", ) sub = parser.add_subparsers(dest="command") sub.add_parser("serve", help="Start the gateway") sub.add_parser("version", help="Show version") return parser ...
Q9: How does the Gateway health check authenticate instead of bypassing auth?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/handlers/auth.py:1-50— """Auth RPC handler -- authenticate. Implements SPEC-014 auth.authenticate method. """ from typing import Any from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.gateway.auth import TokenAuthenticator from orchestrata_ultimate.gateway.config import GatewayConfig from orchestrata_ultimate.gateway.exceptions import AuthenticationError from orchestrata_ultimate.ga.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/auth_handler.py:46-95— @traces( spec="014", story="US-006", reqs=["REQ-014-135", "REQ-004-SUCC-002"], ) async def authenticate( websocket: Any, conn_id: str, config: GatewayConfig, auth: TokenAuthenticator, connections: Any, ws_map: dict[str, Any], handle_jsonrpc_fn: Any, ) -> str | None: """Wait for an AUTH frame and validate the token. Supports both internal WSMessag.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/handlers/auth.py:46-76— Returns: Authentication result with scope. """ token = (params or {}).get("token", "") if token: try: claims = self._auth.validate_token(token) except AuthenticationError as exc: return {"authenticated": False, "error": str(exc)} client_id = str(claims["sub"]) elif self.config.allow...
Q10: What is the PID file protocol for long-running services?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/plugin/interfaces.py:316-365— Called after all plugins have been initialized. Begin background tasks or long-running operations here. """ ... @traces(spec="SPEC-015", story="US-015-001", reqs=["REQ-015-030", "REQ-031-044"]) async def stop(self) -> None: """Stop the plugin. Called on system shutdown or plugin disable. Clean up resources, cancel tasks, close connection.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/cli.py:1-50— # mypy: ignore-errors """Orchestrata Kernel CLI — standalone kernel process. Commands: start Boot the kernel component status Show kernel status Author: Mohamed """ # Spec: 001 - Orchestration Kernel import asyncio import logging import os import signal from pathlib import Path from typing import Any import click import yaml from orchestrata_ultimate.core.base.traces import traces .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/protocols/base.py:1-50— """ Base component protocol types for Orchestrata Ultimate. Foundational types and protocols shared across all core protocol modules. This module has no intra-package or cross-package dependencies — it is the canonical source for these base types; the legacy interfaces-facade equivalent module re-exports them instead of redefining them. """ import uuid from collections.abc import Awaitable, C...
Q11: Why must readiness wait for an application-level health check rather than just TCP accept?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/http_surface.py:46-84— async def health() -> dict[str, Any]: return {"status": "alive"} @app.get("/ready") @traces(spec="014", story="US-009", reqs=["REQ-014-140"]) async def ready(response: Response) -> dict[str, Any]: # Gateway-level readiness first — if the WS server has not started, # report not-ready without touching infrastructure. if readiness_probe is not None and .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/http_surface.py:1-50— """Gateway HTTP surface —/health,/ready,/metrics. The gateway is a WebSocket server; SPEC-014 FR-021 additionally requires a small HTTP surface for liveness/readiness probes (k8s/load-balancer friendly) and Prometheus metrics scraping. This module builds that FastAPI app and serves it alongside the WS server. It reuses the existingpostman.healthHealthChecker (real Redis/Ne.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/server.py:361-410— # in-process PipelineKernelClient below instead of accepting # intents that would only fail at send_packet time. ready_timeout = float( os.environ.get("GATEWAY_GRPC_READY_TIMEOUT_S", "3.0") ) await asyncio.wait_for( kernel_client.health_check(), timeout=ready_timeout ...
Q12: What is the graceful shutdown sequence for services?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/server.py:451-500— "Gateway HTTP surface on http://%s:%d (/health /ready /metrics)", self._config.host, self._config.http_port, ) @traces(spec="014", story="US-009", reqs=["REQ-014-092", "REQ-004-US-003-018"]) async def stop(self) -> None: """Gracefully shut down the gateway.""" if self._server: self._server.close() await .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/cli_main.py:226-275— except ImportError as e: logger.warning("Agent runtime not available: %s", e) except Exception as e: logger.warning("Agent runtime init failed: %s", e) def _boot_start_services(config: dict[str, Any], ready_event: threading.Event) -> None: api_cfg = config.get("infrastructure", {}).get("api", {}) health_host = api_cfg.get("host", "0.0.0.0") health_port = api_cf.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/di/container.py:586-635— ): dependencies.append(param_type.name) return dependencies @traces(spec="024", story="Introspection", reqs=["REQ-024-049", "REQ-003-US-008-006"]) def set_context(self, key: str, value: Any) -> None: """ Set context value for dependency resolution. Args: key: Context key value: Context value """ ...
Q13: How is idempotent start implemented for test services?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/isolation.py:136-185— path is that host directory: the caller stages files there (visible inside the container) and drives the container withincus exec. Theoh-trial-prefix guarantees trial containers never collide with standing infrastructure. """ _NAME_PREFIX = "oh-trial-" _MOUNT_POINT = "/workspace" async def create(self, test_id: str) -> Path: if shutil.which("incus") .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/handlers/test_metrics.py:46-95— """ if self.runner and self.runner.is_running: return {"error": "Test run already in progress"} run_id = f"run{datetime.now(UTC).strftime('%Y%m%d%H%M%S')}" if self._runner: # Start the runner in a background task asyncio.create_task( self._runner.run( run_id=run_id, test_path=.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/handlers/test_metrics.py:1-50— """Test execution and metrics RPC handlers.""" import asyncio import logging from datetime import UTC, datetime from typing import Any from orchestrata_ultimate.core.base.traces import traces logger = logging.getLogger(name) @traces(spec="016", story="US-007", reqs=["REQ-016-167", "REQ-016-168"]) class TestMetricsHandler: """Handles test execution and quality metrics RPCs.""" ...
Q14: What action is taken when a port is in use by a foreign process?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/governance/resource_access.py:136-185— @traces( spec="007", story="US-PGOV-004", reqs=["REQ-007-FR4-010", "REQ-007-US-004-002", "REQ-004-US-005-010"], ) async def check_network_access( self, actor: str, domain: str, port: int ) -> AccessResult: """Check if actor can access a network domain:port.""" try: allowed = self._prolog_client.bridge.query_bool( .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/runtime_main.py:1-50— """Agent Runtime gRPC server entry point. Provides a gRPC-based agent runtime service using the CXP protocol. Used by Docker container orchestrata-agent-runtime. """ import asyncio import logging import signal import sys from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.cxp.endpoint_config import cxp_grpc_port from orchestrata_ultimate.cxp.grpc_server import .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/dispatcher.py:316-365— successfully constructed. Never raises and never influences the dispatch outcome -- this is pure observation while the governance layer is staged for real traffic (2026-07-06). Any evaluator failure is caught and logged at WARNING so audit mode can never break dispatch. Args: packet: CXP packet about to be dispatched. recipient: R...
Q15: Why must subprocess.DEVNULL never be used for service logs?
Category: Architecture & Design
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/audit/logger.py:46-95— "REQ-001-US-005-028", "REQ-001-US-006-020", "REQ-001-US-006-024", "REQ-001-US-006-027", "REQ-001-US-006-028", "REQ-001-US-006-029", "REQ-001-US-006-030", "REQ-001-US-007-008", "REQ-001-US-007-018", "REQ-001-US-007-026", "REQ-001-US-007-030", "REQ-008-001", "REQ-008-002", "REQ-008-003", "REQ-008-004", "REQ-008-005", "REQ-008-006", "REQ-0.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/cli.py:46-95— Returns: Configured ArgumentParser with serve and version subcommands. """ parser = argparse.ArgumentParser( prog="orchestrata-gateway", description="SPEC-014 WebSocket Gateway", ) sub = parser.add_subparsers(dest="command") sub.add_parser("serve", help="Start the gateway") sub.add_parser("version", help="Show version") return parser .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/threading/test_executor.py:1-50— # mypy: ignore-errors """Parallel test executor using ThreadPoolManager + batched subprocess. Design (per architecture/THREADPOOL-PARALLEL-TESTING.md): The ThreadPoolExecutor is the thread pool (Section 2.1). Each thread dispatches ONE pytest subprocess that runs a batch of modules. Instead of one subprocess per test MODULE (978 subprocesses, ~24 min), modules are batched into ``ma...
Q16: What does SessionController.begin() do automatically regarding unclosed sessions?
Category: Session Lifecycle & CLI
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/hooks/session_auto_start_hook.py:451-500— return # Only auto-start on fresh startup, not resume/clear/compact if source != "startup": # For resume, just show current session info print("## Session Resumed") print(f"Source: {source}") return # Initialize controller try: controller = SessionController(project_path) except Exception as e: print("## Session Controller Er.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/controller/_core.py:226-275— "REQ-021-060", "REQ-021-061", "REQ-021-062", "REQ-021-063", "REQ-021-064", "REQ-021-065", "REQ-021-066", "REQ-021-067", "REQ-021-068", "REQ-021-069", "REQ-021-070", ], ) def begin( self, purpose: str, objectives: list[str] | None = None,.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/event_emitting_session_controller.py:91-140— objectives: Optional list of goals. Returns: Created SessionRecord. """ result = self._session_controller.begin(purpose=purpose, objectives=objectives) if self._event_bus is not None: _fire_and_forget( self._event_bus, SystemEvent( type="session.opened", source="sessi...
Q17: What are the five checklist items validated by SessionController.validate_and_end()?
Category: Session Lifecycle & CLI
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/controller/_core.py:541-590— try: ctx = self.ctx.read() except FileNotFoundError: return None latest = ctx.latest_session() if latest and latest.length is None: # Session was never closed properly latest.length = "auto-closed" latest.achievements.append("(auto-closed by new session)") self.ctx.update_session( la.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/hooks/session_pre_close_hook.py:1-50— # mypy: ignore-errors #!/usr/bin/env python3 """PreToolUse hook that validates session checklist before close. This hook runs before session close operations and: 1. Validates all checklist items are complete 2. Blocks close if any required item fails 3. Returns JSON with continue: false if validation fails Installation: Add to ~/.claude/settings.json: { "hooks": { "PreT.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/controller/_core.py:586-612— @traces(spec="SPEC-100", story="US-100-001", reqs=["REQ-100-001"]) def validate_and_end( self, achievements: list[str] | None = None, upcoming: list[str] | None = None, ) -> SessionRecord: """End session with strict checklist validation. Args: achievements: What was completed. upcoming: Next steps. Returns: ...
Q18: How does SessionController.end() handle git commits and tags?
Category: Session Lifecycle & CLI
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/controller/_git.py:1-50— # mypy: ignore-errors """GitOpsMixin — git commit/push/tag operations for SessionController.""" import logging import os import subprocess from pathlib import Path from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.session.models import SessionRecord logger = logging.getLogger(name) def _git_env() -> dict[str, str]: """Return the current process envi.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/controller/_core.py:316-365— """End current session (thread-safe). Args: achievements: What was completed. upcoming: Next steps. Returns: Ended session or None. """ with self._lock: if self._active is None: return None if achievements: self._active.achievements = achievements if upcomin.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/cli/lifecycle.py:181-230— ) @click.option( "--interactive", "-i", is_flag=True, help="Prompt for missing fields interactively." ) @click.option( "--smart", is_flag=True, help="Auto-detect artifacts, tests, and git status." ) @click.option( "--commit", is_flag=True, help="Commit all code changes before ending session." ) @click.option( "--validate/--no-validate", default=True, help="Run strict che...
Q19: What is the difference between end() and validate_and_end()?
Category: Session Lifecycle & CLI
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/controller/_core.py:586-612— @traces(spec="SPEC-100", story="US-100-001", reqs=["REQ-100-001"]) def validate_and_end( self, achievements: list[str] | None = None, upcoming: list[str] | None = None, ) -> SessionRecord: """End session with strict checklist validation. Args: achievements: What was completed. upcoming: Next steps. Returns: .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/cli/lifecycle.py:226-275— ach, up, commit = _run_end_interactive( ctrl, achievements, upcoming_items, commit ) if commit: _commit_code_changes(root) session = _finalize_session(ctrl, list(ach), list(up), validate) _report_ended_session(session) def _run_end_interactive( ctrl: SessionController, achievements: Sequence[str], upcoming_items: Sequence[str], commit:.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/cli/lifecycle.py:181-230— ) @click.option( "--interactive", "-i", is_flag=True, help="Prompt for missing fields interactively." ) @click.option( "--smart", is_flag=True, help="Auto-detect artifacts, tests, and git status." ) @click.option( "--commit", is_flag=True, help="Commit all code changes before ending session." ) @click.option( "--validate/--no-validate", default=True, help="Run strict che...
Q20: What shell command starts a session?
Category: Session Lifecycle & CLI
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/cli/lifecycle.py:1-50— # mypy: ignore-errors """Session lifecycle commands (begin, work, end, resume, summary) for the CLI.""" from future import annotations import sys from collections.abc import Sequence from pathlib import Path from typing import TYPE_CHECKING import click from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.session.models import default_ai_model if TYPE_CHEC.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/cli/lifecycle.py:46-95— for obj in session.objectives: click.echo(f" - {obj}") @click.command() @click.argument("desc") @click.option( "--detail", "details", multiple=True, help="Detail bullet (can be repeated)." ) @click.pass_context @traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"]) def work(ctx: click.Context, desc: str, details: tuple[str, ...]) -> None: """Log work in the curren.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/cli_main.py:631-680— @cli.group() @traces(spec="000", story="US-009", reqs=["REQ-000-073", "REQ-006-FR7-006"]) def knowledge() -> None: """Query the knowledge graph.""" @knowledge.command() @click.argument("query") @traces( spec="000", story="US-009", reqs=[ "REQ-000-073", "REQ-000-074", "REQ-000-075", "REQ-000-076", "REQ-006-FR8-001", ], ) def query(que...
Q21: How do you set handoff data for the next session via the CLI?
Category: Session Lifecycle & CLI
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/cli/lifecycle.py:136-185— @click.pass_context @traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"]) def handoff( ctx: click.Context, decisions: tuple[str, ...], blockers: tuple[str, ...], active_work: tuple[str, ...], key_files: tuple[str, ...], cmds: tuple[str, ...], alerts: tuple[str, ...], ) -> None: """Set handoff data for the next session.""" root = ctx.obj["root"] .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/controller/_core.py:451-500— key_files: list[str] | None = None, cmds: list[str] | None = None, alerts: list[str] | None = None, ) -> None: """Set handoff data (thread-safe). Args: decisions: Key decisions. blockers: Current blockers. active_work: In-progress work. key_files: Important files. cmds: Useful commands. .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/controller/_core.py:406-455— filepath=filepath, line_count=line_count, summary=summary, ) self._active.artifacts.append(artifact) self.ctx.update_session(self._active.seq, artifacts=self._active.artifacts) @traces(spec="SPEC-100", story="US-100-001", reqs=["REQ-100-001"]) def log_choice(self, choice: str, reason: str) -> None: """Log techn...
Q22: What happens to old sessions in context.md during end()?
Category: Session Lifecycle & CLI
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/context.py:451-500— Args: text: Item text to match. Returns: True if found and completed. """ with self._transaction(): ctx = self.read() for item in ctx.deferred_work: if item.text == text: item.mark_done() self.write(ctx) return True return False @t.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/context.py:316-365— def _write_json(self, ctx: ContextData) -> None: """Write context to JSON: validate, back up, then write atomically.""" payload = ctx.model_dump_json(indent=2) json.loads(payload) # guard: never persist anything unparseable if self.json_path.exists(): try: current = self.json_path.read_text() json.loads(current) # on.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/archive.py:91-140— sessions: Sessions to archive. """ if not sessions: return tpl = self.jinja.get_template("session.md.j2") new_content = "" for session in sessions: new_content += tpl.render(session=session) + "\n\n" if self.history_exists(): existing = self.history_path.read_text() header = "# Session History Arc...
Q23: Where does the session CHANGELOG get updated?
Category: Session Lifecycle & CLI
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/controller/_changelog.py:46-95— "Fix": "Fixed", "Fixed": "Fixed", "Refactor": "Changed", "Perf": "Changed", "Performance": "Changed", } @traces(spec="SPEC-100", story="US-100-001", reqs=["REQ-100-001"]) def _update_changelog(self, session: SessionRecord) -> bool: """Update CHANGELOG.md with session achievements. Categorizes entries by keyword prefix: - Add/.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/controller/_changelog.py:91-140— Args: changelog: Path to CHANGELOG.md. seq: Session sequence number for commit message. """ subprocess.run( ["git", "add", str(changelog)], cwd=self.root, capture_output=True, ) subprocess.run( ["git", "commit", "-m", f"Update CHANGELOG.md for session {seq:03d}"], cwd=self.root, .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/controller/_changelog.py:1-50— # mypy: ignore-errors """ChangelogMixin — CHANGELOG.md categorization and splicing for SessionController.""" import logging import subprocess from pathlib import Path from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.session.models import SessionRecord logger = logging.getLogger(name) class ChangelogMixin: """CHANGELOG.md update, categorization, and...
Q24: What is the session git tag format?
Category: Session Lifecycle & CLI
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/controller/_git.py:91-140— except FileNotFoundError: logger.debug("git not found; skipping auto-commit/push") def _current_branch(self) -> str | None: """Return the current git branch name, or None on failure.""" try: result = subprocess.run( ["git", "rev-parse", "--abbrev-ref", "HEAD"], cwd=self.root, capture_output=True, .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/controller/_git.py:136-173— """Push session tag to origin. Args: session: The completed session. Returns: True if push succeeded, False otherwise. """ tag_name = f"session-{session.seq:03d}-{session.when.strftime('%Y-%m-%d-%H%M')}" try: result = subprocess.run( ["git", "push", "origin", tag_name], cwd=self.root, .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/context.py:496-545— if self.history_path.exists(): existing = self.history_path.read_text() self._atomic_write_text( self.history_path, existing + "\n\n" + archive_text ) else: self._atomic_write_text(self.history_path, archive_text) @traces(spec="SPEC-100", story="US-100-001", reqs=["REQ-100-001"]) def _get_ma...
Q25: Why is it mandatory to log work, artifacts, choices, and handoff before ending?
Category: Session Lifecycle & CLI
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/hooks/session_pre_close_hook.py:91-140— session = controller.active_session if session is None: failures.append("No active session") return failures # Check 1: Work logged (log not empty) if not session.log: failures.append("No work logged (session.log is empty)") # Check 2: Artifacts recorded (artifacts not empty) if not session.artifacts: failures.append("No artifacts recorded .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/cli/lifecycle.py:91-140— session = _get_active_session(ctrl) if session is None: click.echo("No active session. Start one with 'session begin'.", err=True) sys.exit(1) ctrl.log_artifact(filepath=filepath, line_count=lines, summary=summary) click.echo(f"Artifact logged: {filepath}") @click.command() @click.argument("choice") @click.option("--reason", required=True, help="Rationale for the .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/controller/_core.py:406-455— filepath=filepath, line_count=line_count, summary=summary, ) self._active.artifacts.append(artifact) self.ctx.update_session(self._active.seq, artifacts=self._active.artifacts) @traces(spec="SPEC-100", story="US-100-001", reqs=["REQ-100-001"]) def log_choice(self, choice: str, reason: str) -> None: """Log techn...
Q26: What is the repository pattern abstraction used to hide Bolt details?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/repository.py:1-48— """Repository factory for semantic graph backends. Provides a single entry point for creating aSemanticGraphadapter based on configuration. The factory hides backend-specific instantiation so callers do not import Bolt drivers or adapter classes directly. """ from future import annotations from orchestrata_ultimate.semantic_graph.adapters.memgraph import ( MemgraphSemanticGraph.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/postman/receipt_store.py:91-140— class ReceiptStore(IReceiptStore): """Delivery receipt store.""" @traces(spec="SPEC-002", story="US-002-001", reqs=["REQ-001-US-002-001"]) def init( self, sqlite_path: str = "./data/receipts.db", redis_host: str = os.environ.get("REDIS_HOST", "localhost"), redis_port: int = 6379, neo4j_uri: str = os.environ.get("NEO4J_URI", "bolt://localhost:768.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/base/repository.py:1-50— # mypy: ignore-errors """BaseRepository — async CRUD interface for entity persistence.""" from future import annotations import threading from typing import Protocol, TypeVar from orchestrata_ultimate.core.base.entity import BaseEntity from orchestrata_ultimate.core.base.traces import traces T = TypeVar("T", bound=BaseEntity) @traces(spec="001", story="US-001", reqs=["REQ-001-US-001-052"...
Q27: What are the two embedding backends in semantic_graph/embedders.py?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/embedders.py:1-50— """Embedding backends for the semantic graph. Provides a small protocol plus two concrete implementations: * :class:OpenRouterEmbedder- cloud embeddings via OpenRouter (Codestral). * :class:SentenceTransformerEmbedder- local GPU embeddings viasentence-transformers(Qwen3-Embedding, etc.). Both expose the same async interface so ingestion code does not need to know which backend .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/embedders.py:136-185— logger.error("OpenRouter embeddings request failed: %s", body) raise VectorSearchError( f"OpenRouter embeddings HTTP error {exc.response.status_code if exc.response else 'unknown'}" ) from exc except httpx.RequestError as exc: logger.error("OpenRouter embeddings request failed: %s", exc) raise VectorSearchError(f"OpenRouter.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/repository.py:1-48— """Repository factory for semantic graph backends. Provides a single entry point for creating aSemanticGraphadapter based on configuration. The factory hides backend-specific instantiation so callers do not import Bolt drivers or adapter classes directly. """ from future import annotations from orchestrata_ultimate.semantic_graph.adapters.memgraph import ( MemgraphSemanticGraph...
Q28: Why does the Memgraph adapter use chunk-label and vector-index isolation in tests?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:1-50— """Memgraph-backed semantic graph adapter. Uses the official Neo4j Bolt driver because Memgraph speaks the Bolt protocol. The synchronous driver is used internally and wrapped in threads to preserve the async repository interface; this avoids event-loop binding issues in tests. Vector search uses Memgraph's native HNSW vector index via thevector_search.searchquery module. The index is c.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:451-500— for r in rels: relationships[r.id] = r return GraphRAGContext( entities=list(entities.values()), relationships=list(relationships.values()), chunks=chunks, ) async def _all_entities(self) -> list[Entity]: """Return all entities in the graph (used by retrieve heuristic).""" def _fetch(driver: A.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/config.py:1-50— """Configuration for semantic graph backends. Values are resolved from environment variables with sensible localhost fallbacks, matching the project's no-hardcoded-IPs rule. """ from future import annotations import os from dataclasses import dataclass, field @dataclass class SemanticGraphConfig: """Connection config for a semantic graph backend.""" backend: str uri: str ...
Q29: What bug was fixed in _all_vector_indexes() regarding label strings?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:316-365— ) with driver.session() as session: session.run(cypher) logger.info("Creating vector index %s with dimension %s", self._config.vector_index_name, dimension) await self._execute_with_reconnect(_create) await self._wait_for_vector_index() self._vector_index_dimension = dimension async def _wait_for_vector_index(self) -> None: .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:271-320— """Return metadata for the configured vector index, or None if absent.""" def _fetch(driver: Any) -> dict[str, Any] | None: with driver.session() as session: result = session.run( "CALL vector_search.show_index_info() YIELD index_name, dimension " "WHERE index_name = $index_name " "RETURN dimension".../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:361-410— "label": record["label"].lstrip(":"), "property": record["property"], "dimension": record["dimension"], } for record in result ] return await self._execute_with_reconnect(_fetch) async def _drop_vector_index_by_name(self, name: str) -> None: """Drop a vector index by its...
Q30: What happens when a dimension mismatch is detected on an existing vector index?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:271-320— """Return metadata for the configured vector index, or None if absent.""" def _fetch(driver: Any) -> dict[str, Any] | None: with driver.session() as session: result = session.run( "CALL vector_search.show_index_info() YIELD index_name, dimension " "WHERE index_name = $index_name " "RETURN dimension".../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:226-275— return (list(entities.values()), list(relationships.values())) return await self._execute_with_reconnect(_fetch) async def add_chunks(self, chunks: list[Chunk]) -> None: if not chunks: return first_embedding = next((c.embedding for c in chunks if c.embedding), None) if first_embedding is not None: await self._ensure_vector_index(len.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:91-140— async with httpx.AsyncClient(timeout=5.0) as client: response = await client.get( f"{self._config.ha_controller_url}/main" ) response.raise_for_status() data = response.json() uri = data.get("uri") if uri: return uri ...
Q31: How does the HA controller discover the current Memgraph MAIN node?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:46-95— self._driver: Any | None = None self._current_uri: str | None = None self._vector_index_dimension: int | None = None async def connect(self) -> bool: """Open the Bolt driver and verify connectivity. If an HA controller URL is configured, discover the current MAIN URI from the controller before connecting. """ uri = await self._resolv.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ha_controller.py:1-50— """High-availability controller for Memgraph Community replication. The controller monitors a set of Memgraph data instances, enforces a single MAIN role, performs automatic failover, and reintegrates recovered nodes as REPLICAs (autofailback). It exposes an HTTP API so clients can discover the current MAIN URI without hardcoding it. This is intentionally a single-controller design. It is the.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ha_controller.py:46-95— healthy: bool = False last_seen: float = 0.0 error: str | None = None @dataclass class ClusterStatus: """Snapshot returned by the HTTP status endpoint.""" current_main: str | None epoch: int nodes: list[NodeState] class MemgraphHAController: """Monitor and manage a Memgraph replication cluster.""" def init( self, nodes: list[NodeConfig],...
Q32: What is the vector index name, dimension, and metric for the orchestrata-ultimate Codestral chunks?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:271-320— """Return metadata for the configured vector index, or None if absent.""" def _fetch(driver: Any) -> dict[str, Any] | None: with driver.session() as session: result = session.run( "CALL vector_search.show_index_info() YIELD index_name, dimension " "WHERE index_name = $index_name " "RETURN dimension".../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:1-50— """Memgraph-backed semantic graph adapter. Uses the official Neo4j Bolt driver because Memgraph speaks the Bolt protocol. The synchronous driver is used internally and wrapped in threads to preserve the async repository interface; this avoids event-loop binding issues in tests. Vector search uses Memgraph's native HNSW vector index via thevector_search.searchquery module. The index is c.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ingestion.py:1-50— """Source-code ingestion into a semantic graph. Chunks source files, embeds the chunks, and stores them as vector-indexed nodes. Each chunk carries aproject_idso one graph can host multiple projects without collision. """ from future import annotations import logging import mimetypes from pathlib import Path from typing import Any from orchestrata_ultimate.semantic_graph.embedders...
Q33: What reranker model is used for the orchestrata-ultimate semantic graph?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/rerankers.py:1-50— """Reranking backend for semantic graph retrieval. Uses OpenRouter's rerank endpoint with Cohere rerankers to reorder candidate chunks by relevance to a query. """ from future import annotations import logging import httpx from orchestrata_ultimate.semantic_graph.exceptions import VectorSearchError from orchestrata_ultimate.semantic_graph.models import Chunk logger = logging.getLogger.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/context_pack/selector.py:1-50— """Deterministic selection — per-task-type templates + graph-signal ranking (SPEC-106 FR-003). One parameterized Cypher template per :class:TaskType(REQ-106-020), generalizing the six SPEC-009inject_for_taskhandlers (_inject_build_context.._inject_review_context) onto the FR-001 canon (REQ-106-025): * containment — the legacy File/Class/Method AST-subtree walk becomes can.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/__init__.py:1-50— """Semantic graph repository abstraction for RAG / GraphRAG. Example: from orchestrata_ultimate.semantic_graph import MemgraphSemanticGraph graph = MemgraphSemanticGraph() await graph.connect() await graph.upsert_entities([Entity(id="doc-1", labels=["Document"])]) """ from orchestrata_ultimate.semantic_graph.adapters import ( InMemorySemanticGraph, MemgraphSemanticGra...
Q34: How many chunks were ingested for the orchestrata-ultimate project?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ingestion.py:181-230— directory: Root directory to search. pattern: Glob pattern for source files. Returns: Number of chunks stored. """ chunks = self._chunker.chunk_directory(directory, self._project_id, pattern) if not chunks: logger.warning("No chunks found in %s for pattern %s", directory, pattern) return 0 for chunk in chu.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ingestion.py:1-50— """Source-code ingestion into a semantic graph. Chunks source files, embeds the chunks, and stores them as vector-indexed nodes. Each chunk carries aproject_idso one graph can host multiple projects without collision. """ from future import annotations import logging import mimetypes from pathlib import Path from typing import Any from orchestrata_ultimate.semantic_graph.embedders.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/tools/ast_query_cli.py:406-455— print(f" - {c}") if len(orchestrata_only) > 30: print(f" ... and {len(orchestrata_only) - 30} more") print(f"\nUltimate-only (new additions): {len(ultimate_only)}") for c in ultimate_only[:30]: print(f" - {c}") if len(ultimate_only) > 30: print(f" ... and {len(ultimate_only) - 30} more") if json_output: _output(result, True) @app.command...
Q35: What property tags every chunk with its project for multi-project support?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ingestion.py:1-50— """Source-code ingestion into a semantic graph. Chunks source files, embeds the chunks, and stores them as vector-indexed nodes. Each chunk carries aproject_idso one graph can host multiple projects without collision. """ from future import annotations import logging import mimetypes from pathlib import Path from typing import Any from orchestrata_ultimate.semantic_graph.embedders.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/canonical_schema.py:91-140— # | Decision | id | (uncontested) | # | Artifact | id | (uncontested) | CANONICAL_MERGE_KEYS: dict[str, str] = { LABEL_MODULE: "id", LABEL_CLASS: "id", LABEL_FUNCTION: "id", LABEL_SPEC: "id", LABEL_REQUIREMENT: "id", LABEL_REQUIREMENT_ITEM: "id", LABEL_VOLK: "alias", LABEL_SES.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/models.py:1-50— """Data models for semantic graph / RAG / GraphRAG. Lightweight dataclasses used by the repository interfaces. They intentionally avoid upward dependencies on orchestrata_ultimate.knowledge_graph so the semantic_graph package can be reused with different backends. """ from future import annotations from dataclasses import dataclass, field from typing import Any @dataclass class Entity:...
Q36: What is the Embedder protocol interface?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/embedders.py:1-50— """Embedding backends for the semantic graph. Provides a small protocol plus two concrete implementations: * :class:OpenRouterEmbedder- cloud embeddings via OpenRouter (Codestral). * :class:SentenceTransformerEmbedder- local GPU embeddings viasentence-transformers(Qwen3-Embedding, etc.). Both expose the same async interface so ingestion code does not need to know which backend .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/embedders.py:46-95—"document"and"query". Backends that do not use prompt prefixes ignore this value. Returns: A list of embedding vectors parallel totexts. """ ... class OpenRouterEmbedder: """Cloud embedding backend using the OpenRouter embeddings endpoint. Defaults tomistralai/codestral-embed-2505(1536 dimensions). The call.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/interfaces.py:1-50— """Repository interfaces for semantic graph / RAG / GraphRAG. The interfaces are Protocols so adapters can be swapped without inheritance. """ from future import annotations from typing import Protocol, runtime_checkable from orchestrata_ultimate.semantic_graph.models import ( Chunk, Entity, GraphRAGContext, Relationship, ) @runtime_checkable class VectorStore(Protocol...
Q37: How does OpenRouterEmbedder handle out-of-order provider responses?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/embedders.py:136-185— logger.error("OpenRouter embeddings request failed: %s", body) raise VectorSearchError( f"OpenRouter embeddings HTTP error {exc.response.status_code if exc.response else 'unknown'}" ) from exc except httpx.RequestError as exc: logger.error("OpenRouter embeddings request failed: %s", exc) raise VectorSearchError(f"OpenRouter.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/embedders.py:46-95—"document"and"query". Backends that do not use prompt prefixes ignore this value. Returns: A list of embedding vectors parallel totexts. """ ... class OpenRouterEmbedder: """Cloud embedding backend using the OpenRouter embeddings endpoint. Defaults tomistralai/codestral-embed-2505(1536 dimensions). The call.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/embedders.py:91-140— @property def dimension(self) -> int: return self._dimension async def embed( self, texts: list[str], prompt_type: str = "document" ) -> list[list[float]]: """Embed texts via OpenRouter in bounded batches.""" del prompt_type # Codestral does not use prompt prefixes. if not texts: return [] results: list[list[float] | Non...
Q38: Why is SentenceTransformerEmbedder loaded lazily?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/embedders.py:136-185— logger.error("OpenRouter embeddings request failed: %s", body) raise VectorSearchError( f"OpenRouter embeddings HTTP error {exc.response.status_code if exc.response else 'unknown'}" ) from exc except httpx.RequestError as exc: logger.error("OpenRouter embeddings request failed: %s", exc) raise VectorSearchError(f"OpenRouter.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/embedders.py:226-269— model = self._load_model() prompt_kwargs = self._prompt_kwargs_for(prompt_type) def _encode() -> Any: kwargs: dict[str, Any] = { "batch_size": self._batch_size, "normalize_embeddings": self._normalize_embeddings, "show_progress_bar": False, } kwargs.update(prompt_kwargs) return model.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/embedders.py:181-230— self, model_name: str = "Qwen/Qwen3-Embedding-0.6B", dimension: int = 1024, device: str | None = None, query_prompt: str | None = None, document_prompt: str = "", trust_remote_code: bool = True, batch_size: int = 32, normalize_embeddings: bool = True, ) -> None: """Initialize the embedder. Args: mod...
Q39: What is the default Qwen embedding model and dimension?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/embedders.py:181-230— self, model_name: str = "Qwen/Qwen3-Embedding-0.6B", dimension: int = 1024, device: str | None = None, query_prompt: str | None = None, document_prompt: str = "", trust_remote_code: bool = True, batch_size: int = 32, normalize_embeddings: bool = True, ) -> None: """Initialize the embedder. Args: mod.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/embedders.py:136-185— logger.error("OpenRouter embeddings request failed: %s", body) raise VectorSearchError( f"OpenRouter embeddings HTTP error {exc.response.status_code if exc.response else 'unknown'}" ) from exc except httpx.RequestError as exc: logger.error("OpenRouter embeddings request failed: %s", exc) raise VectorSearchError(f"OpenRouter.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/embedders.py:1-50— """Embedding backends for the semantic graph. Provides a small protocol plus two concrete implementations: * :class:OpenRouterEmbedder- cloud embeddings via OpenRouter (Codestral). * :class:SentenceTransformerEmbedder- local GPU embeddings viasentence-transformers(Qwen3-Embedding, etc.). Both expose the same async interface so ingestion code does not need to know which backend ...
Q40: How does the ingestion script verify retrieval quality after indexing?
Category: Semantic Graph & RAG
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/verification/__init__.py:1-1— """Core verification engine and feedback."""/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/orchestration/__init__.py:1-1— """Verification workflow coordination."""/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/config/__init__.py:1-1— """Verification configuration."""
Q41: Why is pytest-xdist banned for this project?
Category: Testing & Quality
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/standards.py:136-185— --- Tests --- 32. Tests present and correct. Clear assertions. No false positives. NO MOCKING — test real behavior only. 32b. TEST CLEANUP: Usetmp_pathpytest fixture for temporary files. 32c. NO MONKEY-PATCHING: Use real error conditions, not method replacement. --- Traceability --- 33. TRACEABILITY: Every class and every public function MUST have a @traces(spec=..., story=..., req.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/threading/test_executor.py:1-50— # mypy: ignore-errors """Parallel test executor using ThreadPoolManager + batched subprocess. Design (per architecture/THREADPOOL-PARALLEL-TESTING.md): The ThreadPoolExecutor is the thread pool (Section 2.1). Each thread dispatches ONE pytest subprocess that runs a batch of modules. Instead of one subprocess per test MODULE (978 subprocesses, ~24 min), modules are batched into ``ma.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/threading/test_executor.py:451-500— class ParallelTestExecutor: """Run test modules in parallel via batched subprocess workers. The ThreadPoolExecutor dispatches subprocesses (one per batch). Each batch runs as a full pytest session with all fixtures intact. """ @traces(spec="000", story="US-000", reqs=["REQ-000-000"]) def init(self, max_workers: int = 4, batch_multiplier: int = 1): # batch_m...
Q42: What are the two supported test runners?
Category: Testing & Quality
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/standards.py:136-185— --- Tests --- 32. Tests present and correct. Clear assertions. No false positives. NO MOCKING — test real behavior only. 32b. TEST CLEANUP: Usetmp_pathpytest fixture for temporary files. 32c. NO MONKEY-PATCHING: Use real error conditions, not method replacement. --- Traceability --- 33. TRACEABILITY: Every class and every public function MUST have a @traces(spec=..., story=..., req.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/test_runner.py:1-50— """Async pytest runner with real-time progress broadcasting.""" import asyncio import json import logging import os import re from datetime import UTC, datetime from typing import Any from orchestrata_ultimate.core.base.traces import traces logger = logging.getLogger(name) # Pytest output line parsers _PYTEST_COLLECTED = re.compile(r"collected\s+(\d+)\s+item") _PYTEST_PASSED = re.compil.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/test_runner.py:91-140— if self._running: return {"error": "Test run already in progress"} self._running = True self._current_run_id = run_id start_time = datetime.now(UTC) cmd = [ "python", "-m", "pytest", "-v", "-s", # unbuffered output for real-time progress "--timeout=120", "--tb=short...
Q43: What is the minimum per-file coverage required by the Definition of Done?
Category: Testing & Quality
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/requirements/constraint_definitions.py:1-50— """Predefined constraint specifications and templates.""" from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.verification.types import Constraint, ConstraintType # Predefined functional constraints FUNCTIONAL_CONSTRAINTS = [ Constraint( id="func-file-exists", type=ConstraintType.FUNCTIONAL, predicate="file_exists", expected=T.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/domain/goal.py:46-94— P2 = "P2" P3 = "P3" @traces(spec="012", story="US-012-Goals", reqs=["REQ-012-038"]) class PortfolioSuccessCriteria(BaseModel): """Completion criteria for a portfolio goal. Attributes: tests_pass: Whether all tests must pass. min_coverage_pct: Minimum code coverage percentage. min_spec_coverage_pct: Minimum spec coverage percentage. max_iterations: .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/pm/goal_report.py:271-320— "", ] _render_criteria_sections(lines, report) _render_gaps_and_recs(lines, report) _render_detail_sections(lines, report) return "\n".join(lines) @traces(spec="012", story="US-012-Reporting", reqs=["REQ-012-043"]) # Test results def _evaluate_tests(report: GoalReport, result: dict[str, Any]) -> None: """Evaluate test pass/fail criterion. Args...
Q44: What does rule T2 forbid regarding os.environ in ported tests?
Category: Testing & Quality
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/review/checks_safety.py:1-50— """Safety checks: security, TODO comments, mocks, lint. Covers checklist items 30, 12b, and mock/lint detection. """ import ast import os import re import subprocess from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.verification.review.types import Issue _MOCK_PATTERNS = [ (r"from\s+unittest.mock\s+import", "unittest.mock import"), (r"import\s+unitt.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/review/checks_safety.py:91-140— continue issues.append(("SECURITY", rel_name, f"L{i}: {desc}.")) break return issues @traces(spec="007b", story="US-003", reqs=["REQ-009-US-003-001"]) def check_env_access( filepath: str, tree: ast.Module, ) -> list[Issue]: """Check 30c: CONFIG -- bareos.environ["X"]subscript reads. Readingos.environ["X"]raisesKeyErrorwhen th.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/review/checks_safety.py:136-185— "CONFIG", rel_name, f"L{node.lineno}: bareos.environ[{target}]raises " f"KeyError when unset -- use " f"os.environ.get({target}, default).", ) ) return issues # Parallel test isolation (standards rule 34). Mirrors the blocking # patterns of scripts/gate_parallel_safety.py so t...
Q45: Why must fixtures be function-scope rather than module or class scope?
Category: Testing & Quality
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/threading/pytest_live_observer.py:46-95— import logging import os import time from pathlib import Path from typing import Any from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.core.patterns.observer import SyncObservable logger = logging.getLogger(name) # Read once at import time: each batch subprocess is a fresh pytest session # (test_executor.py's design -- one subprocess per batch), so module.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/_dep_install.py:271-320— "# G-76-01: expose orchestrata monorepo packages as fallback\n" "# G-90-02: use append so workspace/src shadows monorepo on namespace conflicts\n" "_monorepo = '/opt/orchestrata/packages'\n" "if os.path.isdir(_monorepo):\n" " for _pkg_src in glob.glob(f'{_monorepo}/*/src'):\n" " if _pkg_src not in sys.path:\n" " .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/volk.py:811-860— "- SISMEMBER/EXISTS/DEL return ints (1/0), NOT bools\n" "- Streams: create consumer group before XREADGROUP\n" "- Short delays for retry tests: backoff_factor=0.01\n\n" "### Async Rules\n" "- async source -> async test + @pytest.mark.asyncio + await\n" "- Async fixtures: @pytest_asyncio.fixture (NOT @pytest.fixture)\n" "- S...
Q46: What is the isolation fixture used for?
Category: Testing & Quality
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/isolation.py:1-50— # mypy: ignore-errors import shutil import tempfile import uuid from pathlib import Path from types import TracebackType from typing import Self from orchestrata_ultimate.core.base.traces import traces @traces(spec="000", story="US-000", reqs=["REQ-000-000", "REQ-002-US-004-004"]) class Isolation: """Utility for creating isolated temporary directories and files.""" @traces( spe.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/isolation.py:91-140— capture_output=True, text=True, ) if branch_name in (branch_list.stdout or ""): self._run_git( ["git", "branch", "-D", branch_name], cwd=repo_root, check=False, ) @staticmethod def _git_toplevel() -> Path: result = subprocess.run( ["git", "rev-parse", "--show-toplevel.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/standards.py:136-185— --- Tests --- 32. Tests present and correct. Clear assertions. No false positives. NO MOCKING — test real behavior only. 32b. TEST CLEANUP: Usetmp_pathpytest fixture for temporary files. 32c. NO MONKEY-PATCHING: Use real error conditions, not method replacement. --- Traceability --- 33. TRACEABILITY: Every class and every public function MUST have a @traces(spec=..., story=..., req...
Q47: How are Redis keys isolated per worker in tests?
Category: Testing & Quality
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/review/checks_safety.py:136-185— "CONFIG", rel_name, f"L{node.lineno}: bareos.environ[{target}]raises " f"KeyError when unset -- use " f"os.environ.get({target}, default).", ) ) return issues # Parallel test isolation (standards rule 34). Mirrors the blocking # patterns of scripts/gate_parallel_safety.py so t.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/isolation.py:181-230— self._run_incus(["incus", "delete", container, "--force"], check=False) shutil.rmtree(self._host_dir(test_id), ignore_errors=True) @classmethod def _container_name(cls, test_id: str) -> str: """Derive a valid, collision-safe Incus instance name from test_id. Incus names allow only lowercase letters, digits and hyphens, must start with a letter, and ar.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/volk.py:811-860— "- SISMEMBER/EXISTS/DEL return ints (1/0), NOT bools\n" "- Streams: create consumer group before XREADGROUP\n" "- Short delays for retry tests: backoff_factor=0.01\n\n" "### Async Rules\n" "- async source -> async test + @pytest.mark.asyncio + await\n" "- Async fixtures: @pytest_asyncio.fixture (NOT @pytest.fixture)\n" "- S...
Q48: What is the real-Prolog fixture pattern for test isolation?
Category: Testing & Quality
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/volk.py:811-860— "- SISMEMBER/EXISTS/DEL return ints (1/0), NOT bools\n" "- Streams: create consumer group before XREADGROUP\n" "- Short delays for retry tests: backoff_factor=0.01\n\n" "### Async Rules\n" "- async source -> async test + @pytest.mark.asyncio + await\n" "- Async fixtures: @pytest_asyncio.fixture (NOT @pytest.fixture)\n" "- S.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/governance/prolog_gate.py:136-178— all_passed = False return all_passed, gates except Exception as e: gates.append(f"✓ Prolog error: {e} — permissive fallback") return True, gates @traces( spec="SPEC-007", story="US-007-001", reqs=["REQ-001-US-007-001", "REQ-004-US-006-002"] ) def assert_fact(self, fact: str): """Assert a Prolog fact into the session.""" i.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/orchestrator/prolog_bridge.py:91-140— def init(self, rules_dir: Path | None = None) -> None: """Initialize PrologBridge. Args: rules_dir: Directory containing .pl rule files """ self._worker = get_prolog_worker() self._rules_dir = rules_dir self._loaded_files: set[str] = set() self._facts: dict[str, PrologFact] = {} self._session_facts: dict[str, set[s...
Q49: Why are mocks, stubs, fakes, and monkeypatch forbidden?
Category: Testing & Quality
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/review/checks_safety.py:1-50— """Safety checks: security, TODO comments, mocks, lint. Covers checklist items 30, 12b, and mock/lint detection. """ import ast import os import re import subprocess from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.verification.review.types import Issue _MOCK_PATTERNS = [ (r"from\s+unittest.mock\s+import", "unittest.mock import"), (r"import\s+unitt.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/review/checks_safety.py:226-275— """ issues: list[Issue] = [] rel_name = os.path.basename(filepath) tags = ("TODO", "FIXME", "HACK", "XXX") for i, line in enumerate(lines, 1): stripped = line.strip() if stripped.startswith("#"): upper = stripped.upper() if any(tag in upper for tag in tags): issues.append( ( "TODO.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/codedirector/review_checks.py:676-725— (r"AsyncMock\s*(", "AsyncMock() instantiation"), (r"@\s*(mock.)?patch\s*(", "@patch decorator"), (r"mocker.(patch|spy|stub|mock)\s*(", "pytest-mock usage"), (r"create_autospec\s*(", "create_autospec()"), (r"patch.object\s*(", "patch.object()"), (r"patch.dict\s*(", "patch.dict()"), (r"patch.multiple\s*(", "patch.multiple()"), ( r"class\s+(Mock|Fake...
Q50: What are the four blocking safety gates run before tests?
Category: Testing & Quality
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/volk.py:316-365— # G218: Capture requirement IDs for traceability self._structured_requirements = packet.payload.get( "structured_requirements", [] ) self._user_stories = packet.payload.get("user_stories", []) # Reset cycle counter on new thread self._reset_thread_state(packet.thread_id) # Read the source files file_contents = self._read_sour.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/volk.py:361-410— ( 2, "Lint & style", review_gate_lint, " [U:Qa] GATE 2 FAIL — {n} lint/style errors. STOP. Fix lint first.", "GATE 2 FAIL: Lint/style errors", ), ( 3, "Checklist", review_gate_checklist, " [U:Qa] GATE 3 FAIL — {n} checklist iss.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/standards.py:136-185— --- Tests --- 32. Tests present and correct. Clear assertions. No false positives. NO MOCKING — test real behavior only. 32b. TEST CLEANUP: Usetmp_pathpytest fixture for temporary files. 32c. NO MONKEY-PATCHING: Use real error conditions, not method replacement. --- Traceability --- 33. TRACEABILITY: Every class and every public function MUST have a @traces(spec=..., story=..., req...
Q51: What is the source monorepo path and how many packages does it contain?
Category: Porting & AST
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/architect/architect_context.py:271-320— for _path, (ref, content) in loaded.items() ] return ( "\n\nEXISTING API SOURCES (read these carefully — use the ACTUAL " "class signatures, constructor parameters, and import paths shown " "below. The file path tells you the exact import):\n\n" + "\n\n".join(sections) ) @traces(spec="SPEC-005", story="US-005-001", req.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/_dep_install.py:406-455— capture_output=True, text=True, timeout=120, ) @traces(spec="SPEC-005", story="US-005-001", reqs=["REQ-001-US-005-001"]) def _install_monorepo_deps(self, pip: list[str]) -> None: """G-77-07: Install runtime deps of all monorepo packages into the test venv. When generated code imports from orchestrata_ultimate.kernel.event_bus, the kernel .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/tools/ast_pipeline.py:1-50— #!/usr/bin/env python3 """Automated AST serialization and deep-comparison pipeline. Runs the three-step pipeline that is the source of truth for porting progress: 1. Serialize orchestrata-ultimate (local package) 2. Serialize orchestrata monorepo (frozen source) 3. Deep-compare with semantic class mapping Usage: uv run ast-sync # or python scripts/run_ast_pipeline.py """ from __f...
Q52: What is the target package path for the ported code?
Category: Porting & AST
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/__init__.py:1-6— """ Orchestrata Ultimate — Unified orchestration platform. Ported from 26-package monorepo into single cohesive package. """ version = "1.0.0"/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/_dep_install.py:226-275— # Common import-to-package name mappings pkg_map = { "cryptography": "cryptography", "yaml": "pyyaml", "cv2": "opencv-python", "PIL": "pillow", "bs4": "beautifulsoup4", "attr": "attrs", "dotenv": "python-dotenv", "jwt": "pyjwt", "msgpack": "msgpack", "aiohttp": "aiohtt.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/_autofix.py:721-770— content, ) # Replacefrom orchestrata_ultimate.agents.volks.qa. import Xwithfrom pkg import Xcontent = re.sub( r"from . import", f"from {pkg_name} import", content, ) if content != original: with open(fpath, "w", encoding="utf-8"...
Q53: What is the purpose of scripts/serialize_codebase_ast.py?
Category: Porting & AST
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/tools/ast_pipeline.py:1-50— #!/usr/bin/env python3 """Automated AST serialization and deep-comparison pipeline. Runs the three-step pipeline that is the source of truth for porting progress: 1. Serialize orchestrata-ultimate (local package) 2. Serialize orchestrata monorepo (frozen source) 3. Deep-compare with semantic class mapping Usage: uv run ast-sync # or python scripts/run_ast_pipeline.py """ from __f.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/tools/ast_query.py:586-616— except OSError: return None stale = 0 for py_file in src_root.rglob("*.py"): try: if py_file.stat().st_mtime > snapshot_mtime: stale += 1 except OSError: continue if stale == 0: return None return stale, snapshot_mtime def staleness_warning(self) -> str | None.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/tools/ast_pipeline.py:46-95— PLANS_DIR.mkdir(exist_ok=True) # Step 1 — serialize ultimate run( [ "serialize", "--source", str(SOURCE_ULTIMATE), "--name", "orchestrata-ultimate", "--output", str(PLANS_DIR / "ast-ultimate.json"), ], "Serializing orchestrata-ultimate", ) # Step 2 — serialize orchestrat...
Q54: How many classes are in the orchestrata source AST?
Category: Porting & AST
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/tools/ast_query_cli.py:361-410— ast = _get_ast(ast_path, None) results = ast.exported_classes() for ci in results: print(f" {ci.name:40s} {ci.module_path}") print(f"\n Total: {len(results)} exported declarations") if json_output: _output([ci.dict for ci in results], True) @app.command() def diff( class_name: str = typer.Argument(..., help="Class name to diff"), ast_path: str = .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/class_extractor.py:1-50— # mypy: ignore-errors """Class extraction from Python AST. Extracts class definitions, methods, base classes, decorators, and docstrings from parsed Python source. """ import ast import logging from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.knowledge_graph.types import ClassNode, FunctionNode logger = logging.getLogger(name) @traces(spec="008", stor.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/tools/ast_query.py:1-50— """ AST Query Engine — Programmatic access to serialized Python AST. Loadsplans/ast-ultimate.json(and optionally the orchestrata comparison) and exposes a rich query API: find classes, list stubs, compute coverage, trace inheritance, search methods, and more. Usage:: from orchestrata_ultimate.tools.ast_query import ASTQuery ast = ASTQuery() ast.summary() ast.stubs() ...
Q55: What is the three-pass class pairing algorithm in the deep compare tool?
Category: Porting & AST
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/tools/ast_query.py:451-500— } return result for tier, tier_entry in self._compare_data.get("coverage_by_tier", {}).items(): result[str(tier)] = { "packages": len(tier_entry.get("packages", [])), "orchestrata_loc": tier_entry.get("orch_loc", 0), "ultimate_loc": tier_entry.get("ult_loc", 0), "coverage": tier_entry.get("loc_covera.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/tools/ast_query.py:271-320— oc = pair.get("orchestrata_class") or pair.get("orchestrata") or {} u_methods = uc.get("method_count", uc.get("methods")) if not isinstance(u_methods, int): u_methods = len(u_methods or []) o_methods = oc.get("method_count", oc.get("methods")) if not isinstance(o_methods, int): .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/tools/ast_pipeline.py:46-95— PLANS_DIR.mkdir(exist_ok=True) # Step 1 — serialize ultimate run( [ "serialize", "--source", str(SOURCE_ULTIMATE), "--name", "orchestrata-ultimate", "--output", str(PLANS_DIR / "ast-ultimate.json"), ], "Serializing orchestrata-ultimate", ) # Step 2 — serialize orchestrat...
Q56: What is the readiness classification for ported classes?
Category: Porting & AST
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/tools/ast_query.py:271-320— oc = pair.get("orchestrata_class") or pair.get("orchestrata") or {} u_methods = uc.get("method_count", uc.get("methods")) if not isinstance(u_methods, int): u_methods = len(u_methods or []) o_methods = oc.get("method_count", oc.get("methods")) if not isinstance(o_methods, int): .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/hooks/session_auto_start_hook.py:181-230— total_pairs = summary.get("total_pairs", 0) unmapped_ult = summary.get("unmapped_ultimate", 0) unmapped_orc = summary.get("unmapped_orchestrata", 0) lines: list[str] = [] lines.append("### Porting Progress") lines.append("") lines.append("| Metric | Count |") lines.append("|--------|-------|") lines.append(f"| Semantic pairs | {total_pairs} |") for key, val .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/tools/ast_query.py:226-275— """Return method dicts matching a regex against qualified name.""" self._build() flags = 0 if case_sensitive else re.IGNORECASE rx = re.compile(pattern, flags) return [ m for m in self._method_index.values() if rx.search(m["qualified_name"]) ] def find_module(self, path: str) -> dict | None: """Return module dict by path (e.g. 'ag...
Q57: What is the standard import translation from orchestrata.core.interfaces?
Category: Porting & AST
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/interfaces/components.py:1-46— """ Core component interfaces for Orchestrata. This module is a thin alias layer: the canonical definitions (kernel, postman, orchestrator, and state-manager protocols, plus the forward-reference sentinels used to break circular dependencies) live in :mod:orchestrata_ultimate.core.protocols.componentsand are re-exported here under their historical names to preserve the public API. """ from.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/interfaces/__init__.py:1-50— """ Core component interfaces for Orchestrata. These interfaces define the contract for all core components in the system. All components must implement these interfaces to ensure compatibility. This package is a re-export facade. The definitions live in domain submodules and are re-exported here to preserve the historical public API: * :mod:orchestrata_ultimate.core.interfaces.base— foun.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/interfaces/providers.py:1-50— """ Provider manager interface types for Orchestrata (SPEC-011). Thin compatibility facade. The canonical definitions live in :mod:orchestrata_ultimate.core.protocols.providers(the interfaces->protocols twin-collapse); this module re-exports them under the historicalI-prefixed names so existingfrom core.interfaces.providers import ...imports keep working and resolve to the single...
Q58: What is the standard import translation from orchestrata.volks.<role>?
Category: Porting & AST
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/task_executor.py:46-95— def _task_to_packet(task: TaskDefinition) -> CXPPacket: """Convert a TaskDefinition to a CXPPacket.""" from orchestrata_ultimate.cxp.packet import CXPPacket, WorkStatus metadata = task.parameters or {} return CXPPacket( from_agent=metadata.get("from_agent", "SYSTEM"), to_agent=task.target, thread_id=metadata.get("thread_id", ""), subject=task.act.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/types.py:46-95— # ---------- Volk identity constants ---------- VOLK_ALIASES: dict[str, str] = { "Orchestrator": "U:Ox", "Postman": "U:PS", "CodeDirector": "U:Cd", "UX": "U:Ux", "PM": "U:PM", "Architect": "U:Ar", "QA": "U:Qa", "Concierge": "U:Cx", } VOLK_ROLES: dict[str, str] = { "U:Ox": "Executive", "U:PS": "Router", "U:Cd": "Coder", "U:Ux": "Designer", "U.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/executor.py:136-185— "analyst": ["AnalystAgent"], "architect": ["ArchitectVolk"], "codedirector": ["CodeDirectorVolk", "CodeDirectorAgent"], "compliance": ["ComplianceVolk", "ComplianceAgent"], "concierge": ["ConciergeVolk", "ConciergeAgent"], "junior": ["JuniorVolk", "JuniorAgent"], "pm": ["PMVolk", "ProjectManagerAgent"], "qa": ["...
Q59: How are dynamic string literal imports handled during porting?
Category: Porting & AST
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/_autofix.py:676-725— content, ) # Pattern: port=6379 or 6380 hardcoded (replace with env var) content = re.sub( r"port\s*=\s*(?:6379|6380)\b", "port=int(os.environ['REDIS_PORT'])", content, ) if content != original: # Ensure 'import os' is present if "import os" not in content: content = "import os\n".../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/_autofix.py:721-770— content, ) # Replacefrom orchestrata_ultimate.agents.volks.qa. import Xwithfrom pkg import Xcontent = re.sub( r"from . import", f"from {pkg_name} import", content, ) if content != original: with open(fpath, "w", encoding="utf-8".../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/_testgen.py:226-275— G113: Relative imports (from orchestrata_ultimate.agents.volks.qa. import X, from orchestrata_ultimate.agents.volks.qa.module import Y) only work with installed packages. Rewrite them to absolute imports using the package directory name so they work with sys.path. G115: Also scan root-level files — strip the dot prefix so 'from orchestrata_ultimate.agents.volks.q...
Q60: What is the default value rule for os.environ["REDIS_HOST"]?
Category: Porting & AST
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/redis_factory.py:1-50— """Thread-safe Redis client factory for Python 3.14t (free-threaded). Provides centralized construction of sync and async Redis clients with proper connection pooling for GIL-free environments. Sync clients use BlockingConnectionPool (thread-safe checkout). Async clients use redis.asyncio.ConnectionPool (asyncio serializes). Usage: from orchestrata_ultimate.core.redis_factory import ( .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/standards.py:91-140— 27. Default iterators (for key in adict:notfor key in adict.keys():). 28. Nostaticmethod.classmethodonly for named constructors. 29. Proper locking for threading. No blocking in async without to_thread(). PYTHON 3.13 COMPAT:asyncio.Awaitable,asyncio.Coroutine,asyncio.Generatorwere REMOVED in Python 3.13. Usefrom collections.abc import Awaitable, Coroutineo.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/config.py:91-140— All connection parameters read from environment variables with sensible localhost defaults. """ # Optional explicit environment mapping for testing. # When provided, post_init overrides all fields from this dict. environ: dict[str, str] | None = None # Server host: str = field( default_factory=lambda: os.environ.get("GATEWAY_HOST", "localhost") ) ...
Q61: What are the connection env vars for Neo4j?
Category: Knowledge Graph & Neo4j
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/constants.py:1-46— """Knowledge Graph configuration constants.""" import os # Spec: 008 - Knowledge Graph # Neo4j connection defaults. # The URI is built from NEO4J_HOST / NEO4J_PORT when NEO4J_URI is not set so # that no full connection URI literal appears in source. _NEO4J_HOST: str = os.environ.get("NEO4J_HOST", "localhost") _NEO4J_PORT: str = os.environ.get("NEO4J_PORT", "7687") NEO4J_URI: str = os.environ.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/neo4j_factory.py:1-50— """Neo4j driver factory. Creates sync or async Neo4j drivers from environment configuration. Real drivers only — no in-memory or fake fallbacks. """ from future import annotations import os from typing import Any from neo4j import AsyncGraphDatabase, Auth, GraphDatabase, basic_auth def _resolve_auth(auth: tuple[str, str] | Auth | None) -> Any: """Resolve Neo4j auth from env vars w.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/config.py:91-140— All connection parameters read from environment variables with sensible localhost defaults. """ # Optional explicit environment mapping for testing. # When provided, post_init overrides all fields from this dict. environ: dict[str, str] | None = None # Server host: str = field( default_factory=lambda: os.environ.get("GATEWAY_HOST", "localhost") ) ...
Q62: What is the Neo4jSyncer class responsible for?
Category: Knowledge Graph & Neo4j
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/sync.py:1-50— # mypy: ignore-errors """Neo4jSyncer — sync serialized AST JSON into Neo4j knowledge graph. 12 node types, 18 relationship types, nodes-first-edges-second pattern. """ from future import annotations import logging import os from typing import Any from neo4j import GraphDatabase from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.knowledge_graph._sync_quer.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/synchronizer.py:46-95— @traces(spec="SPEC-008", story="US-008-001", reqs=["REQ-008-001", "REQ-016-165"]) def init( self, neo4j_client: Neo4jClient | None = None, parser: CodeParser | None = None, config: SyncConfig | None = None, driver: neo4j.Driver | None = None, ) -> None: self._client = neo4j_client self._parser = parser or CodeParser() s.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/kg_sync/__init__.py:1-50— # mypy: ignore-errors """Knowledge Graph Syncer — 8-stage sync pipeline. Syncs full project state to Neo4j: project, sessions, folders, code, specs, traces, session-file links, and documentation coverage. Uses sync Neo4j driver (not async) since CLI and hooks are synchronous. All operations use MERGE for idempotency. Spec: 006 - Session Management Decomposed into responsibility mixins: stag...
Q63: How many node types and relationship types does the Neo4j sync engine use?
Category: Knowledge Graph & Neo4j
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/sync.py:1-50— # mypy: ignore-errors """Neo4jSyncer — sync serialized AST JSON into Neo4j knowledge graph. 12 node types, 18 relationship types, nodes-first-edges-second pattern. """ from future import annotations import logging import os from typing import Any from neo4j import GraphDatabase from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.knowledge_graph._sync_quer.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/types_enums.py:1-50— """Knowledge Graph enumeration types and UTC helper. Enums for node types, relationship types, change types, and sync status, plus the sharedutc_nowfactory used across the knowledge graph models. """ from datetime import UTC, datetime from enum import StrEnum from orchestrata_ultimate.core.base.traces import traces @traces(spec="008", story="US-001", reqs=["REQ-008-001", "REQ-015-193.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/constants.py:91-101— # Neo4j labels and relationship types NEO4J_SESSION_LABEL = "Session" NEO4J_FACT_LABEL = "Fact" NEO4J_PORTFOLIO_LABEL = "Portfolio" NEO4J_WORKITEM_LABEL = "WorkItem" NEO4J_BELONGS_TO = "BELONGS_TO" NEO4J_ABOUT = "ABOUT" NEO4J_NEXT_SESSION = "NEXT_SESSION" NEO4J_CARRIED_OVER_FROM = "CARRIED_OVER_FROM" NEO4J_CARRIED_OVER_TO = "CARRIED_OVER_TO"
Q64: What is the nodes-first-edges-second pattern in the sync engine?
Category: Knowledge Graph & Neo4j
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/sync.py:1-50— # mypy: ignore-errors """Neo4jSyncer — sync serialized AST JSON into Neo4j knowledge graph. 12 node types, 18 relationship types, nodes-first-edges-second pattern. """ from future import annotations import logging import os from typing import Any from neo4j import GraphDatabase from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.knowledge_graph._sync_quer.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/synchronizer.py:586-635— are persisted via UNWIND for edges where both endpoints are new. 2. Scoped Cypher queries resolve against nodes linked to files in the current batch (callers/children), while still matching callees/parents from the full graph (pre-existing nodes). This keeps memory bounded regardless of total graph size. Args: call_edges: CALLS edges.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/synchronizer.py:496-545— _logger.warning("Resolution edge persist failed: %s", exc) elapsed = (time.perf_counter() - start) * 1000 return ResolutionResult( sync_results=sync_results, call_edges=len(call_edges), inherit_edges=len(inherit_edges), import_edges=len(import_edges), method_edges=len(method_edges), total_edges_created=edg...
Q65: How does the codebase prevent Neo4j CREATE without test_id?
Category: Knowledge Graph & Neo4j
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/sync.py:721-770— "qualified_name", f"{mod_path}.{cls['name']}.{method['name']}", ) self._collect_test_entity( method, test_id, mod_path, all_targets, test_nodes, tested_by_edges, ) .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/sync.py:766-815— tested_by_edges: list[dict[str, str]], ) -> None: """Append the Test node and any TESTED_BY edges for one test entity.""" test_nodes.append( { "id": test_id, "name": entity.get("name", ""), "qualified_name": entity.get("qualified_name", ""), "module_path": mod_path, "file_path": entit.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/sync.py:46-95— def connect(self) -> None: """Establish Neo4j driver connection.""" self._driver = GraphDatabase.driver( self._uri, auth=(self._user, self._password), ) self._driver.verify_connectivity() def close(self) -> None: """Close Neo4j driver connection.""" if self._driver: self._driver.close() self._dr...
Q66: What command verifies knowledge graph connectivity?
Category: Knowledge Graph & Neo4j
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/kg_context.py:46-95— neo4j_user: str = "", neo4j_password: str = "", driver: neo4j.Driver | None = None, ) -> None: """Initialize KG context. Args: neo4j_uri: Bolt URI for Neo4j. neo4j_user: Neo4j username. neo4j_password: Neo4j password. driver: Optional shared Neo4j driver. If provided, uri/user/password are ignor.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/cli/kg.py:1-50— # mypy: ignore-errors """Knowledge graph sync command and query group for the session CLI.""" from future import annotations import json import sys from typing import TYPE_CHECKING import click from orchestrata_ultimate.core.base.traces import traces if TYPE_CHECKING: from orchestrata_ultimate.session.kg_context import KnowledgeGraphContext from orchestrata_ultimate.session.kg_.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/cli_main.py:586-635— @traces( spec="000", story="US-007", reqs=["REQ-000-062", "REQ-000-063", "REQ-006-FR7-003"] ) def log() -> None: """Show governance decision history.""" click.echo("No governance decisions found") @governance.command() @traces( spec="000", story="US-007", reqs=["REQ-000-064", "REQ-000-065", "REQ-006-FR7-004"] ) def policies() -> None: """List active governance policies."""...
Q67: Where is the single source of truth for service IPs and ports?
Category: Knowledge Graph & Neo4j
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/endpoint_config.py:1-50— """Single source of truth for CXP gRPC endpoint configuration. Ports are read from the environment (ultimately from.env) at call time, with no hardcoded numeric default anywhere — a missing port raises rather than silently falling back, so the port lives in exactly one place and cannot drift between the kernel, the dispatcher, the harness, and the deployment manifests. Host values ar.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ha_controller.py:1-50— """High-availability controller for Memgraph Community replication. The controller monitors a set of Memgraph data instances, enforces a single MAIN role, performs automatic failover, and reintegrates recovered nodes as REPLICAs (autofailback). It exposes an HTTP API so clients can discover the current MAIN URI without hardcoding it. This is intentionally a single-controller design. It is the.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/kernel.py:1-50— """ Orchestration Kernel - Central coordination engine This module provides the central orchestration engine that manages all system components, handles lifecycle events, and provides essential services like service discovery. """ # Spec: 001 - Orchestration Kernel import logging import os from typing import Any from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultim...
Q68: What is the default Neo4j URI fallback if env var is missing?
Category: Knowledge Graph & Neo4j
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/config.py:46-95— SPEC-099 (P0): a deploy that forgets to setJWT_SECRETmust not silently accept the well-known"dev-secret"default -- anyone who knows that default could then forge valid tokens. The default is permitted only in a recognized dev/test context (see :func:_dev_secret_allowed); anywhere else, a missing/default secret is a startup error naming the fix. Raises: .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/neo4j_factory.py:1-50— """Neo4j driver factory. Creates sync or async Neo4j drivers from environment configuration. Real drivers only — no in-memory or fake fallbacks. """ from future import annotations import os from typing import Any from neo4j import AsyncGraphDatabase, Auth, GraphDatabase, basic_auth def _resolve_auth(auth: tuple[str, str] | Auth | None) -> Any: """Resolve Neo4j auth from env vars w.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/constants.py:1-46— """Knowledge Graph configuration constants.""" import os # Spec: 008 - Knowledge Graph # Neo4j connection defaults. # The URI is built from NEO4J_HOST / NEO4J_PORT when NEO4J_URI is not set so # that no full connection URI literal appears in source. _NEO4J_HOST: str = os.environ.get("NEO4J_HOST", "localhost") _NEO4J_PORT: str = os.environ.get("NEO4J_PORT", "7687") NEO4J_URI: str = os.environ...
Q69: How are test_id-based cleanups performed in Neo4j tests?
Category: Knowledge Graph & Neo4j
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/standards.py:136-185— --- Tests --- 32. Tests present and correct. Clear assertions. No false positives. NO MOCKING — test real behavior only. 32b. TEST CLEANUP: Usetmp_pathpytest fixture for temporary files. 32c. NO MONKEY-PATCHING: Use real error conditions, not method replacement. --- Traceability --- 33. TRACEABILITY: Every class and every public function MUST have a @traces(spec=..., story=..., req.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/isolation.py:181-230— self._run_incus(["incus", "delete", container, "--force"], check=False) shutil.rmtree(self._host_dir(test_id), ignore_errors=True) @classmethod def _container_name(cls, test_id: str) -> str: """Derive a valid, collision-safe Incus instance name from test_id. Incus names allow only lowercase letters, digits and hyphens, must start with a letter, and ar.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:451-500— for r in rels: relationships[r.id] = r return GraphRAGContext( entities=list(entities.values()), relationships=list(relationships.values()), chunks=chunks, ) async def _all_entities(self) -> list[Entity]: """Return all entities in the graph (used by retrieve heuristic).""" def _fetch(driver: A...
Q70: What is the relationship between sync_ast_to_neo4j.py and knowledge_graph/sync.py?
Category: Knowledge Graph & Neo4j
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/sync.py:1-50— # mypy: ignore-errors """Neo4jSyncer — sync serialized AST JSON into Neo4j knowledge graph. 12 node types, 18 relationship types, nodes-first-edges-second pattern. """ from future import annotations import logging import os from typing import Any from neo4j import GraphDatabase from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.knowledge_graph._sync_quer.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/kg_sync/__init__.py:1-50— # mypy: ignore-errors """Knowledge Graph Syncer — 8-stage sync pipeline. Syncs full project state to Neo4j: project, sessions, folders, code, specs, traces, session-file links, and documentation coverage. Uses sync Neo4j driver (not async) since CLI and hooks are synchronous. All operations use MERGE for idempotency. Spec: 006 - Session Management Decomposed into responsibility mixins: stag.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/sync.py:91-140— with self._driver.session() as session: for cql in constraints: try: session.run(cql) except Exception as exc: _logger.warning("Constraint failed (may already exist): %s", exc) for cql in indexes: try: session.run(cql) except Exception as exc: ...
Q71: What is the renamed class for BaseVolk in the target codebase?
Category: Agents & Volks
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/task_executor.py:46-95— def _task_to_packet(task: TaskDefinition) -> CXPPacket: """Convert a TaskDefinition to a CXPPacket.""" from orchestrata_ultimate.cxp.packet import CXPPacket, WorkStatus metadata = task.parameters or {} return CXPPacket( from_agent=metadata.get("from_agent", "SYSTEM"), to_agent=task.target, thread_id=metadata.get("thread_id", ""), subject=task.act.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/interfaces/__init__.py:91-140— "MemoryClassification", "SummarizationStrategy", "MemoryScope", "MemoryEntry", "MemorySummary", "ITokenCounter", "IMemoryEngine", # providers (SPEC-011) "ModelCapability", "RoutingStrategy", "TokenUsage", "CostRecord", "CostEstimate", "Message", "ProviderRequest", "ProviderResponse", "BudgetStatus", "UsageReport", "ModelPri.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/executor.py:91-140— Optional inscenario.input: -args: Dictionary of keyword arguments passed to the method. """ volk_class = self._import_volk_class(scenario.volk) instance = volk_class() method_name = scenario.input.get("method") if not method_name: raise ValueError("direct_call mode requires 'method' in scenario.input") args = scenar...
Q72: What is the base class for agents: Pydantic BaseModel or dataclass?
Category: Agents & Volks
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/base.py:1-50— # mypy: ignore-errors """Base agent class. Inherits from Pydantic BaseModel for validation, serialization, and identity. """ from future import annotations import logging import re from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING, Any from uuid import UUID, uuid4 from pydantic import BaseModel, ConfigDict, Field, PrivateAttr from orchestrata_.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/base.py:46-95— Returns the surrounding prose with any DSML markup spans removed and leading/trailing whitespace stripped. Content with no markup passes through with only whitespace-stripping applied (module-level pure function -- no side effects, no dependency onself-- so it is directly unit-testable in isolation of the tool loop). """ if not content: return content r.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/models.py:1-50— """Core dataclasses and Pydantic models for the Volk Test Harness.""" from future import annotations import logging import os from dataclasses import dataclass, field from datetime import UTC, datetime from pathlib import Path from typing import Any, ClassVar, Literal from pydantic import BaseModel, Field from orchestrata_ultimate.core.base.traces import traces # ----------------------...
Q73: How many role volks are being ported?
Category: Agents & Volks
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/volk_registry.py:1171-1186— ] return results @property def count(self) -> int: """Get number of registered Volks.""" return len(self._volks) # Class alias for orchestrata-ultimate naming @traces(spec="000", story="US-000", reqs=["REQ-000-000", "REQ-007-FR10-126"]) class AgentRegistry(VolkRegistry): """Registry alias for VolkRegistry.""" pass/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/pipeline.py:136-185— print("[Kernel] Ready.") print() self.booted = True # Publish the standing roster into the KG so the semantic layer is # queryable with live Volk nodes (best-effort; no-op without Neo4j). synced = self.sync_volks_to_kg() if synced: print(f" [KG Volks] {synced} volk nodes synced") self.events.emit_boot(list(self.volks.keys())).../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/pipeline_kg.py:91-140— # context_injector._driver is a raw (sync) neo4j.Driver -> pass it to # FileSynchronizer'sdriver=param. Passing it asneo4j_client=# leaves self._driver=None, so _acquire_sync_driver falls through to #getattr(client, "_config"), raises "Neo4j client has no _config", # the bare except swallows it, and the sync writes ZERO nodes. ...
Q74: What is the import translation for orchestrata.volks.analyst.volk?
Category: Agents & Volks
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/analyst/__init__.py:1-5— """Volk: analyst (U:BA) — Business Analyst for requirements elicitation.""" from orchestrata_ultimate.agents.volks.analyst.agent import AnalystAgent all = ["AnalystAgent"]/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/task_executor.py:46-95— def _task_to_packet(task: TaskDefinition) -> CXPPacket: """Convert a TaskDefinition to a CXPPacket.""" from orchestrata_ultimate.cxp.packet import CXPPacket, WorkStatus metadata = task.parameters or {} return CXPPacket( from_agent=metadata.get("from_agent", "SYSTEM"), to_agent=task.target, thread_id=metadata.get("thread_id", ""), subject=task.act.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/junior/__init__.py:1-5— """Volk: junior (J:Junior) -- Junior developer for design review and code generation.""" from orchestrata_ultimate.agents.volks.junior.volk import JuniorAgent, JuniorVolk all = ["JuniorAgent", "JuniorVolk"]
Q75: What pattern replaces ABC interfaces in the target codebase?
Category: Agents & Volks
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/tools/base.py:46-95— @runtime_checkable class Tool(Protocol): """Thin structural interface for a single callable tool. Composition-over-inheritance boundary (seeplans/build-factory-architecture.md): a concrete tool implements this Protocol — the ONE ABC/Protocol boundary — and COMPOSES its behaviour (executor, validator, tracer, registry lookups, ...) via constructor DI rather than a d.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/interfaces/providers.py:1-50— """ Provider manager interface types for Orchestrata (SPEC-011). Thin compatibility facade. The canonical definitions live in :mod:orchestrata_ultimate.core.protocols.providers(the interfaces->protocols twin-collapse); this module re-exports them under the historicalI-prefixed names so existingfrom core.interfaces.providers import ...imports keep working and resolve to the single.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/interfaces/memory.py:1-37— """ Memory engine interface types for Orchestrata (SPEC-010). Thin compatibility facade. The canonical definitions live in :mod:orchestrata_ultimate.core.protocols.memory(the interfaces->protocols twin-collapse); this module re-exports them under the historicalI-prefixed names so existingfrom core.interfaces.memory import ...imports keep working and resolve to the single canonica...
Q76: What is the CXPMixin used for in BaseAgent?
Category: Agents & Volks
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/base.py:1-50— # mypy: ignore-errors """Base agent class. Inherits from Pydantic BaseModel for validation, serialization, and identity. """ from future import annotations import logging import re from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING, Any from uuid import UUID, uuid4 from pydantic import BaseModel, ConfigDict, Field, PrivateAttr from orchestrata_.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/base.py:181-230— """ await self._emit_observation(event_type, payload) packet = CXPPacket( from_agent=self.alias, to_agent="SYSTEM", subject=event_type, payload=payload, work_status=WorkStatus.RESOLVED, ) await self.send(packet) @traces( spec="005", story="US-001", reqs=["REQ-005-FR8-001"], .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/base.py:91-140— @property def events(self) -> AsyncObservable[dict[str, Any]]: """Read-only handle to this volk's event mesh node. Other volks (or any subscriber) attach viavolk.events.subscribe(observer)to receive every observation emitted through_emit_observation/emit_event. """ return self._events @traces( spec="005", s...
Q77: What is the agents/lifecycle.py module responsible for?
Category: Agents & Volks
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/lifecycle.py:91-140— "REQ-005-NFR5-001", "REQ-005-US-003-001", "REQ-005-US-004-001", ], ) class AgentLifecycle: """Manages formal lifecycle of agents with constitutional checks. Lifecycle stages: 1. Spawn: validate constraints → create instance → add to hierarchy 2. Run: monitor lifespan, handle actions 3. Terminate: cleanup → notify parent → remove from hierarchy .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/lifecycle.py:1-50— # mypy: ignore-errors """Formal agent lifecycle with constitutional checks. Manages spawn, terminate, and failure handling with constitutional validation at each stage. """ # Spec: 005 - Multi-Agent System from future import annotations import logging import time from dataclasses import dataclass, field from datetime import UTC, datetime from enum import StrEnum from typing import TYPE_.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/runtime.py:1-50— """Volk Runtime — instance lifecycle management. Implements IComponent with ComponentType.VOLK_RUNTIME. Manages spawning, terminating, and tracking Volk instances with Prolog governance checks and Redis state persistence. Author: Mohamed Spec: 005-multi-agent (amended) """ from future import annotations import asyncio import json import logging import time import uuid from datetime impo...
Q78: How are agents created and configured according to sample search results?
Category: Agents & Volks
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/orchestrator/pipeline_router.py:46-95— FIX = "fix" REVIEW = "review" SPEC = "spec" DEPENDENCY = "dependency" PERFORMANCE = "performance" # ── Capability → Agent ───────────────────────────────────────────────── CAPABILITY_AGENT: dict[Capability, str] = { Capability.INTAKE: "U:Cx", Capability.ANALYSIS: "U:BA", Capability.SCOPING: "U:PM", Capability.ARCHITECTURE: "U:Ar", Capability.IMPLEMENTATIO.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/tool_registry.py:361-410— "description": ( "Free-text search matched case-insensitively against " "pattern/guide name, problem, solution, use_when, and " "summary fields. Empty string with a category returns " "every entry in that category." ), }, "category": { .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/lifecycle.py:91-140— "REQ-005-NFR5-001", "REQ-005-US-003-001", "REQ-005-US-004-001", ], ) class AgentLifecycle: """Manages formal lifecycle of agents with constitutional checks. Lifecycle stages: 1. Spawn: validate constraints → create instance → add to hierarchy 2. Run: monitor lifespan, handle actions 3. Terminate: cleanup → notify parent → remove from hierarchy ...
Q79: What is the difference between agents/swarm and agents/volks?
Category: Agents & Volks
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/orchestrator/pipeline_router.py:46-95— FIX = "fix" REVIEW = "review" SPEC = "spec" DEPENDENCY = "dependency" PERFORMANCE = "performance" # ── Capability → Agent ───────────────────────────────────────────────── CAPABILITY_AGENT: dict[Capability, str] = { Capability.INTAKE: "U:Cx", Capability.ANALYSIS: "U:BA", Capability.SCOPING: "U:PM", Capability.ARCHITECTURE: "U:Ar", Capability.IMPLEMENTATIO.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/persona.py:1-41— """Shared volk identity + communication-fabric preamble. Every volk carries a strong role persona (its own system prompt: Concierge, Architect, QA, …). What none of them carried was the common framing that makes them a coherent team — who they are within Orchestrata, that they hand work to peer volks over CXP, and that the orchestrator observes their status at all times. ``compose_person.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/volk_registry.py:1-50— """Volk Registration and Discovery for Orchestration Kernel. This module implements Volk (agent) registration, validation, and discovery as specified in US-001 of the Orchestration Kernel spec. Requirements Implemented: - REQ-001-001: Register with unique alias (U:Ox, U:PS, etc.) - REQ-001-002: Include capabilities declaration - REQ-001-003: Include supported protocols (CXP, gRPC, WS) - REQ-0...
Q80: What design decision replaced the 26-package monorepo with a single package?
Category: Agents & Volks
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/__init__.py:1-6— """ Orchestrata Ultimate — Unified orchestration platform. Ported from 26-package monorepo into single cohesive package. """ version = "1.0.0"/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/architect/architect_prompts.py:136-185— +subscribe(topic_pattern: str, handler: Callable, subscriber_id: str) EventSubscription }} IEventBus <|.. EventEmittingFoo : uses ``` > HARD RULES (violations cause implementation failure): - Python 3.14+ with type hints on ALL public functions and methods - Absolute imports only (no relative imports) - No mocking in tests — real implementations only - pyproject.toml for project metada.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/architect/architect_prompts.py:226-248— ARCHITECTURE: {architecture} REQUIREMENTS: {requirements_doc} FILE_TREE: {file_tree} API_SPEC: {api_spec} CLASS_DIAGRAM: {class_diagram} Junior developer's questions: {questions} {monorepo_src_section} Update the design to address ALL questions. Output the COMPLETE updated \ design in the same format (ARCHITECTURE, REQUIREMENTS, FILE_TREE, API_SPEC, \ CLASS_DIAGRAM sections). Do not skip a...
Q81: What does CXP stand for and what are its main components?
Category: CXP & Protocols
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/__init__.py:1-50— """ Cross-agent eXchange Protocol (CXP) Implementation. CXP is the structured communication protocol used by Volks for agent-to-agent communication in Orchestrata. Module structure: Core: packet.py - CXPPacket class and enums types.py - Extended types (ThreadMetadata, RoutingResult, VOLK_ALIASES) constants.py - Protocol constants and limits exception.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/__init__.py:91-140— from orchestrata_ultimate.cxp.exceptions import ( CXPError, DeliveryError, DeliveryTimeoutError, DeserializationError, FieldValidationError, InvalidPacketError, MaxRetriesExceededError, NoRouteError, PrologBridgeError, RoutingError, SchemaValidationError, SerializationError, StateTransitionError, ThreadClosedError, ThreadError, Thr.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/__init__.py:271-313— "IThreadManager", "IDeliveryTracker", "IPrologBridge", # Header/Payload "HeaderManager", "extract_headers", "validate_volk_alias", "validate_uuid", "validate_subject", "PayloadManager", "JSON_MAX_BYTES", "BINARY_MAX_BYTES", # State Machines "DeliveryTracker", "validate_delivery_transition", "WorkTracker", "validate_work_transition"...
Q82: What is a CXPPacket?
Category: CXP & Protocols
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/_packet_impl.py:496-545— "REQ-004-US-001-020", "REQ-004-US-001-022", "REQ-004-US-001-023", "REQ-004-US-001-025", "REQ-004-US-001-027", "REQ-004-US-001-028", "REQ-004-US-001-029", "REQ-004-US-001-030", "REQ-004-US-002-001", "REQ-004-US-002-019", "REQ-004-US-002-020", "REQ-004-US-002-021", "REQ-004-US-002-023", "REQ-004-US-004-006",.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/__init__.py:1-50— """ Cross-agent eXchange Protocol (CXP) Implementation. CXP is the structured communication protocol used by Volks for agent-to-agent communication in Orchestrata. Module structure: Core: packet.py - CXPPacket class and enums types.py - Extended types (ThreadMetadata, RoutingResult, VOLK_ALIASES) constants.py - Protocol constants and limits exception.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/proto/cxp_pb2.pyi:46-95— ACK_STATUS_SUCCESS: _ClassVar[AckStatus] ACK_STATUS_FAILURE: _ClassVar[AckStatus] ACK_STATUS_RETRY: _ClassVar[AckStatus] DELIVERY_STATUS_UNSPECIFIED: DeliveryStatus DELIVERY_STATUS_QUEUED: DeliveryStatus DELIVERY_STATUS_DELIVERED: DeliveryStatus DELIVERY_STATUS_ACKNOWLEDGED: DeliveryStatus DELIVERY_STATUS_FAILED: DeliveryStatus WORK_STATUS_UNSPECIFIED: WorkStatus WORK_STATUS_PENDING: W...
Q83: Where is the protobuf definition located?
Category: CXP & Protocols
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/proto/cxp_pb2.py:1-50— # Generated by the protocol buffer compiler. DO NOT EDIT! # NO CHECKED-IN PROTOBUF GENCODE # source: cxp.proto # Protobuf Python Version: 6.31.1 """Generated protocol buffer code.""" from google.protobuf import descriptor as _descriptor from google.protobuf import descriptor_pool as _descriptor_pool from google.protobuf import runtime_version as _runtime_version from google.protobuf import sy.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/proto/__init__.py:1-43— """Generated protobuf classes for CXP gRPC. This module is auto-generated from cxp.proto. Do not edit manually. """ # Pre-load well-known protobuf types so the descriptor pool has them # before cxp_pb2 tries to resolve google/protobuf/timestamp.proto. # Required on protobuf 6.x with Python 3.14t. import google.protobuf.struct_pb2 # noqa: F401 import google.protobuf.timestamp_pb2 # noqa: F40.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/proto/cxp_pb2.py:46-66— _globals["_CXPHEALTHREQUEST"]._serialized_start = 795 _globals["_CXPHEALTHREQUEST"]._serialized_end = 831 _globals["_CXPHEALTHRESPONSE"]._serialized_start = 833 _globals["_CXPHEALTHRESPONSE"]._serialized_end = 957 _globals["_CXPRECEIVEREQUEST"]._serialized_start = 959 _globals["_CXPRECEIVEREQUEST"]._serialized_end = 1043 _globals["_CXPSUBSCRIPTION"]._serialized_start = 1...
Q84: What is the role of the CXP serializer?
Category: CXP & Protocols
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/serializer.py:1-50— """ CXP Serializer - Serializes and deserializes CXP packets. Handles JSON serialization for network transmission and storage. Spec: 004-cxp Author: Mohamed """ import json import logging from typing import Any from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultimate.cxp.packet import CXPPacket logger = logging.getLogger(name) @traces( spec="004", st.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/binary_serializer.py:1-50— # mypy: ignore-errors """ CXP Binary Serializer. Binary serialization and deserialization for CXP packets. Uses MessagePack when available, falls back to JSON + gzip. Implements ISerializer binary methods per Spec 004 Section US-006. Performance targets: - to_bytes: < 5ms for 1KB payload - from_bytes: < 5ms Author: Mohamed """ import base64 import gzip import json import logging fro.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/json_serializer.py:1-50— # mypy: ignore-errors """ CXP JSON Serializer. JSON serialization and deserialization for CXP packets. Implements ISerializer JSON methods per Spec 004 Section US-006. Performance targets: - to_json: < 10ms for 1KB payload - from_json: < 10ms Author: Mohamed """ # Spec: 004 - CXP Protocol import base64 import json import logging import math from datetime import datetime from enum i...
Q85: How does the CXP dispatcher route packets?
Category: CXP & Protocols
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/dispatcher.py:46-95— class CXPDispatcher: """CXP-level packet dispatcher. Combines routing (via :class:CXPRouter) with delivery tracking to actually send packets and record the outcome. The dispatch flow: 1. Route the packet via the router to determine the target(s). 2. Attempt delivery (mark DELIVERED on success, FAILED on failure). 3. Record the result in the delivery tracker (if a.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/dispatcher.py:1-50— # mypy: ignore-errors """ CXP Packet Dispatcher. Responsible for actually SENDING a CXP packet: routes via the CXPRouter, updates delivery status, records results in the DeliveryTracker, and maintains dispatch history. Spec: 004-cxp Author: Mohamed """ import logging from datetime import UTC, datetime from typing import TYPE_CHECKING from orchestrata_ultimate.core.base.traces import traces .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/cxp/dispatcher.py:91-140— Uses the router to determine the destination, then delivers. If no router is configured, delivers directly topacket.to_agent. Args: packet: The CXP packet to dispatch. Returns: A DeliveryResult describing the outcome. Raises: DeliveryError: If delivery fails after all retry attempts. """ destination = pa...
Q86: What Incus container runs Redis?
Category: Infrastructure & Deployment
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/isolation.py:181-230— self._run_incus(["incus", "delete", container, "--force"], check=False) shutil.rmtree(self._host_dir(test_id), ignore_errors=True) @classmethod def _container_name(cls, test_id: str) -> str: """Derive a valid, collision-safe Incus instance name from test_id. Incus names allow only lowercase letters, digits and hyphens, must start with a letter, and ar.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/isolation.py:136-185— path is that host directory: the caller stages files there (visible inside the container) and drives the container withincus exec. Theoh-trial-prefix guarantees trial containers never collide with standing infrastructure. """ _NAME_PREFIX = "oh-trial-" _MOUNT_POINT = "/workspace" async def create(self, test_id: str) -> Path: if shutil.which("incus") .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/integrations/__init__.py:1-1— """External integrations (Intent Broker, Redis)."""
Q87: What Incus container runs Neo4j?
Category: Infrastructure & Deployment
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/isolation.py:181-230— self._run_incus(["incus", "delete", container, "--force"], check=False) shutil.rmtree(self._host_dir(test_id), ignore_errors=True) @classmethod def _container_name(cls, test_id: str) -> str: """Derive a valid, collision-safe Incus instance name from test_id. Incus names allow only lowercase letters, digits and hyphens, must start with a letter, and ar.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/models.py:91-140— """Runtime context passed to setup hooks, evaluators, and executors.""" scenario_id: str volk_name: str trial_number: int test_id: str # UUID for resource isolation worktree_path: Path redis_db: int # Worker-isolated Redis DB number neo4j_prefix: str # test_id prefix for Neo4j resources config: HarnessConfig logger: logging.Logger @dataclass @traces(spe.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/isolation.py:136-185— path is that host directory: the caller stages files there (visible inside the container) and drives the container withincus exec. Theoh-trial-prefix guarantees trial containers never collide with standing infrastructure. """ _NAME_PREFIX = "oh-trial-" _MOUNT_POINT = "/workspace" async def create(self, test_id: str) -> Path: if shutil.which("incus") ...
Q88: What command starts all four test services in parallel?
Category: Infrastructure & Deployment
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/threading/test_executor.py:586-635— return results @traces(spec="000", story="US-000", reqs=["REQ-000-000"]) def stop(self) -> None: self._pool.stop(wait=True, cancel_futures=True) @traces(spec="000", story="US-000", reqs=["REQ-000-000"]) def run_tests_in_parallel( test_paths: list[str], pytest_args: list[str] | None = None, max_workers: int = 4, timeout_per_test: float = 600.0, batch_multip.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/cli_main.py:226-275— except ImportError as e: logger.warning("Agent runtime not available: %s", e) except Exception as e: logger.warning("Agent runtime init failed: %s", e) def _boot_start_services(config: dict[str, Any], ready_event: threading.Event) -> None: api_cfg = config.get("infrastructure", {}).get("api", {}) health_host = api_cfg.get("host", "0.0.0.0") health_port = api_cf.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/threading/test_executor.py:451-500— class ParallelTestExecutor: """Run test modules in parallel via batched subprocess workers. The ThreadPoolExecutor dispatches subprocesses (one per batch). Each batch runs as a full pytest session with all fixtures intact. """ @traces(spec="000", story="US-000", reqs=["REQ-000-000"]) def init(self, max_workers: int = 4, batch_multiplier: int = 1): # batch_m...
Q89: How do you check service status without failing if services are down?
Category: Infrastructure & Deployment
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/postman/health.py:91-140— return { "status": "healthy" if pong else "unhealthy", "service": "redis", } # pragma: no cover except Exception as e: return {"status": "unhealthy", "service": "redis", "error": str(e)} @traces(spec="SPEC-002", story="US-002-001", reqs=["REQ-001-US-002-001"]) async def check_neo4j(self) -> dict[str, Any]: """Check Neo4j.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/postman/health.py:136-185— # Neo4j check if available try: import neo4j # noqa: F401 checks.append(await self.check_neo4j()) except ImportError: logger.debug("Neo4j package not installed, skipping check") checks.append( { "status": "unknown", "service": "neo4j", "error": "neo4j not.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/orchestrator/engine.py:316-365— self._logger.debug("Redis state manager does not have close method") except Exception as e: self._logger.error(f"Error closing Redis connection: {e}") try: await self.neo4j.close() except AttributeError: self._logger.debug("Neo4j audit trail does not have close method") except Exception as e: self._logger.error(f"E...
Q90: What is the Memgraph HA cluster composed of?
Category: Infrastructure & Deployment
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ha_controller.py:1-50— """High-availability controller for Memgraph Community replication. The controller monitors a set of Memgraph data instances, enforces a single MAIN role, performs automatic failover, and reintegrates recovered nodes as REPLICAs (autofailback). It exposes an HTTP API so clients can discover the current MAIN URI without hardcoding it. This is intentionally a single-controller design. It is the.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ha_controller_cli.py:91-140— ) parser.add_argument( "--db-path", default=os.environ.get("MEMGRAPH_HA_DB_PATH", "/var/lib/memgraph-ha/state.db"), help="SQLite state file path", ) parser.add_argument( "--poll-interval", type=float, default=float(os.environ.get("MEMGRAPH_HA_POLL_INTERVAL", "2.0")), help="Polling interval in seconds", ) parser.add_argu.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ha_controller.py:46-95— healthy: bool = False last_seen: float = 0.0 error: str | None = None @dataclass class ClusterStatus: """Snapshot returned by the HTTP status endpoint.""" current_main: str | None epoch: int nodes: list[NodeState] class MemgraphHAController: """Monitor and manage a Memgraph replication cluster.""" def init( self, nodes: list[NodeConfig],...
Q91: What is the Memgraph HA controller URL?
Category: Infrastructure & Deployment
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:46-95— self._driver: Any | None = None self._current_uri: str | None = None self._vector_index_dimension: int | None = None async def connect(self) -> bool: """Open the Bolt driver and verify connectivity. If an HA controller URL is configured, discover the current MAIN URI from the controller before connecting. """ uri = await self._resolv.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ha_controller.py:46-95— healthy: bool = False last_seen: float = 0.0 error: str | None = None @dataclass class ClusterStatus: """Snapshot returned by the HTTP status endpoint.""" current_main: str | None epoch: int nodes: list[NodeState] class MemgraphHAController: """Monitor and manage a Memgraph replication cluster.""" def init( self, nodes: list[NodeConfig],.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ha_controller_cli.py:1-50— """CLI entry point for the Memgraph HA controller. Usage: python -m orchestrata_ultimate.semantic_graph.ha_controller_cli \ --nodes 'memgraph-1=bolt://10.92.53.10:7687;10.92.53.10' \ --nodes 'memgraph-2=bolt://10.92.53.11:7687;10.92.53.11' \ --nodes 'memgraph-3=bolt://10.92.53.12:7687;10.92.53.12' \ --port 8080 Environment variables (alternative to --nodes)...
Q92: Why did the standalone memgraph-db container need to be restarted during this work?
Category: Infrastructure & Deployment
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:316-365— ) with driver.session() as session: session.run(cypher) logger.info("Creating vector index %s with dimension %s", self._config.vector_index_name, dimension) await self._execute_with_reconnect(_create) await self._wait_for_vector_index() self._vector_index_dimension = dimension async def _wait_for_vector_index(self) -> None: .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:451-500— for r in rels: relationships[r.id] = r return GraphRAGContext( entities=list(entities.values()), relationships=list(relationships.values()), chunks=chunks, ) async def _all_entities(self) -> list[Entity]: """Return all entities in the graph (used by retrieve heuristic).""" def _fetch(driver: A.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/schema.py:91-140— # Provenance / ownership / decision edges (W5) — the agent↔work and # artifact-provenance layer SPEC-008 specified but that had zero occurrences. REL_ASSIGNED_TO = "ASSIGNED_TO" # (:Task)-[:ASSIGNED_TO]->(:Volk) REL_OWNS = "OWNS" # (:Volk)-[:OWNS]->(:Component|:File) REL_EXECUTED_BY = "EXECUTED_BY" # (:PipelineRun)-[:EXECUTED_BY]->(:Volk) REL_TRIGGERED = "TRIGGERED" # (:Intent)-[:TRIGGERED]...
Q93: What command launches a Memgraph container from the project?
Category: Infrastructure & Deployment
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/cli_main.py:361-410— try: url = f"http://{host}:{port}/health/live" with urllib.request.urlopen(url, timeout=5) as resp: data = json.loads(resp.read()) click.echo(f"Status: {data.get('status', 'unknown')}") except (OSError, ValueError) as e: click.echo(f"Health check failed: {e}", err=True) raise SystemExit(1) # ── INIT COMMAND ──────────────────────────.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ha_controller_cli.py:91-140— ) parser.add_argument( "--db-path", default=os.environ.get("MEMGRAPH_HA_DB_PATH", "/var/lib/memgraph-ha/state.db"), help="SQLite state file path", ) parser.add_argument( "--poll-interval", type=float, default=float(os.environ.get("MEMGRAPH_HA_POLL_INTERVAL", "2.0")), help="Polling interval in seconds", ) parser.add_argu.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ha_controller_cli.py:1-50— """CLI entry point for the Memgraph HA controller. Usage: python -m orchestrata_ultimate.semantic_graph.ha_controller_cli \ --nodes 'memgraph-1=bolt://10.92.53.10:7687;10.92.53.10' \ --nodes 'memgraph-2=bolt://10.92.53.11:7687;10.92.53.11' \ --nodes 'memgraph-3=bolt://10.92.53.12:7687;10.92.53.12' \ --port 8080 Environment variables (alternative to --nodes)...
Q94: How is the wiki.js instance accessed?
Category: Infrastructure & Deployment
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/integrations/__init__.py:1-1— """External integrations (Intent Broker, Redis)."""/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/rpc.py:316-365— "session.list": session.handle_list, "session.get": session.handle_get, "session.active": session.handle_active, "session.open": session.handle_open, "session.close": session.handle_close, "session.events": session.handle_events, "session.facts.subscribe": session.handle_facts_subscribe, # Governance (FR-018) "governance.decisions".../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/drivers/browser_providers.py:46-95— base_url: str login_url: str input_selector: str submit_selector: str response_selector: str completion_indicator: str code_block_selector: str rate_limit_seconds: int max_session_minutes: int timeout_seconds: int @traces( spec="025", reqs=[ "REQ-025-073", "REQ-025-074", "REQ-025-075", "REQ-025-076", "REQ-025-...
Q95: What is the wiki graph indexer built on?
Category: Infrastructure & Deployment
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/constants.py:1-46— """Knowledge Graph configuration constants.""" import os # Spec: 008 - Knowledge Graph # Neo4j connection defaults. # The URI is built from NEO4J_HOST / NEO4J_PORT when NEO4J_URI is not set so # that no full connection URI literal appears in source. _NEO4J_HOST: str = os.environ.get("NEO4J_HOST", "localhost") _NEO4J_PORT: str = os.environ.get("NEO4J_PORT", "7687") NEO4J_URI: str = os.environ.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:1-50— """Memgraph-backed semantic graph adapter. Uses the official Neo4j Bolt driver because Memgraph speaks the Bolt protocol. The synchronous driver is used internally and wrapped in threads to preserve the async repository interface; this avoids event-loop binding issues in tests. Vector search uses Memgraph's native HNSW vector index via thevector_search.searchquery module. The index is c.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/schema.py:136-185— f"CREATE CONSTRAINT intent_id IF NOT EXISTS FOR (i:{LABEL_INTENT}) REQUIRE i.id IS UNIQUE", f"CREATE CONSTRAINT pipeline_run_id IF NOT EXISTS FOR (r:{LABEL_PIPELINE_RUN}) REQUIRE r.id IS UNIQUE", f"CREATE CONSTRAINT component_name IF NOT EXISTS FOR (c:{LABEL_COMPONENT}) REQUIRE c.name IS UNIQUE", ] # Index DDL INDEXES: list[str] = [ f"CREATE INDEX file_name IF NOT EXISTS FOR (f:{LA...
Q96: What is the rule about silent exception handlers?
Category: Code Standards & Patterns
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/review/checks_structure.py:226-275— ] @traces(spec="007b", story="US-003", reqs=["REQ-009-US-003-001"]) def check_error_handling(filepath: str, tree: ast.Module) -> list[Issue]: """Check 13: ERROR_HANDLING -- no bare except. Args: filepath: Absolute path to the file. tree: Parsed AST module. Returns: List of bare except issues. """ issues: list[Issue] = [] rel_name = os.path.basena.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/codedirector/review_checks.py:676-725— (r"AsyncMock\s*(", "AsyncMock() instantiation"), (r"@\s*(mock.)?patch\s*(", "@patch decorator"), (r"mocker.(patch|spy|stub|mock)\s*(", "pytest-mock usage"), (r"create_autospec\s*(", "create_autospec()"), (r"patch.object\s*(", "patch.object()"), (r"patch.dict\s*(", "patch.dict()"), (r"patch.multiple\s*(", "patch.multiple()"), ( r"class\s+(Mock|Fake.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/codedirector/review_checks.py:406-455— break # raise NotImplementedError if ( isinstance(child, ast.Raise) and child.exc and isinstance(child.exc, ast.Call) and isinstance(child.exc.func, ast.Name) and child.exc.func.id == "NotImplementedError" ): issues.append( ...
Q97: Why is os.environ["X"] forbidden in favor of os.environ.get("X", default)?
Category: Code Standards & Patterns
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/review/checks_safety.py:91-140— continue issues.append(("SECURITY", rel_name, f"L{i}: {desc}.")) break return issues @traces(spec="007b", story="US-003", reqs=["REQ-009-US-003-001"]) def check_env_access( filepath: str, tree: ast.Module, ) -> list[Issue]: """Check 30c: CONFIG -- bareos.environ["X"]subscript reads. Readingos.environ["X"]raisesKeyErrorwhen th.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/standards.py:91-140— 27. Default iterators (for key in adict:notfor key in adict.keys():). 28. Nostaticmethod.classmethodonly for named constructors. 29. Proper locking for threading. No blocking in async without to_thread(). PYTHON 3.13 COMPAT:asyncio.Awaitable,asyncio.Coroutine,asyncio.Generatorwere REMOVED in Python 3.13. Usefrom collections.abc import Awaitable, Coroutineo.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/verification/review/checks_safety.py:136-185— "CONFIG", rel_name, f"L{node.lineno}: bareos.environ[{target}]raises " f"KeyError when unset -- use " f"os.environ.get({target}, default).", ) ) return issues # Parallel test isolation (standards rule 34). Mirrors the blocking # patterns of scripts/gate_parallel_safety.py so t...
Q98: What is the factory pattern used for external services?
Category: Code Standards & Patterns
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/integrations/__init__.py:1-1— """Integration adapters for external APIs and services."""/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/adapters/__init__.py:1-47— """External adapters package. Fresh implementation for service integration adapters. """ from future import annotations from typing import TYPE_CHECKING, Any from orchestrata_ultimate.core.base.traces import traces if TYPE_CHECKING: from orchestrata_ultimate.core.adapters.claude.hooks import HookManager from orchestrata_ultimate.core.adapters.claude.tool_interceptor import ( .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/di/container.py:406-455— return self._singletons.get(interface_name) # For REQUEST scope (REQ-024-015): context-specific cache if registration.scope == ComponentScope.REQUEST: cache_key = f"{interface_name}:{context.context_id}" return context.cache.get(cache_key) return None @traces(spec="024", story="Resolution", reqs=["REQ-024-017", "REQ-024-053"]) def store...
Q99: What is the UUID-prefixed resource pattern for tests?
Category: Code Standards & Patterns
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/isolation.py:181-230— self._run_incus(["incus", "delete", container, "--force"], check=False) shutil.rmtree(self._host_dir(test_id), ignore_errors=True) @classmethod def _container_name(cls, test_id: str) -> str: """Derive a valid, collision-safe Incus instance name from test_id. Incus names allow only lowercase letters, digits and hyphens, must start with a letter, and ar.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/isolation.py:226-232— @traces(spec="102", story="US-006", reqs=["REQ-102-093"]) def get_neo4j_prefix(test_id: str) -> str: """Return the Neo4j resource prefix for a given test run. The prefix is simply the test_id itself. """ return test_id/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/session/constants.py:46-95— ts = when or datetime.now() stamp = ts.strftime("%Y%m%d_%H%M%S") short_uuid = uuid.uuid4().hex[:8] return f"session_{stamp}_{short_uuid}" # Redis key prefixes SESSION_REDIS_PREFIX = "orchestrata:session" PORTFOLIO_REDIS_PREFIX = "orchestrata:portfolio" # Redis key patterns SESSION_KEY = f"{SESSION_REDIS_PREFIX}:{{id}}" SESSION_ACTIVE_SET = f"{SESSION_REDIS_PREFIX}:active" SESSION...
Q100: What is the project's policy on dismissing test failures?
Category: Code Standards & Patterns
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/volk.py:676-725— test_passed, test_output = await self._run_tests() self.test_results = test_output self.output_artifacts.append(f"=== TEST RERUN OUTPUT ===\n{test_output}") # Print summary (same as _handle_test_gen) self._print_test_summary(test_output, test_passed) # Phase C: test-fix escalation — G226: no cap, no forced handoff. # Tests must pass. Period. .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/volk.py:46-95— ISSUES: -HANDOFF: On PASS, results route to U:Cp (compliance_check); on FAIL, issues route back to U:Cd (fix_issues). RULES: - pytest only (no unittest) - ZERO mocking: no unittest.mock, MagicMock, patch, AsyncMock, Mock, no custom Mock/Fake/Stub/Dummy classes. Test REAL behavior only. - Python 3.14+: never use asyncio.Awaitable/asyncio.Coroutine — use collections.abc.Awai... /home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/volks/qa/volk.py:946-995— ): summary_line = stripped if summary_line: print(f" [U:Qa] {summary_line}") else: print(f" [U:Qa] Tests {'PASSED' if test_passed else 'FAILED'}") # G82: Print individual FAILED test names for conversation visibility if not test_passed: for tline in test_output.splitlines(): stripped = tline.s...