Orchestrata Ultimate › Plans › archive › semantic_graph_qa_results_final
Semantic Graph Q&A Results (Final: Hybrid Vector + Keyword + Module Graph)
Source: plans/archive/semantic_graph_qa_results_final.md at 8a2cb42ad · last changed 2026-08-19 · first committed 2026-07-08 · 906 lines.
Prose pending.
Plan text
Semantic Graph Q&A Results (Final: Hybrid Vector + Keyword + Module Graph)
Model: mistralai/codestral-embed-2505 | Reranker: cohere/rerank-4-pro | Project: orchestrata-ultimate
Average keyword coverage: 57% Low-quality answers (<50%): 32/100
Q1: Why must the Gateway never serve HTML, JS, or CSS?
Category: Architecture & Design Keywords: Gateway, serve, HTML Coverage: 40%
/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 Keywords: VizInt, communicate, Gateway, different, ports Coverage: 38%
/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/orchestrator/pipeline_kernel.py:181-230— { "event": "pipeline.complete", "overall_status": result.overall_status.value, "stages_completed": len(result.stages), "total_artifacts": len(result.artifacts), "thread_id": thread_id, "timestamp": _now_iso(), }, ) except Exception as e: ...
Q3: What is the exact port and protocol for each service: VizInt UI, Gateway, Kernel, Redis, Neo4j?
Category: Architecture & Design Keywords: port, protocol, service, VizInt, Gateway Coverage: 70%
/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/standards/architecture.md:1-29— # Orchestrata Ultimate Architecture Overview ## High-Level Design - Monorepo to Single Package Port: 122K LOC from 26 packages tosrc/orchestrata_ultimate/using Python 3.14+, uv manager. - Core Components: Agents (volks), CXP protocol, Providers (LLM adapters), Kernel (event bus, registry), Orchestrator (intent parsing), Knowledge Graph (Neo4j), Session/Memory management, Gateway (W.../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 Keywords: gRPC, kernel, required, process, Gateway Coverage: 83%
/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 Keywords: fallback, gRPC, kernel, unreachable Coverage: 83%
/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/rpc.py:136-185— def set_neo4j_driver(self, driver: Any) -> None: """Inject the Neo4j async driver for knowledge queries. Args: driver: An AsyncGraphDatabase driver instance. """ self._knowledge.set_neo4j_driver(driver) self._projects.set_neo4j_driver(driver) self._session.set_neo4j_driver(driver) self._governance.set_neo4j_driver(driver) ...
Q6: Why are hardcoded service IPs and ports forbidden in this codebase?
Category: Architecture & Design Keywords: hardcoded, service, ports, forbidden Coverage: 50%
/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/standards/orchestrata_specific_checklist.md:316-365— # core/application/user_service.py class UserService: def init(self, cache_port: ICachePort): self.cache = cache_port def cache_user(self, user: User): self.cache.set(f"user:{user.id}", user.to_json())### Violation: Hardcoded Configurationpython # ❌ WRONG - Hardcoded values REDIS_HOST = "localhost" # Hardcoded! redis = Redis(host=REDIS_HOST)/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...
Q7: How does the project enforce the rule against except: pass and except Exception: pass?
Category: Architecture & Design Keywords: except: pass, except Exception: pass, enforce, pass, Exception Coverage: 50%
/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/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...
Q8: What is the three-tier gRPC logging policy for the Gateway?
Category: Architecture & Design Keywords: three, tier, gRPC, logging, policy Coverage: 71%
/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 Keywords: Gateway, health, check, authenticate, bypassing Coverage: 38%
/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 Keywords: file, protocol, long, running, services Coverage: 50%
/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.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/agents/interfaces.py:1-50— """Multi-Agent Runtime interfaces. Runtime-checkable protocols for Volk lifecycle, task execution, pool management, and multi-Volk coordination. Author: Mohamed Spec: 005-multi-agent (amended) """ # Spec: 005 - Multi-Agent System from typing import Any, Protocol, runtime_checkable from orchestrata_ultimate.agents.types import ( ConsensusResult, InstanceStatus, PoolConfig, P.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/protocols/components.py:1-50— """ Core component protocols for Orchestrata Ultimate. Defines the kernel, postman, orchestrator, and state-manager protocols, plus the forward-reference sentinels used to break circular dependencies. Depends on :mod:orchestrata_ultimate.core.protocols.basefor the shared base types. """ from typing import Any from orchestrata_ultimate.core.base.traces import traces from orchestrata_ultim...
Q11: Why must readiness wait for an application-level health check rather than just TCP accept?
Category: Architecture & Design Keywords: readiness, wait, application, level, health Coverage: 55%
/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 Keywords: graceful, shutdown, sequence, services Coverage: 60%
/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 Keywords: idempotent, start, implemented, test, services Coverage: 60%
/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/semantic_graph/adapters/memgraph.py:496-545— async def link_chunks_to_modules(self) -> None: """Create Module nodes and BELONGS_TO edges from chunk metadata. Chunks must have amodulemetadata key. The operation is idempotent. """ def _link(driver: Any) -> None: with driver.session() as session: session.run( f""" MATCH (c:{self._chunk.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/gateway/rpc.py:91-140— self._state = StateHandler(ring_buffer) self._auth = AuthHandler() self._fireflies = FirefliesHandler() self._memory = MemoryHandler() self._test_metrics = TestMetricsHandler() # Wire pub/sub into the orchestrator unconditionally so the UI is # subscribed to pipeline.<thread_id> before dispatch even when Neo4j is # absent. set_neo4j_drive...
Q14: What action is taken when a port is in use by a foreign process?
Category: Architecture & Design Keywords: port, foreign, process Coverage: 14%
/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/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.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/kernel/pam/enforcer.py:316-365— loop, volk, component, execution_time ) if delegation_result is not None: return delegation_result actual_owner = await self._lookup_actual_component_owner(loop, component) return OwnershipResult( status=OwnershipStatus.INVALID, volk=volk, component=component, reason=f"{volk} does not own {component}",...
Q15: Why must subprocess.DEVNULL never be used for service logs?
Category: Architecture & Design Keywords: subprocess.DEVNULL, used, service, logs Coverage: 29%
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/drivers/qwen_driver.py:181-230— _logger.info("Generated %d files", len(created)) return True, output or f"Created {len(created)} files" @traces( spec="SPEC-025", story="US-025-001", reqs=["REQ-025-001", "REQ-003-US-009-024"] ) async def _init_project(self, qwen_bin: str) -> None: """Run qwen init if QWEN.md doesn't exist yet. Args: qwen_bin: Path to qwen binary. ".../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/drivers/claude_api_driver.py:181-230— @traces( spec="SPEC-025", story="US-025-001", reqs=["REQ-025-001", "REQ-003-US-008-035"] ) async def _init_project(self, claude_bin: str) -> None: """Run claude init if CLAUDE.md doesn't exist yet. Args: claude_bin: Path to claude binary. """ claude_md = os.path.join(self.workspace, "CLAUDE.md") if os.path.exists(claude_md): .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/drivers/deepseek_coder_driver.py:181-230— continue if ruff_bin: try: proc = await asyncio.create_subprocess_exec( ruff_bin, "check", "--select=F821,F811,F401", filepath, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL, ) stdout, _ = ...
Q16: What does SessionController.begin() do automatically regarding unclosed sessions?
Category: Session Lifecycle & CLI Keywords: SessionController.begin(), SessionController.begin, automatically, regarding, unclosed Coverage: 38%
/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 Keywords: SessionController.validate_and_end(), five, checklist, items, validated Coverage: 71%
/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 Keywords: SessionController.end(), SessionController.end, handle, commits, tags Coverage: 40%
/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 Keywords: end(), validate_and_end(), difference, validate_and_end Coverage: 50%
/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 Keywords: shell, command, starts, session Coverage: 40%
/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 Keywords: handoff, data, next, session Coverage: 100%
/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 Keywords: context.md, end(), happens, sessions Coverage: 40%
/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 Keywords: session, CHANGELOG, updated Coverage: 60%
/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/_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...
Q24: What is the session git tag format?
Category: Session Lifecycle & CLI Keywords: session, format Coverage: 67%
/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 Keywords: mandatory, work, artifacts, choices, handoff Coverage: 71%
/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 Keywords: repository, pattern, abstraction, used, hide Coverage: 62%
/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/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/standards/orchestrata_specific_checklist.md:46-95— - [ ] Application Layer: Use-case orchestration, agent coordination - [ ] Infrastructure Layer: External system integrations, persistence, messaging - [ ] Interface Segregation: Ports define minimal, focused interfaces --- ## Domain Model Rules ### Entity Inheritance - [ ] All domain entities inherit fromDomainEntity(which inherits fromNode) - [ ] **Graph Serialization...
Q27: What are the two embedding backends in semantic_graph/embedders.py?
Category: Semantic Graph & RAG Keywords: semantic_graph/embedders.py, embedding, backends, semantic_graph, embedders.py Coverage: 60%
/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 Keywords: Memgraph, adapter, chunk, label, vector Coverage: 78%
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:496-545— async def link_chunks_to_modules(self) -> None: """Create Module nodes and BELONGS_TO edges from chunk metadata. Chunks must have amodulemetadata key. The operation is idempotent. """ def _link(driver: Any) -> None: with driver.session() as session: session.run( f""" MATCH (c:{self._chunk.../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— if entity.id in chunk.content or _any_string_property_in_text( entity.properties, chunk.content ): entities[entity.id] = entity ents, rels = await self.get_neighbors(entity.id, graph_depth) for e in ents: entities[e.id] = e for r...
Q29: What bug was fixed in _all_vector_indexes() regarding label strings?
Category: Semantic Graph & RAG Keywords: _all_vector_indexes(), fixed, _all_vector_indexes, regarding, label Coverage: 33%
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:316-365— "CREATE VECTOR INDEX " f"{self._config.vector_index_name} ON :{label}(embedding) " "WITH CONFIG {" f"dimension: {dimension}, " f"capacity: {self._config.vector_index_capacity}, " f"metric: '{self._config.vector_index_metric}'" "}" ) with driver.session() as session: .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:271-320— async def _load_vector_index_dimension(self) -> None: """Set_vector_index_dimensionfrom an existing index, if any.""" info = await self._vector_index_info() if info is not None: self._vector_index_dimension = info["dimension"] async def _vector_index_info(self) -> dict[str, Any] | None: """Return metadata for the configured vector index, or.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:361-410— "CALL vector_search.show_index_info() " "YIELD index_name, label, property, dimension " "RETURN index_name, label, property, dimension" ) return [ { "name": record["index_name"], "label": record["label"].lstrip(":"), "property": reco...
Q30: What happens when a dimension mismatch is detected on an existing vector index?
Category: Semantic Graph & RAG Keywords: happens, dimension, mismatch, detected, existing Coverage: 44%
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:271-320— async def _load_vector_index_dimension(self) -> None: """Set_vector_index_dimensionfrom an existing index, if any.""" info = await self._vector_index_info() if info is not None: self._vector_index_dimension = info["dimension"] async def _vector_index_info(self) -> dict[str, Any] | None: """Return metadata for the configured vector index, or.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:226-275— rel_id = rel.get("id") if rel_id and rel_id not in relationships: relationships[rel_id] = _rel_to_relationship( rel, source_id=start_node["id"], target_id=end_node["id"], ) return (list(entities.values()), lis.../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...
Q31: How does the HA controller discover the current Memgraph MAIN node?
Category: Semantic Graph & RAG Keywords: controller, discover, current, Memgraph, MAIN Coverage: 86%
/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/adapters/memgraph.py:91-140— Priority: 1. HA controller's /main endpoint. 2. Configured fallback URIs. 3. Single configured URI. """ if self._config.ha_controller_url: try: async with httpx.AsyncClient(timeout=5.0) as client: response = await client.get( f"{self._config.ha_controller_url}/main" ...
Q32: What is the vector index name, dimension, and metric for the orchestrata-ultimate Codestral chunks?
Category: Semantic Graph & RAG Keywords: vector, index, name, dimension, metric Coverage: 70%
/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.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:361-410— "CALL vector_search.show_index_info() " "YIELD index_name, label, property, dimension " "RETURN index_name, label, property, dimension" ) return [ { "name": record["index_name"], "label": record["label"].lstrip(":"), "property": reco...
Q33: What reranker model is used for the orchestrata-ultimate semantic graph?
Category: Semantic Graph & RAG Keywords: reranker, model, used, orchestrata, ultimate Coverage: 88%
/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/semantic_graph/ingestion.py:226-275— Args: query: Raw query text. top_k: Maximum number of results. Returns: Tuple of (chunks, similarities). Similarities are returned in the same order as chunks. """ embeddings = await self._embedder.embed([query], prompt_type="query") if not embeddings: return ([], []) candidate_k = max(top_k, s.../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...
Q34: How many chunks were ingested for the orchestrata-ultimate project?
Category: Semantic Graph & RAG Keywords: chunks, were, ingested, orchestrata, ultimate Coverage: 57%
/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/standards/orchestrata_specific_checklist.md:361-367— Version: 1.0.0 Created: 2026-04-09 Last Updated: 2026-04-09 Status: Draft Author: Mohamed Applies To: orchestrata-ultimate project only/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 Keywords: property, tags, chunk, multi, support Coverage: 60%
/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 Keywords: Embedder, protocol, interface Coverage: 75%
/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/ingestion.py:136-185— for path in directory.glob(pattern): if not path.is_file(): continue chunks.extend(self.chunk_file(path, project_id)) return chunks class SemanticGraphIngester: """Ingest source-code chunks into a semantic graph using an embedder.""" def init( self, graph: Any, embedder: Embedder, chunker: CodeChunker...
Q37: How does OpenRouterEmbedder handle out-of-order provider responses?
Category: Semantic Graph & RAG Keywords: OpenRouterEmbedder, handle, order, provider, responses Coverage: 67%
/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 Keywords: SentenceTransformerEmbedder, loaded, lazily Coverage: 100%
/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 Keywords: default, Qwen, embedding, model, dimension Coverage: 83%
/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 Keywords: ingestion, script, verify, retrieval, quality Coverage: 0%
/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 Keywords: pytest-xdist, pytest, xdist, banned Coverage: 80%
/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 Keywords: supported, test, runners Coverage: 25%
/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.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/standards/code-review-checklist.md:46-80— - TOO_MANY_ARGUMENTS: Functions with >5 arguments ### Best Practices/Error Handling - BARE_EXCEPT: Bare 'except:' clauses - MUTABLE_DEFAULT_ARGUMENT: Mutable defaults like [] or {} ## 4. Checklist-Based Reviews ### Design - CLASS_TOO_MANY_METHODS: Classes with >10 methods (single responsibility) - METHOD_NO_SELF_ACCESS: Methods not using self/cls (should be static or module-level) - ABSTRACT_...
Q43: What is the minimum per-file coverage required by the Definition of Done?
Category: Testing & Quality Keywords: minimum, file, coverage, required, Definition Coverage: 57%
/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 Keywords: os.environ, forbid, regarding, ported, tests Coverage: 50%
/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 Keywords: fixtures, function, scope, rather, module Coverage: 67%
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/standards/code-review-checklist.md:46-80— - TOO_MANY_ARGUMENTS: Functions with >5 arguments ### Best Practices/Error Handling - BARE_EXCEPT: Bare 'except:' clauses - MUTABLE_DEFAULT_ARGUMENT: Mutable defaults like [] or {} ## 4. Checklist-Based Reviews ### Design - CLASS_TOO_MANY_METHODS: Classes with >10 methods (single responsibility) - METHOD_NO_SELF_ACCESS: Methods not using self/cls (should be static or module-level) - ABSTRACT_.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/standards/architecture.md:1-29— # Orchestrata Ultimate Architecture Overview ## High-Level Design - Monorepo to Single Package Port: 122K LOC from 26 packages tosrc/orchestrata_ultimate/using Python 3.14+, uv manager. - Core Components: Agents (volks), CXP protocol, Providers (LLM adapters), Kernel (event bus, registry), Orchestrator (intent parsing), Knowledge Graph (Neo4j), Session/Memory management, Gateway (W.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/standards/Python Code Review Checklist — Python Guidelines 0.1.0 documentation.md:91-140— ## Types Have immutable types, tuple, frozenset, Enum, etc. been used in place of mutable types whenever possible? ## Functions Functions are pu...
Q46: What is the isolation fixture used for?
Category: Testing & Quality Keywords: isolation, fixture, used Coverage: 50%
/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/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...
Q47: How are Redis keys isolated per worker in tests?
Category: Testing & Quality Keywords: Redis, keys, isolated, worker, tests Coverage: 100%
/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/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/standards/orchestrata_specific_checklist.md:136-185— ### Graph Operations - [ ] Read-Model Separation: Complex queries use dedicated read models (not domain entities) - [ ] Graph Port Abstraction: Graph operations throughIGraphPort/IGraphQueryPort- [ ] Transaction Management: Graph operations within transactions - [ ] Consistency: Graph state consistent with domain model ### Performance - [ ] Indexing: Graph queries use...
Q48: What is the real-Prolog fixture pattern for test isolation?
Category: Testing & Quality Keywords: real, Prolog, fixture, pattern, test Coverage: 71%
/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 Keywords: mocks, stubs, fakes, monkeypatch, forbidden Coverage: 60%
/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 Keywords: four, blocking, safety, gates, tests Coverage: 43%
/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/standards/unified_code_review_standards.md:361-410— - [ ] All checks run in CI pipeline - [ ] Pre-commit hooks configured - [ ] Code review checklist integrated into PR process --- ## Auto-Fixers (SPEC-031 QA Volk) The QA Volk (U:Qa) includes 13 deterministic auto-fixers that correct known LLM-generated defects before any review gates run: ### aioredis Migration 1. aioredis import rewrite -import aioredis→ `import redis.asyncio as ai.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/standards/unified_code_review_standards.md:406-455— - Failure → bounce to U:Cd with label "GATE 2 FAIL: Lint/style errors" - Gate 3 and subsequent pipeline phases skipped on failure ### Gate 3: Checklist - Deterministic structural checks: - Traceability decorators present - Import order correct (stdlib → third-party → local) - Public functions have docstrings - No relative imports - Failure → bounce to U:Cd with label "GATE 3 FAIL: Chec...
Q51: What is the source monorepo path and how many packages does it contain?
Category: Porting & AST Keywords: source, monorepo, path, packages, contain Coverage: 50%
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/standards/architecture.md:1-29— # Orchestrata Ultimate Architecture Overview ## High-Level Design - Monorepo to Single Package Port: 122K LOC from 26 packages tosrc/orchestrata_ultimate/using Python 3.14+, uv manager. - Core Components: Agents (volks), CXP protocol, Providers (LLM adapters), Kernel (event bus, registry), Orchestrator (intent parsing), Knowledge Graph (Neo4j), Session/Memory management, Gateway (W.../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 ...
Q52: What is the target package path for the ported code?
Category: Porting & AST Keywords: target, package, path, ported, code Coverage: 67%
/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/standards/architecture.md:1-29— # Orchestrata Ultimate Architecture Overview ## High-Level Design - Monorepo to Single Package Port: 122K LOC from 26 packages tosrc/orchestrata_ultimate/using Python 3.14+, uv manager. - Core Components: Agents (volks), CXP protocol, Providers (LLM adapters), Kernel (event bus, registry), Orchestrator (intent parsing), Knowledge Graph (Neo4j), Session/Memory management, Gateway (W.../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...
Q53: What is the purpose of scripts/serialize_codebase_ast.py?
Category: Porting & AST Keywords: scripts/serialize_codebase_ast.py, purpose, scripts, serialize_codebase_ast.py Coverage: 50%
/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:406-455— qn = k break if qn not in ig: return None children = [k for k, parents in ig.items() if qn in parents] return { "class": qn, "parents": ig.get(qn, []), "children": children, "child_count": len(children), } def module_deps(self, module_path: str) -> list[str]: """Return m.../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...
Q54: How many classes are in the orchestrata source AST?
Category: Porting & AST Keywords: classes, orchestrata, source Coverage: 75%
/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 Keywords: three, pass, class, pairing, algorithm Coverage: 44%
/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: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): ...
Q56: What is the readiness classification for ported classes?
Category: Porting & AST Keywords: readiness, classification, ported, classes Coverage: 40%
/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 Keywords: orchestrata.core.interfaces, import, translation Coverage: 62%
/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
Keywords: orchestrata.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/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 Keywords: dynamic, string, literal, imports, handled Coverage: 29%
/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 Keywords: os.environ["REDIS_HOST"], default, value, os.environ, REDIS_HOST Coverage: 67%
/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/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/standards/orchestrata_specific_checklist.md:316-365— # core/application/user_service.py class UserService: def init(self, cache_port: ICachePort): self.cache = cache_port def cache_user(self, user: User): self.cache.set(f"user:{user.id}", user.to_json())### Violation: Hardcoded Configurationpython # ❌ WRONG - Hardcoded values REDIS_HOST = "localhost" # Hardcoded! redis = Redis(host=REDIS_HOST)
Q61: What are the connection env vars for Neo4j?
Category: Knowledge Graph & Neo4j Keywords: connection, vars, Neo4j Coverage: 67%
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/harness/history.py:91-140— payload = json.dumps(records, indent=2, ensure_ascii=False) # Atomic write: write to temp file in same directory, then rename fd, tmp_path = tempfile.mkstemp( dir=self.history_dir, prefix=f".{month_key}", suffix=".tmp" ) try: with os.fdopen(fd, "w", encoding="utf-8") as fh: fh.write(payload) os.replace(tmp_path.../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...
Q62: What is the Neo4jSyncer class responsible for?
Category: Knowledge Graph & Neo4j Keywords: Neo4jSyncer, class, responsible Coverage: 50%
/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 Keywords: node, types, relationship, Neo4j, sync Coverage: 62%
/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 Keywords: nodes, first, edges, second, pattern Coverage: 75%
/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 Keywords: CREATE, test_id, prevent, Neo4j Coverage: 33%
/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/standards.py:181-217— No mutable defaults (list, dict, set). 8. LINE_LENGTH: Max 80 characters per line. 4-space indent. 9. PYTHON 3.13/3.14: -asyncio.AwaitableREMOVED -> usecollections.abc.Awaitable-asyncio.get_event_loop()DEPRECATED in sync context -datetime.utcnow()DEPRECATED ->datetime.now(timezone.utc)-contextlib.AsyncContextManagerREMOVED in 3.14 10. TESTS: No mocking. No mo.../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, ) ...
Q66: What command verifies knowledge graph connectivity?
Category: Knowledge Graph & Neo4j Keywords: command, verifies, knowledge, graph, connectivity Coverage: 83%
/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 Keywords: single, source, truth, service, ports Coverage: 83%
/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/llm/config.py:136-185— # Load profiles at import time _PROFILES: dict[str, dict[str, dict[str, str]]] = _load_profiles() def _resolve_profile_name() -> str: """Determine which profile to use. ChecksSPIKE_PROFILEenv var, falls back to"default". """ return os.environ.get("SPIKE_PROFILE", "default") def _get_active_profile() -> str: """Return the active profile name, falling back to '...
Q68: What is the default Neo4j URI fallback if env var is missing?
Category: Knowledge Graph & Neo4j Keywords: default, Neo4j, fallback, missing Coverage: 75%
/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 Keywords: test_id, based, cleanups, performed, Neo4j Coverage: 40%
/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/standards.py:181-217— No mutable defaults (list, dict, set). 8. LINE_LENGTH: Max 80 characters per line. 4-space indent. 9. PYTHON 3.13/3.14: -asyncio.AwaitableREMOVED -> usecollections.abc.Awaitable-asyncio.get_event_loop()DEPRECATED in sync context -datetime.utcnow()DEPRECATED ->datetime.now(timezone.utc)-contextlib.AsyncContextManagerREMOVED in 3.14 10. TESTS: No mocking. No mo.../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...
Q70: What is the relationship between sync_ast_to_neo4j.py and knowledge_graph/sync.py?
Category: Knowledge Graph & Neo4j Keywords: sync_ast_to_neo4j.py, knowledge_graph/sync.py, relationship, knowledge_graph, sync.py Coverage: 50%
/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 Keywords: BaseVolk, renamed, class, target Coverage: 33%
/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/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.../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": ["...
Q72: What is the base class for agents: Pydantic BaseModel or dataclass?
Category: Agents & Volks Keywords: base, class, agents, Pydantic, BaseModel Coverage: 100%
/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 Keywords: role, volks, ported Coverage: 40%
/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 Keywords: orchestrata.volks.analyst.volk, import, translation Coverage: 71%
/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 Keywords: pattern, replaces, interfaces, target Coverage: 83%
/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/standards/orchestrata_specific_checklist.md:271-320— 3. Implement Adapters: Create concrete implementations inadapters/4. Add Application Logic: Orchestrate incore/application/5. Write Tests: Follow zero-mocking policy with real infrastructure ### For Code Review 1. Check Architecture First: Verify dependency direction and layer separation 2. Validate Domain Purity: Ensure domain models have no infrastructure referenc.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/core/patterns/observer.py:1-50— """Generic, thread-safe Observer/Observable pattern (GoF) — sync and async. Layer-1 reusable infrastructure, sibling tocore.patterns.error_handling. The event/payload type is generic (TypeVar("T")) so this module has no opinion about what it carries; its first real consumer is expected to be a live test-run event stream (anObservable[TestEvent]feeding an NDJSON sink and/or a live UI),...
Q76: What is the CXPMixin used for in BaseAgent?
Category: Agents & Volks Keywords: CXPMixin, BaseAgent, used Coverage: 50%
/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/integrations/cheetahclaws/__init__.py:1-50— """CheetahClaws Integration — UltimateTUI wiring CC commands to Orchestrata agents. SPEC-050 FR-001 through FR-015. Architecture: CC Slash Command → UltimateTUI → CommandRouter → CXP Dispatcher │ ┌─────────────────────────────────────────┤ │ │ ▼ ▼ Gov...
Q77: What is the agents/lifecycle.py module responsible for?
Category: Agents & Volks Keywords: agents/lifecycle.py, agents, lifecycle.py, module, responsible Coverage: 40%
/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 Keywords: agents, created, configured, according, sample Coverage: 29%
/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 Keywords: agents/swarm, agents/volks, difference, agents, swarm Coverage: 71%
/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/agents/volks/memory_context.py:1-50— """Run-scoped memory-context provider registry. The Memory Engine (SPEC-010) stored every conversation turn but nothing ever retrieved it — volks received zero memory context. This module closes that gap the same way tracing did: a memory engine is bound per run bythread_id(the sync-router bridges volks onto worker threads, so a ContextVar would not survive; volks look the provider up ...
Q80: What design decision replaced the 26-package monorepo with a single package?
Category: Agents & Volks Keywords: design, decision, replaced, package, monorepo Coverage: 67%
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/standards/architecture.md:1-29— # Orchestrata Ultimate Architecture Overview ## High-Level Design - Monorepo to Single Package Port: 122K LOC from 26 packages tosrc/orchestrata_ultimate/using Python 3.14+, uv manager. - Core Components: Agents (volks), CXP protocol, Providers (LLM adapters), Kernel (event bus, registry), Orchestrator (intent parsing), Knowledge Graph (Neo4j), Session/Memory management, Gateway (W.../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...
Q81: What does CXP stand for and what are its main components?
Category: CXP & Protocols Keywords: stand, main, components Coverage: 17%
/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 Keywords: CXPPacket Coverage: 50%
/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 Keywords: protobuf, definition, located Coverage: 25%
/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 Keywords: role, serializer Coverage: 33%
/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 Keywords: dispatcher, route, packets Coverage: 75%
/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 Keywords: Incus, container, runs, Redis Coverage: 80%
/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/concierge/volk.py:46-95— integration points? D5. QUALITY ATTRIBUTES: Performance targets, error handling strategy, \ testing requirements, logging/observability needs. D6. DEPLOYMENT: Default is Incus containers (Linux). If the human says \ Docker, use Docker. If unspecified, use Incus. Never assume Docker. D7. CONSTRAINTS: Budget, timeline, team size, existing codebase to \ integrate with, regu.../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") ...
Q87: What Incus container runs Neo4j?
Category: Infrastructure & Deployment Keywords: Incus, container, runs, Neo4j Coverage: 100%
/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/agents/volks/concierge/volk.py:46-95— integration points? D5. QUALITY ATTRIBUTES: Performance targets, error handling strategy, \ testing requirements, logging/observability needs. D6. DEPLOYMENT: Default is Incus containers (Linux). If the human says \ Docker, use Docker. If unspecified, use Incus. Never assume Docker. D7. CONSTRAINTS: Budget, timeline, team size, existing codebase to \ integrate with, regu...
Q88: What command starts all four test services in parallel?
Category: Infrastructure & Deployment Keywords: command, starts, four, test, services Coverage: 43%
/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 Keywords: check, service, status, failing, services Coverage: 43%
/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/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.../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...
Q90: What is the Memgraph HA cluster composed of?
Category: Infrastructure & Deployment Keywords: Memgraph, cluster, composed Coverage: 50%
/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 Keywords: Memgraph, controller Coverage: 67%
/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).../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/ha_controller_cli.py:46-95— if len(parts) < 2: raise ValueError(f"Invalid node spec: {value}") bolt_uri = parts[0] replication_spec = parts[1] if ":" in replication_spec: replication_host, replication_port = replication_spec.rsplit(":", 1) replication_port = int(replication_port) else: replication_host = replication_spec replication_port = 10000 return NodeConf...
Q92: Why did the standalone memgraph-db container need to be restarted during this work?
Category: Infrastructure & Deployment Keywords: memgraph-db, standalone, memgraph, container, need Coverage: 38%
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:496-545— async def link_chunks_to_modules(self) -> None: """Create Module nodes and BELONGS_TO edges from chunk metadata. Chunks must have amodulemetadata key. The operation is idempotent. """ def _link(driver: Any) -> None: with driver.session() as session: session.run( f""" MATCH (c:{self._chunk.../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py:316-365— "CREATE VECTOR INDEX " f"{self._config.vector_index_name} ON :{label}(embedding) " "WITH CONFIG {" f"dimension: {dimension}, " f"capacity: {self._config.vector_index_capacity}, " f"metric: '{self._config.vector_index_metric}'" "}" ) with driver.session() as session: .../home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/knowledge_graph/sync.py:316-365— "SET c.name = r.name, c.qualified_name = r.qualified_name, " "c.module_path = r.module_path, c.file_path = r.file_path, " "c.line = r.line, c.is_abstract = r.is_abstract, " "c.is_dataclass = r.is_dataclass, c.is_pydantic = r.is_pydantic, " "c.is_protocol = r.is_protocol, c.is_enum = r.is_enum, " "c.bases = r.bases",...
Q93: What command launches a Memgraph container from the project?
Category: Infrastructure & Deployment Keywords: command, launches, Memgraph, container Coverage: 57%
/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:46-95— if len(parts) < 2: raise ValueError(f"Invalid node spec: {value}") bolt_uri = parts[0] replication_spec = parts[1] if ":" in replication_spec: replication_host, replication_port = replication_spec.rsplit(":", 1) replication_port = int(replication_port) else: replication_host = replication_spec replication_port = 10000 return NodeConf.../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...
Q94: How is the wiki.js instance accessed?
Category: Infrastructure & Deployment Keywords: wiki.js, instance, accessed Coverage: 33%
/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/kernel/policy_evaluator.py:721-770— return PermissionLevel.DENY, "Permission denied by policy" elif level_str == "require_approval": return PermissionLevel.REQUIRE_APPROVAL, "Approval required" elif level_str == "escalate": return PermissionLevel.ESCALATE, "Escalation required" else: return PermissionLevel.ALLOW, "Permission granted" else:...
Q95: What is the wiki graph indexer built on?
Category: Infrastructure & Deployment Keywords: wiki, graph, indexer, built Coverage: 40%
/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:451-500— if entity.id in chunk.content or _any_string_property_in_text( entity.properties, chunk.content ): entities[entity.id] = entity ents, rels = await self.get_neighbors(entity.id, graph_depth) for e in ents: entities[e.id] = e for r.../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...
Q96: What is the rule about silent exception handlers?
Category: Code Standards & Patterns Keywords: silent, exception, handlers Coverage: 17%
/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/standards/unified_code_review_standards.md:181-230— - [ ] No mutable default arguments (def f(x=[]):); useNoneand initialize inside function - [ ] Keyword-only arguments (*) used where appropriate - [ ] Positional-only arguments (/) used where appropriate - [ ] Variable arguments (*args,**kwargs) used appropriately --- ## Classes & Objects ### Class Design - [ ] DESIGN: Every class has single well-defined purpose (no mixing...
Q97: Why is os.environ["X"] forbidden in favor of os.environ.get("X", default)?
Category: Code Standards & Patterns Keywords: os.environ["X"], os.environ.get("X", default), os.environ, forbidden, favor Coverage: 60%
/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/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: ...
Q98: What is the factory pattern used for external services?
Category: Code Standards & Patterns Keywords: factory, pattern, used, external, services Coverage: 83%
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/standards/Python Code Review Checklist — Python Guidelines 0.1.0 documentation.md:136-151— - A factory class or function at module level is used for complex class construction (see Design Patterns) to achieve composition - Classes are not dynamically created from strings except where forward reference requires this ## Design 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/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...
Q99: What is the UUID-prefixed resource pattern for tests?
Category: Code Standards & Patterns Keywords: UUID, prefixed, resource, pattern, tests Coverage: 50%
/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/cxp/constants.py:46-84— PROTOCOL_NAME: str = "CXP" # ---------- Validation patterns ---------- UUID_PATTERN: str = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" UUID_REGEX = re.compile(UUID_PATTERN, re.IGNORECASE) SEMVER_PATTERN: str = r"^\d+.\d+.\d+$" SEMVER_REGEX = re.compile(r"^\d+.\d+.\d+$") VOLK_ALIAS_PATTERN: str = r"^U:[A-Z][a-z]$" VOLK_ALIAS_REGEX = re.compile(VOLK_ALIAS_PATTERN) ...
Q100: What is the project's policy on dismissing test failures?
Category: Code Standards & Patterns Keywords: policy, dismissing, test, failures Coverage: 67%
/home/mohamed/workspace/orchestrata-ultimate-architecture/src/orchestrata_ultimate/standards/unified_code_review_standards.md:406-455— - Failure → bounce to U:Cd with label "GATE 2 FAIL: Lint/style errors" - Gate 3 and subsequent pipeline phases skipped on failure ### Gate 3: Checklist - Deterministic structural checks: - Traceability decorators present - Import order correct (stdlib → third-party → local) - Public functions have docstrings - No relative imports - Failure → bounce to U:Cd with label "GATE 3 FAIL: Chec.../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/standards/orchestrata_specific_checklist.md:226-275— - [ ] SPEC-000 Coverage: All SPEC-000 use cases represented in domain model - [ ] SPEC-031 QA: QA Volk rules enforced (auto-fixers, review gates, zero mocking) - [ ] SPEC-100 Logging: Structured logging, exception hierarchy, distributed tracing ### Code Review Pipeline - [ ] Three Gates: Syntax → Lint → Checklist gates pass before LLM review - [ ] Auto-Fixers: 13 auto-fixer...