[Compyle] Task cmgos18vg0001qyic0qoozodb updates for Ordo#37
Open
coderkrp wants to merge 8 commits into
Open
Conversation
added 8 commits
October 13, 2025 07:42
Contributor
WalkthroughAdds a multi-broker RequestOrchestrator with standardized results and session validation, new unified API models, session validation helper, FastAPI v1 endpoints for portfolio and orders plus auth login flows, integration/unit tests, and a local verification script. Implements concurrency, timeouts, error mapping, and overall status aggregation. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant C as Client
participant API as FastAPI /api/v1
participant OR as RequestOrchestrator
participant SM as SessionManager
participant AR as AdapterRegistry
participant B1 as BrokerAdapter A
participant B2 as BrokerAdapter B
C->>API: GET /api/v1/portfolio?brokers=[A,B]
API->>OR: get_portfolio(broker_ids, context)
activate OR
OR->>SM: ensure_valid_session(A, account_id)
SM-->>OR: SessionCheckResult(valid/expired)
OR->>SM: ensure_valid_session(B, account_id)
SM-->>OR: SessionCheckResult(valid/expired)
OR->>AR: instantiate adapters for A,B
par Parallel broker calls
OR->>B1: get_portfolio(session_data) with timeout
B1-->>OR: payload or error
OR->>B2: get_portfolio(session_data) with timeout
B2-->>OR: payload or error
end
OR-->>API: UnifiedResponse(results[], overall_status, correlation_id, elapsed_ms)
deactivate OR
API-->>C: 200 OK + UnifiedResponse
sequenceDiagram
autonumber
participant C as Client
participant API as FastAPI /api/v1
participant OR as RequestOrchestrator
participant SM as SessionManager
participant AR as AdapterRegistry
participant BA as BrokerAdapter
C->>API: POST /api/v1/orders {brokers[], order_data}
API->>OR: place_order(order_data, broker_ids, context)
OR->>SM: ensure_valid_session(broker, account_id) for each
OR->>AR: instantiate adapters
par Parallel order placement
OR->>BA: place_order(order_data, session_data) with timeout
BA-->>OR: result or error
end
OR-->>API: UnifiedResponse
API-->>C: 200 OK + UnifiedResponse
sequenceDiagram
autonumber
participant C as Client
participant API as FastAPI /auth
participant AR as AdapterRegistry
participant BA as BrokerAdapter
C->>API: POST /login/initiate {broker_id, ...}
API->>AR: get adapter(broker_id)
AR-->>API: BrokerAdapter
API->>BA: initiate_login(...)
BA-->>API: challenge/redirect
API-->>C: 200 OK / payload
C->>API: POST /login/complete {broker_id, ...}
API->>AR: get adapter(broker_id)
API->>BA: complete_login(...)
BA-->>API: session tokens
API-->>C: 200 OK / tokens
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Poem
Pre-merge checks and finishing touches❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Comment |
Owner
Author
|
@coderabbitai review |
Contributor
✅ Actions performedReview triggered.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 17
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/ordo/security/session.py (2)
35-44: Handle decrypt failures gracefullyget_session will raise if the stored value can’t be decrypted (e.g., corrupted state). Catch InvalidToken and return None.
from cryptography.fernet import Fernet +from cryptography.fernet import InvalidToken @@ - decrypted_value = self._fernet.decrypt(encrypted_value).decode() - return decrypted_value + try: + return self._fernet.decrypt(encrypted_value).decode() + except InvalidToken: + return None
23-33: Namespace sessions by account_id; ensure_valid_session ignores account_idCurrent keys are broker-scoped only. In multi-account setups, tokens collide. Include account_id in namespacing and use it when reading/writing tokens.
- def _get_namespaced_key(self, broker_id: str, key: str) -> str: + def _get_namespaced_key(self, broker_id: str, key: str, account_id: str | None = None) -> str: """Creates a namespaced key to prevent collisions.""" - return f"{broker_id}:{key}" + return f"{broker_id}:{account_id or '_'}:{key}" @@ - def set_session(self, broker_id: str, key: str, value: str): + def set_session(self, broker_id: str, key: str, value: str, account_id: str | None = None): @@ - namespaced_key = self._get_namespaced_key(broker_id, key) + namespaced_key = self._get_namespaced_key(broker_id, key, account_id) @@ - def get_session(self, broker_id: str, key: str) -> str | None: + def get_session(self, broker_id: str, key: str, account_id: str | None = None) -> str | None: @@ - namespaced_key = self._get_namespaced_key(broker_id, key) + namespaced_key = self._get_namespaced_key(broker_id, key, account_id) @@ - session_token = self.get_session(broker_id, "access_token") + session_token = self.get_session(broker_id, "access_token", account_id)Also, the docstring mentions statuses “refresh_failed” and “unsupported” but those aren’t reachable. Either implement refresh or drop the unused statuses from the docstring.
Also applies to: 35-45, 46-74
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (7)
src/ordo/core/orchestrator.py(1 hunks)src/ordo/main.py(3 hunks)src/ordo/models/api/unified.py(1 hunks)src/ordo/security/session.py(2 hunks)tests/core/test_api_integration.py(1 hunks)tests/core/test_orchestrator.py(1 hunks)verify_implementation.py(1 hunks)
🔇 Additional comments (3)
tests/core/test_orchestrator.py (1)
112-113: Verify orchestrator default timeouts expected by the testAssertions per_broker_timeout == 8 and global_timeout == 12 assume specific defaults. Confirm orchestrator defaults match; otherwise adjust test or implementation.
src/ordo/security/session.py (1)
17-21: Validate and explicitly handle invalid Fernet keysWrap the Fernet instantiation in a try/except and raise a clear ValueError if the key isn’t a 32-byte URL-safe base64 string. For example:
def __init__(self, secret_key: str): if not secret_key: raise ValueError("SECRET_KEY must be provided for session management.") - self._fernet = Fernet(secret_key.encode()) + try: + self._fernet = Fernet(secret_key.encode()) + except Exception as e: + raise ValueError( + "SECRET_KEY must be a 32-byte URL-safe base64 key (use Fernet.generate_key())." + ) from eConfirm how SECRET_KEY is sourced and formatted.
src/ordo/core/orchestrator.py (1)
117-121: Verify propagation of refreshed session data to adaptersIf ensure_valid_session refreshes tokens, the operation uses context['session_data'] (pre-refresh). Ensure the latest session payload is passed to adapters.
Would you prefer to:
- have ensure_valid_session return updated session_data and use it directly, or
- fetch session_data from SessionManager post-validation per broker/account?
I can propose a concrete change once you confirm the SessionManager API.
Also applies to: 199-204, 276-279
Comment on lines
+14
to
+20
| class SessionCheckResult: | ||
| """Result of session validation check.""" | ||
|
|
||
| def __init__(self, status: str, message: str = ""): | ||
| self.status = status # "valid", "expired", "refresh_failed", "unsupported" | ||
| self.message = message | ||
|
|
Contributor
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Remove duplicate/unreferenced SessionCheckResult
This local class isn’t used and likely duplicates the canonical type in ordo.security.session. Keep a single source of truth.
-class SessionCheckResult:
- """Result of session validation check."""
-
- def __init__(self, status: str, message: str = ""):
- self.status = status # "valid", "expired", "refresh_failed", "unsupported"
- self.message = message
+# Session validation result type should be defined in ordo.security.session.
+# Import and use that canonical type there instead of redefining locally.Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In src/ordo/core/orchestrator.py around lines 14 to 20, remove the local
SessionCheckResult class (it's duplicate/unreferenced) and instead import and
use the canonical type from ordo.security.session; delete the class definition,
add the appropriate import (e.g., from ordo.security.session import
SessionCheckResult) at the top, and update any local references to use the
imported symbol.
Comment on lines
+36
to
+56
| def __init__( | ||
| self, | ||
| session_manager: SessionManager, | ||
| adapter_registry: Dict[str, Type[IBrokerAdapter]], | ||
| per_broker_timeout: int = 8, | ||
| global_timeout: int = 12, | ||
| ): | ||
| """ | ||
| Initialize RequestOrchestrator. | ||
|
|
||
| Args: | ||
| session_manager: SessionManager instance for session validation | ||
| adapter_registry: Dict mapping broker_id to adapter class | ||
| per_broker_timeout: Timeout for individual broker operations (default: 8s) | ||
| global_timeout: Timeout for entire orchestration (default: 12s) | ||
| """ | ||
| self.session_manager = session_manager | ||
| self.adapter_registry = adapter_registry | ||
| self.per_broker_timeout = per_broker_timeout | ||
| self.global_timeout = global_timeout | ||
|
|
Contributor
There was a problem hiding this comment.
Implement and enforce global_timeout; cancel pending tasks and map to TIMEOUT
global_timeout is defined but never used. Wrap orchestration in wait_for, cancel pending tasks on timeout, and convert global timeouts to BrokerResult with code TIMEOUT.
Apply these diffs.
Get portfolio: create tasks and enforce global timeout
@@
- # Create tasks for concurrent execution
- tasks = []
- for broker_id in broker_ids:
+ # Create tasks for concurrent execution
+ tasks_by_broker = {}
+ for broker_id in broker_ids:
# Create operation function for this broker
async def operation(bid=broker_id):
adapter_class = self.adapter_registry[bid]
adapter = adapter_class()
session_data = context.get("session_data", {}).get(bid, {})
return await adapter.get_portfolio(session_data)
- # Add to tasks list
- task = self._execute_for_broker(broker_id, operation, context)
- tasks.append(task)
+ # Add to tasks map
+ tasks_by_broker[broker_id] = asyncio.create_task(
+ self._execute_for_broker(broker_id, operation, context)
+ )
@@
- # Execute all tasks concurrently using asyncio.gather
- # return_exceptions=True ensures one failure doesn't stop others
- results = await asyncio.gather(*tasks, return_exceptions=True)
+ # Execute all tasks concurrently with a global timeout
+ try:
+ results = await asyncio.wait_for(
+ asyncio.gather(
+ *tasks_by_broker.values(), return_exceptions=True
+ ),
+ timeout=self.global_timeout,
+ )
+ except asyncio.TimeoutError:
+ # Cancel pending tasks and synthesize timeout results
+ for t in tasks_by_broker.values():
+ if not t.done():
+ t.cancel()
+ results = [
+ (t.result() if t.done() else asyncio.TimeoutError())
+ for t in tasks_by_broker.values()
+ ]
@@
- for i, result in enumerate(results):
+ for i, result in enumerate(results):
if isinstance(result, BrokerResult):
broker_results.append(result)
elif isinstance(result, Exception):
- # Handle exceptions that weren't caught by _execute_for_broker
- broker_results.append(
- BrokerResult(
- broker_id=broker_ids[i],
- status="failed",
- code="INTERNAL_ERROR",
- message=str(result),
- payload=None,
- latency_ms=0,
- )
- )
+ # Map global timeout separately
+ if isinstance(result, asyncio.TimeoutError):
+ broker_results.append(
+ BrokerResult(
+ broker_id=broker_ids[i],
+ status="failed",
+ code="TIMEOUT",
+ message=f"Global timeout after {self.global_timeout}s",
+ payload=None,
+ latency_ms=0,
+ )
+ )
+ else:
+ broker_results.append(
+ BrokerResult(
+ broker_id=broker_ids[i],
+ status="failed",
+ code="INTERNAL_ERROR",
+ message=str(result),
+ payload=None,
+ latency_ms=0,
+ )
+ )Place order: create tasks and enforce global timeout
@@
- # Create tasks for concurrent execution
- tasks = []
- for broker_id in broker_ids:
+ # Create tasks for concurrent execution
+ tasks_by_broker = {}
+ for broker_id in broker_ids:
# Create operation function for this broker
async def operation(bid=broker_id):
adapter_class = self.adapter_registry[bid]
adapter = adapter_class()
session_data = context.get("session_data", {}).get(bid, {})
- # Note: Actual order placement method needs to be added to IBrokerAdapter
- # For now, we'll use a placeholder
- return OrderResponse(order_id=f"{bid}_order_123", status="pending")
+ return await adapter.place_order(order_data, session_data) # async
@@
- # Add to tasks list
- task = self._execute_for_broker(broker_id, operation, context)
- tasks.append(task)
+ # Add to tasks map
+ tasks_by_broker[broker_id] = asyncio.create_task(
+ self._execute_for_broker(broker_id, operation, context)
+ )
@@
- # Execute all tasks concurrently
- results = await asyncio.gather(*tasks, return_exceptions=True)
+ # Execute all tasks concurrently with a global timeout
+ try:
+ results = await asyncio.wait_for(
+ asyncio.gather(
+ *tasks_by_broker.values(), return_exceptions=True
+ ),
+ timeout=self.global_timeout,
+ )
+ except asyncio.TimeoutError:
+ for t in tasks_by_broker.values():
+ if not t.done():
+ t.cancel()
+ results = [
+ (t.result() if t.done() else asyncio.TimeoutError())
+ for t in tasks_by_broker.values()
+ ]
@@
- for i, result in enumerate(results):
+ for i, result in enumerate(results):
if isinstance(result, BrokerResult):
broker_results.append(result)
elif isinstance(result, Exception):
- broker_results.append(
- BrokerResult(
- broker_id=broker_ids[i],
- status="failed",
- code="INTERNAL_ERROR",
- message=str(result),
- payload=None,
- latency_ms=0,
- )
- )
+ if isinstance(result, asyncio.TimeoutError):
+ broker_results.append(
+ BrokerResult(
+ broker_id=broker_ids[i],
+ status="failed",
+ code="TIMEOUT",
+ message=f"Global timeout after {self.global_timeout}s",
+ payload=None,
+ latency_ms=0,
+ )
+ )
+ else:
+ broker_results.append(
+ BrokerResult(
+ broker_id=broker_ids[i],
+ status="failed",
+ code="INTERNAL_ERROR",
+ message=str(result),
+ payload=None,
+ latency_ms=0,
+ )
+ )Also applies to: 195-208, 209-213, 213-229, 272-287, 288-290, 292-306
🤖 Prompt for AI Agents
In src/ordo/core/orchestrator.py around lines 36-56 (and similarly for ranges
195-208, 209-213, 213-229, 272-287, 288-290, 292-306), global_timeout is defined
but unused; wrap the orchestration logic that gathers per-broker tasks in an
asyncio.wait_for using self.global_timeout, catch asyncio.TimeoutError, cancel
any still-pending tasks (call task.cancel() and await them safely), and for each
broker whose task was cancelled or whose wait_for timed out, produce a
BrokerResult with code TIMEOUT and an appropriate error message; ensure
successful and failed tasks still return/propagate their normal BrokerResult
values and that cancelled tasks are mapped into the results list before
returning from the orchestration methods (get_portfolio/place_order).
Comment on lines
+231
to
+254
| # Compute overall status | ||
| overall_status = self._compute_overall_status(broker_results) | ||
|
|
||
| # Calculate total elapsed time | ||
| elapsed_ms = int((time.time() - start_time) * 1000) | ||
|
|
||
| # Collect errors if any | ||
| errors = [] | ||
| for result in broker_results: | ||
| if result.status == "failed": | ||
| errors.append( | ||
| { | ||
| "broker_id": result.broker_id, | ||
| "code": result.code, | ||
| "message": result.message, | ||
| } | ||
| ) | ||
|
|
||
| return UnifiedResponse( | ||
| overall_status=overall_status, | ||
| results=broker_results, | ||
| elapsed_ms=elapsed_ms, | ||
| errors=errors if errors else None, | ||
| ) |
Contributor
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
Reduce duplication in result collation
The error aggregation and UnifiedResponse construction are duplicated across both methods. Extract a small helper to compute errors and build UnifiedResponse.
Also applies to: 314-331
🤖 Prompt for AI Agents
In src/ordo/core/orchestrator.py around lines 231-254 and 314-331, the logic
that collates broker_results into errors, computes elapsed_ms and constructs the
UnifiedResponse is duplicated; extract a small helper method (private on the
orchestrator class or a module-level function) that accepts broker_results,
overall_status and start_time (or elapsed_ms) and returns the UnifiedResponse:
inside the helper compute elapsed_ms, build the errors list by iterating
broker_results and collecting dicts for results with status == "failed" (set
errors to None if empty), and construct/return UnifiedResponse with
overall_status, results, elapsed_ms and errors; then replace the duplicated
blocks at both locations with a single call to this helper, ensuring
imports/typing are adjusted as needed.
Comment on lines
+276
to
+283
| async def operation(bid=broker_id): | ||
| adapter_class = self.adapter_registry[bid] | ||
| adapter = adapter_class() | ||
| session_data = context.get("session_data", {}).get(bid, {}) | ||
| # Note: Actual order placement method needs to be added to IBrokerAdapter | ||
| # For now, we'll use a placeholder | ||
| return OrderResponse(order_id=f"{bid}_order_123", status="pending") | ||
|
|
Contributor
There was a problem hiding this comment.
Order placement is a placeholder; not calling adapters
This silently returns a fake order response and never hits broker APIs.
Apply this diff to call the adapter (and fail fast if unsupported):
- async def operation(bid=broker_id):
+ async def operation(bid=broker_id):
adapter_class = self.adapter_registry[bid]
adapter = adapter_class()
session_data = context.get("session_data", {}).get(bid, {})
- # Note: Actual order placement method needs to be added to IBrokerAdapter
- # For now, we'll use a placeholder
- return OrderResponse(order_id=f"{bid}_order_123", status="pending")
+ if not hasattr(adapter, "place_order"):
+ raise NotImplementedError(f"Adapter {bid} does not support order placement")
+ return await adapter.place_order(order_data, session_data)Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In src/ordo/core/orchestrator.py around lines 276 to 283, the async operation
currently returns a hard-coded OrderResponse and never calls the broker adapter;
replace the placeholder with a real adapter invocation: instantiate the adapter,
retrieve session_data and order details from context, check that the adapter
implements an async place_order (or place_order) method and if not raise
NotImplementedError, then call and await adapter.place_order(session_data,
order_payload) (or await adapter.place_order_async(...)) and return its
OrderResponse; ensure any adapter exceptions are not swallowed (let them
propagate or re-raise) so failures fail fast.
| @@ -1,19 +1,55 @@ | |||
| from fastapi import FastAPI, APIRouter, HTTPException, status, Body | |||
| from fastapi import FastAPI, APIRouter, HTTPException, status, Body, Depends, Query | |||
Contributor
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
Cache DI providers to avoid per-request re-instantiation
Use lru_cache for SessionManager and RequestOrchestrator (pure, stable deps), reducing overhead.
-from fastapi import FastAPI, APIRouter, HTTPException, status, Body, Depends, Query
+from fastapi import FastAPI, APIRouter, HTTPException, status, Body, Depends, Query, Request
+from functools import lru_cache
@@
-def get_session_manager() -> SessionManager:
+@lru_cache()
+def get_session_manager() -> SessionManager:
"""Provide SessionManager instance."""
return SessionManager(secret_key=settings.SECRET_KEY)
@@
-def get_orchestrator() -> RequestOrchestrator:
+@lru_cache()
+def get_orchestrator() -> RequestOrchestrator:
"""Provide RequestOrchestrator instance with dependencies."""
session_manager = get_session_manager()
adapter_registry = {
"mock": MockAdapter,
"fyers": FyersAdapter,
"hdfc": HDFCAdapter,
}
return RequestOrchestrator(session_manager, adapter_registry)Also applies to: 24-27, 29-38
🤖 Prompt for AI Agents
In src/ordo/main.py around line 1 and also covering lines 24-27 and 29-38, the
dependency providers that construct SessionManager and RequestOrchestrator are
currently created per-request; change their factory functions to be cached by
decorating them with functools.lru_cache so they return a singleton-like
instance for the app lifetime (ensure the factories take only pure/stable
arguments), import lru_cache at the top, keep the FastAPI Depends usage but
point it to the now-cached provider functions, and remove any request-scoped
state from those providers so they are safe to reuse.
| import pytest | ||
| import asyncio | ||
| from unittest.mock import Mock, AsyncMock, MagicMock | ||
| from ordo.core.orchestrator import RequestOrchestrator, SessionCheckResult |
Contributor
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
Avoid re-defining SessionCheckResult in orchestrator; import from security.session
Keep a single definition to prevent type drift.
-from ordo.core.orchestrator import RequestOrchestrator, SessionCheckResult
+from ordo.core.orchestrator import RequestOrchestrator
+from ordo.security.session import SessionCheckResult📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| from ordo.core.orchestrator import RequestOrchestrator, SessionCheckResult | |
| from ordo.core.orchestrator import RequestOrchestrator | |
| from ordo.security.session import SessionCheckResult |
🤖 Prompt for AI Agents
In tests/core/test_orchestrator.py at line 7, the test imports
SessionCheckResult from ordo.core.orchestrator which re-defines that type;
update the import to pull SessionCheckResult from the canonical module
(ordo.core.security.session) so the test uses the single shared definition and
remove any duplicate local definitions if present; ensure the import path
matches the package layout and run tests to verify no type drift errors remain.
Comment on lines
+69
to
+86
| def mock_adapter_timeout(): | ||
| """Adapter that times out.""" | ||
|
|
||
| class MockTimeoutAdapter: | ||
| async def get_portfolio(self, session_data): | ||
| await asyncio.sleep(10) # Longer than default timeout | ||
| return Portfolio( | ||
| holdings=[], | ||
| funds=Funds( | ||
| available_balance=0.0, margin_used=0.0, total_balance=0.0 | ||
| ), | ||
| total_pnl=0.0, | ||
| total_day_pnl=0.0, | ||
| total_value=0.0, | ||
| ) | ||
|
|
||
| return MockTimeoutAdapter | ||
|
|
Contributor
There was a problem hiding this comment.
Timeout mock lacks place_order; current test will raise AttributeError, not TIMEOUT
MockTimeoutAdapter only implements get_portfolio, but test calls place_order. Add a timed-out place_order.
class MockTimeoutAdapter:
async def get_portfolio(self, session_data):
await asyncio.sleep(10) # Longer than default timeout
return Portfolio(
holdings=[],
funds=Funds(
available_balance=0.0, margin_used=0.0, total_balance=0.0
),
total_pnl=0.0,
total_day_pnl=0.0,
total_value=0.0,
)
+
+ async def place_order(self, order_data, session_data):
+ await asyncio.sleep(10) # Simulate per-broker timeout
+ return OrderResponse(status="success") # Will be masked by timeout logic📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def mock_adapter_timeout(): | |
| """Adapter that times out.""" | |
| class MockTimeoutAdapter: | |
| async def get_portfolio(self, session_data): | |
| await asyncio.sleep(10) # Longer than default timeout | |
| return Portfolio( | |
| holdings=[], | |
| funds=Funds( | |
| available_balance=0.0, margin_used=0.0, total_balance=0.0 | |
| ), | |
| total_pnl=0.0, | |
| total_day_pnl=0.0, | |
| total_value=0.0, | |
| ) | |
| return MockTimeoutAdapter | |
| def mock_adapter_timeout(): | |
| """Adapter that times out.""" | |
| class MockTimeoutAdapter: | |
| async def get_portfolio(self, session_data): | |
| await asyncio.sleep(10) # Longer than default timeout | |
| return Portfolio( | |
| holdings=[], | |
| funds=Funds( | |
| available_balance=0.0, margin_used=0.0, total_balance=0.0 | |
| ), | |
| total_pnl=0.0, | |
| total_day_pnl=0.0, | |
| total_value=0.0, | |
| ) | |
| async def place_order(self, order_data, session_data): | |
| await asyncio.sleep(10) # Simulate per-broker timeout | |
| return OrderResponse(status="success") # Will be masked by timeout logic | |
| return MockTimeoutAdapter |
🤖 Prompt for AI Agents
In tests/core/test_orchestrator.py around lines 69 to 86, the MockTimeoutAdapter
only implements get_portfolio so calls to place_order in the test raise
AttributeError instead of simulating a timeout; add an async place_order method
to MockTimeoutAdapter that awaits a long sleep (same duration as get_portfolio)
and then raises or returns as appropriate to simulate a timeout, ensuring the
test triggers the orchestrator's timeout handling rather than an AttributeError.
Comment on lines
+327
to
+347
| class DelayedAdapter: | ||
| def __init__(self, delay): | ||
| self.delay = delay | ||
|
|
||
| async def get_portfolio(self, session_data): | ||
| await asyncio.sleep(self.delay) | ||
| return Portfolio( | ||
| holdings=[], | ||
| funds=Funds( | ||
| available_balance=0.0, margin_used=0.0, total_balance=0.0 | ||
| ), | ||
| total_pnl=0.0, | ||
| total_day_pnl=0.0, | ||
| total_value=0.0, | ||
| ) | ||
|
|
||
| adapter_registry = { | ||
| "broker1": DelayedAdapter, | ||
| "broker2": DelayedAdapter, | ||
| "broker3": DelayedAdapter, | ||
| } |
Contributor
There was a problem hiding this comment.
DelayedAdapter requires ctor args but orchestrator likely instantiates with no args
init(self, delay) will cause TypeError. Provide a default or accept unused args.
class DelayedAdapter:
- def __init__(self, delay):
- self.delay = delay
+ def __init__(self, delay: float = 0.5, *_, **__):
+ self.delay = delay📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| class DelayedAdapter: | |
| def __init__(self, delay): | |
| self.delay = delay | |
| async def get_portfolio(self, session_data): | |
| await asyncio.sleep(self.delay) | |
| return Portfolio( | |
| holdings=[], | |
| funds=Funds( | |
| available_balance=0.0, margin_used=0.0, total_balance=0.0 | |
| ), | |
| total_pnl=0.0, | |
| total_day_pnl=0.0, | |
| total_value=0.0, | |
| ) | |
| adapter_registry = { | |
| "broker1": DelayedAdapter, | |
| "broker2": DelayedAdapter, | |
| "broker3": DelayedAdapter, | |
| } | |
| class DelayedAdapter: | |
| def __init__(self, delay: float = 0.5, *_, **__): | |
| self.delay = delay | |
| async def get_portfolio(self, session_data): | |
| await asyncio.sleep(self.delay) | |
| return Portfolio( | |
| holdings=[], | |
| funds=Funds( | |
| available_balance=0.0, margin_used=0.0, total_balance=0.0 | |
| ), | |
| total_pnl=0.0, | |
| total_day_pnl=0.0, | |
| total_value=0.0, | |
| ) | |
| adapter_registry = { | |
| "broker1": DelayedAdapter, | |
| "broker2": DelayedAdapter, | |
| "broker3": DelayedAdapter, | |
| } |
🤖 Prompt for AI Agents
In tests/core/test_orchestrator.py around lines 327 to 347, the DelayedAdapter
class defines __init__(self, delay) but the orchestrator likely instantiates
adapters without arguments causing a TypeError; change the constructor to accept
an optional delay (e.g., delay=0) or accept *args, **kwargs so it can be
instantiated with no args, and ensure any test usages that expect a nonzero
delay explicitly pass the value when instantiating the adapter registry or
adjust the adapter_registry entries to be callables/partials that provide the
delay.
Comment on lines
+8
to
+11
| sys.path.insert(0, '/workspace/cmgos18vg0001qyic0qoozodb/Ordo/src') | ||
|
|
||
| from ordo.core.orchestrator import RequestOrchestrator, SessionCheckResult | ||
| from ordo.models.api.unified import BrokerResult, UnifiedResponse |
Contributor
There was a problem hiding this comment.
Remove hardcoded absolute sys.path; use repo-relative import or install package
The absolute path breaks outside your workspace. Use a relative path fallback or require pip install -e ..
-import sys
-import asyncio
-
-sys.path.insert(0, '/workspace/cmgos18vg0001qyic0qoozodb/Ordo/src')
+import sys
+import asyncio
+from pathlib import Path
+# Fallback for local runs without installation: add ./src relative to this file
+src_path = Path(__file__).resolve().parent / "src"
+if src_path.exists():
+ sys.path.insert(0, str(src_path))📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| sys.path.insert(0, '/workspace/cmgos18vg0001qyic0qoozodb/Ordo/src') | |
| from ordo.core.orchestrator import RequestOrchestrator, SessionCheckResult | |
| from ordo.models.api.unified import BrokerResult, UnifiedResponse | |
| import sys | |
| import asyncio | |
| from pathlib import Path | |
| # Fallback for local runs without installation: add ./src relative to this file | |
| src_path = Path(__file__).resolve().parent / "src" | |
| if src_path.exists(): | |
| sys.path.insert(0, str(src_path)) | |
| from ordo.core.orchestrator import RequestOrchestrator, SessionCheckResult | |
| from ordo.models.api.unified import BrokerResult, UnifiedResponse |
🤖 Prompt for AI Agents
In verify_implementation.py around lines 8 to 11, the code inserts a hardcoded
absolute path into sys.path which breaks outside this workspace; remove that
sys.path.insert call and replace it by either making the package importable
(add/install package via pip install -e . in project setup or include package in
PYTHONPATH in CI) or use a repo-relative import fallback (compute repo root with
pathlib.resolve() and insert a relative path) — prefer installing the package in
editable mode and revert to normal imports so tests and external runs don't rely
on absolute workspace paths.
Comment on lines
+10
to
+14
| from ordo.core.orchestrator import RequestOrchestrator, SessionCheckResult | ||
| from ordo.models.api.unified import BrokerResult, UnifiedResponse | ||
| from ordo.models.api.portfolio import Portfolio, Holding, Funds | ||
| from ordo.security.session import SessionManager | ||
|
|
Contributor
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
Unify SessionCheckResult import with source module
Avoid importing SessionCheckResult from orchestrator; import it from ordo.security.session.
-from ordo.core.orchestrator import RequestOrchestrator, SessionCheckResult
+from ordo.core.orchestrator import RequestOrchestrator
+from ordo.security.session import SessionCheckResult
@@
-from ordo.security.session import SessionManager
+# (SessionManager import unused; can be removed)Also applies to: 44-46
🤖 Prompt for AI Agents
In verify_implementation.py around lines 10–14 (and also where
SessionCheckResult is imported again at lines 44–46), the review requests that
SessionCheckResult be imported from its source module ordo.security.session
rather than from ordo.core.orchestrator; update the imports so
SessionCheckResult is removed from the orchestrator import and instead imported
from ordo.security.session (combine or adjust existing SessionManager import as
needed) to ensure the type comes from the correct module.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Base branch:
feat/issue-09-Request-OrchestratorFeature branch:
compyle/cmgos18vg0001qyic0qoozodb-aa8b29aAdded files
src/ordo/core/__init__.pysrc/ordo/core/orchestrator.pysrc/ordo/models/api/unified.pytests/core/test_api_integration.pytests/core/test_orchestrator.pyverify_implementation.pyEdited files
src/ordo/main.pysrc/ordo/security/session.pySummary by CodeRabbit
New Features
Tests