Telemetry signals - #31
Conversation
…docker buildx bake
# Conflicts: # .gitignore
📝 WalkthroughWalkthroughThe change adds Docker Buildx support, runtime and agent telemetry, durable PostgreSQL sessions, structured failure propagation, Bedrock integration, and finance, portfolio, text-to-SQL, and Hello World examples. ChangesBuild and container delivery
Runtime telemetry and sessions
Bedrock and deployment examples
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Suggested labels: Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 2📝 Generate docstrings 💡
⚔️ Resolve merge conflicts 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
|
||
| node_redis.set(f"controller:{host}:{CONTAINER_PORT}:agent_id", agent_id) | ||
| if spec.get("instance_type"): | ||
| node_redis.set(f"agent:{agent_id}:instance_type", spec["instance_type"]) |
There was a problem hiding this comment.
wanted to keep the RedisClient as a variable to pull a value from it immediately afterwards
There was a problem hiding this comment.
Should I .gitignore it? Or rename it to not sound so nefarious?
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Note
Due to the large number of review comments, Critical severity comments were prioritized as inline comments.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
examples/text2sql/workflow/text2sql_workflow.py (1)
47-87: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winSet a finite timeout for future resolution.
Each changed
value()call uses the defaulttimeout=None.Future.value()then polls indefinitely when a result does not arrive. If an EC2 agent fails or its Redis result is lost, the REST request does not return and can exhaust workflow worker capacity. Pass a finite timeout and handleTimeoutErrorat the workflow boundary.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/text2sql/workflow/text2sql_workflow.py` around lines 47 - 87, Update the workflow’s Future.value() calls for schema generation, candidate generation, lint/cost validation, sample execution, selection, and production execution to use a finite timeout, and handle TimeoutError at the workflow boundary by returning the established error response instead of allowing the request to hang. Preserve the existing successful-result processing and validation flow.ventis/controller/global_controller.py (1)
410-420: 🚀 Performance & Scalability | 🟠 Major | ⚡ Quick winRun the runtime telemetry write once per node Redis, not once per replica.
pull_runtime_informationscans everyfuture:*:metricskey on the node Redis andsend_runtime_informationre-UPSERTs every completed future._get_node_redis_for(host)returns the same client for all replicas on a host, so with N replicas on one host this full scan and full UPSERT batch repeat N times per poll interval. The writes are idempotent, so results stay correct, but Redis scan load and database write volume scale with the replica count instead of the node count.Collect the distinct node Redis clients first, then write once for each.
♻️ Proposed restructure of the poll loop
def _poll_controllers(self): + seen_node_redis = set() for instance in self.instance_manager.list_instances(): name = instance["agent_name"] host = instance["host"] port = instance["host_port"] node_redis = self._get_node_redis_for(host) - try: - send_runtime_information( - pull_runtime_information(node_redis), - node_redis, - self.config.get("database", {}).get("url"), - ) - except Exception as e: - logger.warning( - "Failed to write runtime information for instance %s (%s:%s) " - "(non-fatal): %s", - name, - host, - port, - e, - ) + if id(node_redis) not in seen_node_redis: + seen_node_redis.add(id(node_redis)) + try: + send_runtime_information( + pull_runtime_information(node_redis), + node_redis, + self.config.get("database", {}).get("url"), + ) + except Exception as e: + logger.warning( + "Failed to write runtime information for node %s " + "(non-fatal): %s", + host, + e, + )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/controller/global_controller.py` around lines 410 - 420, Update the polling logic around the instance loop to collect distinct node Redis clients, deduplicated by host or client identity, before processing telemetry. Then call pull_runtime_information and send_runtime_information once for each unique node Redis client, while preserving the existing database URL and error-handling behavior.
🟠 Major comments (20)
ventis/cli.py-321-327 (1)
321-327: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winUse a distinct Docker context for each workflow.
At Line 321, builds start only after all workflow contexts are generated. Each workflow currently writes to
docker_container/Workflowat Line 269. If the configuration has multiple workflows, every target builds the last generated workflow and can run the wrong workflow.Create a workflow context path that includes
agent_name. Add a regression test with two workflows and assert that each Bake target has a distinct context.Proposed fix
- docker_context = os.path.join(project_dir, "docker_container", "Workflow") + docker_context = os.path.join( + project_dir, "docker_container", f"workflow-{agent_name}" + )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/cli.py` around lines 321 - 327, Update the workflow context construction in the workflow-generation logic to include the current agent_name, ensuring each workflow writes to and builds from a distinct Docker context instead of sharing docker_container/Workflow. Preserve the existing Bake target wiring in the bake_targets append block, and add a regression test covering two workflows that verifies their Bake targets have different context paths.examples/helloworld/workflow/example_workflow.py-21-24 (1)
21-24: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winSet a finite timeout for the remote agent call.
Future.valuedefaults totimeout=Noneinventis/future.py, Lines 146-171. If the agent or Redis does not respond, this background workflow thread loops forever. The request remainsrunning, and repeated failures can exhaust worker threads. Pass a configured finite timeout and letTimeoutErrorreachventis/deploy.py, which records the request as failed.Proposed fix
- greeting = agent.hello(name=name) - return {"greeting": greeting.value()} + greeting = agent.hello(name=name).value(timeout=30) + return {"greeting": greeting}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/helloworld/workflow/example_workflow.py` around lines 21 - 24, Update the remote agent call in main, specifically the greeting assignment using agent.hello and Future.value, to pass a configured finite timeout instead of relying on the default None. Preserve propagation of TimeoutError so ventis/deploy.py can mark the request as failed.ventis/llm/bedrock.py-30-50 (1)
30-50: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winDo not let telemetry failures change Bedrock call results.
Line 29 returns only after the
finallyblock completes. If_redis.hset_multiple()fails at Lines 42-50, a successful Bedrock response becomes an exception. If_redis.hincrby()fails at Line 37, it can also replace the original Bedrock exception.Handle Redis telemetry failures as best-effort operations. Preserve the Bedrock response or original Bedrock exception.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/llm/bedrock.py` around lines 30 - 50, Update the Redis telemetry operations in the Bedrock call’s exception and finally handling so failures from _redis.hincrby() and _redis.hset_multiple() are swallowed or otherwise isolated as best-effort telemetry errors. Ensure telemetry exceptions never replace the original Bedrock exception or turn a successful response into an exception, while preserving the existing metrics updates when Redis succeeds.examples/text2sql/agents/vllm_agent.py-38-43 (1)
38-43: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winPropagate SQL-generation failures instead of returning prose.
This agent produces SQL candidates.
_fallback_response()returns normal prose, butgenerate()reports it as a successfulstr. A downstream SQL consumer can treat the fallback as model output and then reject or process invalid SQL.Raise the failure after logging it, or return an explicit failure result that the SQL workflow handles. Do not return a synthetic successful response on this path.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/text2sql/agents/vllm_agent.py` around lines 38 - 43, Update the exception path in the agent’s generation method to propagate the Bedrock/SQL-generation failure after logging it, rather than returning _fallback_response(prompt) as successful output. Remove or bypass the synthetic prose fallback so downstream SQL consumers never receive it as a valid SQL candidate; preserve normal successful generation behavior.examples/text2sql/config/global_controller.yaml-20-21 (1)
20-21: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winUse an instance that satisfies the declared resources.
SchemaRetrievalAgentrequests one GPU and 2,048 MB of memory. AWS listst3.microwith no GPU and 1 GiB of memory. This deployment cannot satisfy the agent configuration. Select an accelerator instance with sufficient memory, or remove the GPU requirement before deployment. (docs.aws.amazon.com)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/text2sql/config/global_controller.yaml` around lines 20 - 21, Update the EC2 instance_type in the deployment configuration to an accelerator-capable instance with at least one GPU and 2,048 MB of memory, or remove the GPU requirement from SchemaRetrievalAgent so it matches t3.micro; keep the provider setting unchanged.examples/text2sql/config/global_controller.yaml-107-107 (1)
107-107: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winSensitive Data Exposure (CWE-798): Use of Hard-coded Credentials
Remove and rotate the embedded database credential.
examples/text2sql/config/global_controller.yaml:107stores a complete PostgreSQL URL with username and password. Move the value to deployment secret injection, then rotate the exposed credential.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/text2sql/config/global_controller.yaml` at line 107, Remove the hardcoded PostgreSQL URL from the configuration entry in global_controller.yaml and replace it with the project’s deployment secret-injection reference. Rotate the exposed database credential in the hosting provider and update the injected secret to use the replacement credential.Source: Linters/SAST tools
examples/finance/config/global_controller.ec2_smoke.yaml-25-31 (1)
25-31: 🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy liftReplace the placeholder EC2 resource identifiers.
The AMI, subnet, and security-group IDs are placeholders. This configuration cannot start the smoke deployment in an AWS account. Use valid resources for the target account and region, or publish this as an explicit template with setup instructions.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/finance/config/global_controller.ec2_smoke.yaml` around lines 25 - 31, Replace the placeholder ami_id, subnet_id, and security_group_ids values in the ec2 configuration with valid resources for us-east-1 and the target AWS account, or convert the file into an explicit template and add setup instructions for supplying these identifiers.examples/finance/agents/finance_agent.py-14-19 (1)
14-19: 🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy liftReturn a company name from
get_company_name.Line 17 delegates to
run, which requests an unstructured finance response. The workflow passes this result toget_competitor_listas the company identifier. Use a ticker-to-company lookup or a constrained response format that returns only the company name.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/finance/agents/finance_agent.py` around lines 14 - 19, Update get_company_name so it returns the actual company name for the supplied ticker rather than the unstructured result from run. Use a ticker-to-company lookup or constrain the underlying response to a company-name-only value before returning it, preserving the method’s string contract for get_competitor_list.examples/finance/workflow/example_workflow.py-42-49 (1)
42-49: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winSet a timeout for every Future resolution.
Future.value()polls forever whentimeoutis omitted. If an agent does not return a result, this workflow remains running indefinitely. Repeated REST requests can accumulate blocked background threads. Pass a configured finite timeout and letdeployrecordTimeoutErroras a failed request.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/finance/workflow/example_workflow.py` around lines 42 - 49, Update the workflow’s Future resolutions in the surrounding function, including the competitors print and returned fields, to pass the configured finite timeout to every value() call. Use the same timeout for all Future resolutions and preserve deploy’s existing TimeoutError failure handling.examples/finance/agents/finance_agent.py-27-31 (1)
27-31: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winRemove
.value()from the direct Bedrock response.
VllmAgent.generatereturns astr. Line 29 calls.value()on that string, so everyruncall raisesAttributeError. Assign the generated response directly.Proposed fix
- response = self.vllm.generate(prompt).value() + response = self.vllm.generate(prompt)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/finance/agents/finance_agent.py` around lines 27 - 31, Update the response assignment in the agent run flow to use the string returned directly by VllmAgent.generate, removing the .value() call before returning response.examples/finance/config/policy.yaml-7-10 (1)
7-10: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winAuthorization Bypass (CWE-807)
Reachability: External
Block self-assigned policy context.
ventis/deploy.pystoresrequest._contextinto Redis andventis/controller/local_controller.pyreads that context for_check_policy. A workflow caller can submit{"_context":{"origin":"ceo"}}and match the finance policy rule that grantsaccess: all. Deriveoriginfrom an authenticated server-side identity, or reject caller-supplied authorization context.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/finance/config/policy.yaml` around lines 7 - 10, Update the finance policy flow so callers cannot control the authorization context used by _check_policy: reject request-supplied _context authorization fields, or overwrite origin with the authenticated server-side identity before deploy.py stores request._context in Redis and local_controller.py evaluates it. Preserve legitimate policy evaluation while preventing a caller-provided origin such as ceo from granting access: all.examples/portfolio/agents/intent_agent.py-10-14 (1)
10-14: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winUse a Converse-compatible default model.
IntentAgentdefaults toopenai.gpt-5.4, butcall_bedrock()invokesbedrock-runtime.converse. OpenAI GPT-5.4 on Bedrock is accessed through the Bedrock Mantle Responses API, not the Converse API. Set the default to a model supported byConverse, or add a separate Responses API path and telemetry handling foropenai.gpt-5.4.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/portfolio/agents/intent_agent.py` around lines 10 - 14, Update IntentAgent’s default model configuration used by call_bedrock() to a model supported by the Bedrock Converse API, keeping the existing Converse invocation and telemetry flow unchanged; do not retain openai.gpt-5.4 unless implementing a separate Mantle Responses API path with equivalent telemetry.examples/portfolio/config/global_controller.yaml-89-90 (1)
89-90: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winSecurity Misconfiguration (CWE-319): Cleartext Transmission of Sensitive Information
Require verify-full TLS for the database connection.
_get_engine()createspostgresql+psycopg://...URLs with nosslmode, so libpq falls back topreferand may use unencrypted transport on SSL failure. Configuresslmode=verify-fullwith a trusted CA/root certificate for the RemoteDB host, and pass any SSL options through a trusted runtime path such asVENTIS_DATABASE_URLorconnect_args.Receipt
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/portfolio/config/global_controller.yaml` around lines 89 - 90, Update the database URL configuration used by _get_engine to require TLS with sslmode=verify-full and a trusted CA certificate for the RemoteDB host. Ensure these SSL options reach the psycopg connection through the established trusted runtime path, such as VENTIS_DATABASE_URL or connect_args, rather than relying on the default prefer behavior.ventis/controller/local_controller.py-72-76 (1)
72-76: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winHandle a missing
agent_idkey.
self.redis.get(...)returnsNonewhen the key is absent.RedisClient.getreturnsNonefor a missing key, soself.agent_idbecomesNone.Line 571 then calls
self.redis.hset(f"future:{future_id}:metrics", "agent", self.agent_id). redis-py rejects aNonevalue withDataError. That call sits after thetry/exceptblock in_execute_locally, so the exception propagates out of the pool task and is swallowed by theconcurrent.futuresFuture. Everything after line 571 is skipped, which means:
ventis_context.set_current_future_id(parent or "")never runs, so the worker thread keeps this execution's future id and misattributes the parent of the next future created on that thread.queue_timeis never written.The key is absent whenever the controller starts before
InstanceManagerpublishes it, or when an agent runs outside a managed deployment. Default it to an empty string.🐛 Proposed fix
self.agent_id = self.redis.get( f"controller:{self.agent_host}:{self.public_port}:agent_id" - ) + ) or ""🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/controller/local_controller.py` around lines 72 - 76, Update the agent_id initialization in the controller setup to default a missing Redis value to an empty string, ensuring self.agent_id is never None before _execute_locally writes it with hset. Preserve the existing Redis key lookup and use the empty-string fallback for unmanaged or not-yet-provisioned agents.ventis/deploy.py-184-202 (1)
184-202: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winBound the synchronous session write on the request path.
POST /{fn_name}callsupsert_session()synchronously on the request thread and only catches failures. The Postgres engine is created with plaincreate_engine(url), so an unreachable database can block the request until psycopg’s TCP timeout instead of failing fast. Add a driver connect timeout, for examplecreate_engine(url, connect_args={"connect_timeout": 3}, pool_pre_ping=True), where the engine is created inventis/controller/utils/session_store.py.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/deploy.py` around lines 184 - 202, The synchronous upsert_session request path can block on unreachable Postgres because the session-store engine lacks connection bounds. Update the engine creation in session_store.py to pass a short driver connect timeout (such as 3 seconds) and enable pool_pre_ping, preserving the existing session write behavior.ventis/deploy.py-120-124 (1)
120-124: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winKeep the
resultRedis payload at the original return shape.Line 120 wraps non-dict workflow results in
{"value": result}and the/statusendpoint returns that wrapper asresponse["result"]. A workflow returning a string, list, number, orNonetherefore changes from{"result": "..."}to{"result": {"value": ...}}.Keep
{"value": result}for the sessionoutput_payload, but serialize the bare return value for the Redisresult_key.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/deploy.py` around lines 120 - 124, The Redis result payload currently wraps non-dict workflow returns, changing the original return shape. In the workflow result handling near output_payload, keep {"value": result} for the session output, but serialize the bare result directly for result_key; leave the status_key update unchanged.ventis/controller/local_controller.py-506-510 (1)
506-510: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winReset the thread-local
request_idafter execution.Line 508 sets the thread-local
request_idonly when the inbound request carries one. Nothing clears it. Worker threads are reused fromself._executor, so the value survives into the next execution on that thread.If a later request arrives without a
request_id,Future.__init__reads the stale value throughventis_context.get_request_id()and stamps the previous request's id onto the new future.send_runtime_informationthen attributes that future's telemetry row to the wrongsession_id, andrequest:{old_id}:futuresgains a member that does not belong to it.Line 572 already restores
current_future_idfor exactly this reason. Apply the same treatment torequest_id.🐛 Proposed fix
- if request_id: - self.redis.sadd(f"request:{request_id}:futures", future_id) - ventis_context.set_request_id(request_id) + if request_id: + self.redis.sadd(f"request:{request_id}:futures", future_id) + ventis_context.set_request_id(request_id or "") ventis_context.set_current_future_id(future_id) ventis_context.set_current_metrics_key(self._metrics_key)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/controller/local_controller.py` around lines 506 - 510, Reset the thread-local request_id after each execution, mirroring the existing current_future_id restoration near line 572. Update the execution cleanup around ventis_context.set_request_id and ensure requests without an inbound request_id clear any stale value before the worker thread is reused.ventis/controller/global_controller.py-83-84 (1)
83-84: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winThe
project_idfallback of0cannot be stored.
runtime_information.project_idis declaredUUID NOT NULLinventis/controller/utils/sqlalchemy.pyLine 69. If the config omitsproject_id,assign_project_idstores0, and every runtime UPSERT then fails the UUID cast on PostgreSQL. The poll loop at Line 421-429 logs that failure as a warning, so all runtime telemetry is dropped without a clear signal.Pass the configured value through unchanged and fail fast, or skip telemetry writes when it is absent.
🐛 Proposed fix for the project id fallback
- assign_project_id(self.config.get("project_id",0)) - + project_id = self.config.get("project_id") + if not project_id: + logger.warning( + "No project_id in %s; runtime and agent telemetry writes are disabled.", + config_path, + ) + assign_project_id(project_id) +🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/controller/global_controller.py` around lines 83 - 84, Update the project ID initialization in the global controller to pass the configured project_id through unchanged instead of defaulting to 0. Ensure missing project_id fails fast or prevents runtime telemetry writes, so assign_project_id and subsequent UPSERTs never receive an invalid non-UUID value.ventis/controller/utils/sqlalchemy.py-277-280 (1)
277-280: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winDerive the agent timestamp shift from the agent id, not a fresh random value.
random.randintproduces a new offset on every UPSERT for the sameagent_id.updated_attherefore jumps forward and backward between polls, and agent rows no longer share a timeline with thesessionandruntime_informationrows. Any query that orders agents by freshness, or that correlates agent heartbeats with a session window, reads incorrect times.
ventis/controller/utils/demo_obfuscation.pyLine 9-14 states that every writer must derive its shift the same way from the same id. Useshift_for_session(agent_id)so each agent keeps one stable offset.🐛 Proposed fix for the agent timestamp shift
- "updated_at": to_timestamptz( - raw.get("updated_at") or now, - random.randint(0, RANDOM_SHIFT_MAX_SECONDS), - ), + "updated_at": to_timestamptz( + raw.get("updated_at") or now, + shift_for_session(agent_id), + ),Then drop the now-unused
randomandRANDOM_SHIFT_MAX_SECONDSimports. Note thattests/test_runtime_sqlalchemy.pyLine 425-429 and Line 460-464 assert only that the shifted value falls inside the window, so they still pass; tighten them to the deterministic offset if you want the stronger guarantee.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/controller/utils/sqlalchemy.py` around lines 277 - 280, Replace the random offset in the agent upsert’s updated_at calculation with the deterministic shift_for_session(agent_id) result, preserving the existing fallback timestamp behavior. Remove the now-unused random and RANDOM_SHIFT_MAX_SECONDS imports from this module; update tests only if needed to assert the stable agent-specific offset.ventis/controller/global_controller.py-434-476 (1)
434-476: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winThe counter reset can discard increments recorded during the poll.
Line 434 reads the metrics hash, and Line 473-476 overwrite
full_failures,error_count, andrequests_servedwith 0. The local controller increments those same fields withhincrbywhile this poll runs, so any increment between the read and the reset is lost. The comment at Line 470-472 states the goal of never losing a poll's counts, but overwriting cannot achieve that.Subtract the observed amounts with
hincrbyinstead.hincrbyis atomic, so concurrent increments survive.🐛 Proposed fix for the counter drain
- requests_served = int(float(metrics.get("requests_served") or 0)) + requests_served = int(float(metrics.get("requests_served") or 0)) + full_failures = int(float(metrics.get("full_failures") or 0)) + error_count = int(float(metrics.get("error_count") or 0))- node_redis.hset_multiple( - metrics_key, - {"full_failures": 0, "error_count": 0, "requests_served": 0}, - ) + for field, observed in ( + ("full_failures", full_failures), + ("error_count", error_count), + ("requests_served", requests_served), + ): + if observed: + node_redis.hincrby(metrics_key, field, -observed)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/controller/global_controller.py` around lines 434 - 476, Update the successful persistence branch after send_agent_information in the metrics polling flow to drain counters with atomic hincrby operations using the observed requests_served, full_failures, and error_count values, rather than resetting them with hset_multiple. Preserve concurrent increments recorded after the initial hgetall so they remain available for the next poll.
🟡 Minor comments (7)
examples/portfolio/agents/intent_agent.py-99-118 (1)
99-118: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winValidate
holdingsand finite weight values before normalization.
raw.items()assumesholdingsis a mapping, so a valid model response such as{"holdings":["AAPL"]}raisesAttributeError.float("Infinity")is> 0, so positive infinite weights can normalize tonanand reachRiskAgent. Accept onlydict, reject non-finite weights including infinite values, and ensure the normalized weights are non-empty beforeparse()returns.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/portfolio/agents/intent_agent.py` around lines 99 - 118, Update the holdings parsing and normalization logic in parse() to accept holdings only when it is a dict, skipping non-mapping values without calling raw.items(). Validate each converted weight with a finite-value check before accepting positive values, then normalize only when the resulting holdings mapping is non-empty. Ensure parse() never returns empty or non-finite normalized holdings to RiskAgent.ventis/deploy.py-47-51 (1)
47-51: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winApply the short TTL only when the Postgres fallback exists.
_expire_request_keysruns unconditionally at lines 126 and 152._status_from_sessionreturnsNoneimmediately whendb_urlorproject_idis missing (line 227).For a deployment without a database, a finished request's status and result therefore disappear 300 seconds after completion, and
/statusanswers 404. The 404 is indistinguishable from an unknown request id, so a client that polls on a slower cadence silently loses its result with no diagnostic.Make the TTL configurable, and use a longer retention when no durable fallback is configured.
🛡️ Proposed fix
-COMPLETED_TTL_SECONDS = 300 +COMPLETED_TTL_SECONDS = int(os.environ.get("VENTIS_COMPLETED_TTL_SECONDS", 300)) +# With no session row to fall back on, Redis is the only record of a finished +# request, so it has to outlive a slow polling client. +COMPLETED_TTL_SECONDS_NO_DB = int( + os.environ.get("VENTIS_COMPLETED_TTL_SECONDS_NO_DB", 86400) +)Then select the value inside the helper:
def _expire_request_keys(request_id): + ttl = ( + COMPLETED_TTL_SECONDS + if (db_url and project_id) + else COMPLETED_TTL_SECONDS_NO_DB + ) for suffix in ("status", "result", "error", "context"): redis_client.expire( - f"request:{request_id}:{suffix}", COMPLETED_TTL_SECONDS + f"request:{request_id}:{suffix}", ttl )Also applies to: 86-97
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/deploy.py` around lines 47 - 51, Make COMPLETED_TTL_SECONDS configurable and select the retention value in _expire_request_keys based on whether the deployment has the Postgres fallback configuration (db_url and project_id). Keep the short TTL when _status_from_session can retrieve durable results; use a substantially longer configured retention when no fallback exists, and ensure both callers of _expire_request_keys use the selected value.ventis/controller/cloud_provider_logic/Local/_runtime.py-89-91 (1)
89-91: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winA null
databaseconfig block crashes workflow bootstrap in both provider runtimes. Both sites read the database URL withconfig.get("database", {}).get("url"). YAML maps an empty or commented-outdatabase:key toNone, not to a missing key, sogetreturnsNoneand.get("url")raisesAttributeError. Workflow container bootstrap then fails for that provider. The shared root cause is treating a present-but-null key as a missing key.
ventis/controller/cloud_provider_logic/Local/_runtime.py#L89-L91: changeconfig.get("database", {}).get("url")to(config.get("database") or {}).get("url").ventis/controller/cloud_provider_logic/EC2/_runtime.py#L279-L280: change_controller.config.get("database", {}).get("url")to(_controller.config.get("database") or {}).get("url").🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/controller/cloud_provider_logic/Local/_runtime.py` around lines 89 - 91, Handle null database configuration blocks in both runtime sites: update ventis/controller/cloud_provider_logic/Local/_runtime.py lines 89-91 in the config lookup and ventis/controller/cloud_provider_logic/EC2/_runtime.py lines 279-280 in the _controller.config lookup to fall back to an empty mapping before reading url, preserving existing behavior for configured database values.ventis/controller/utils/sqlalchemy.py-241-242 (1)
241-242: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winA missing
failedfield records a successful future as failed.
raw.get("failed", 1)defaults to 1, so any completed future whose metrics hash omitsfailedis stored withfailed = true.int("")also raisesValueErrorif the field exists but is empty, which aborts the whole batch inside the open transaction. Default to 0 and parse defensively.🐛 Proposed fix for failure defaulting
- "failed": bool(int(raw.get("failed", 1))), + "failed": bool(int(float(raw.get("failed") or 0))),🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/controller/utils/sqlalchemy.py` around lines 241 - 242, Update the metrics parsing around the failed field to default missing values to 0 and handle empty or otherwise invalid values defensively without aborting the batch transaction. Preserve failed = true for valid nonzero values and failed = false for missing, empty, or invalid values.ventis/controller/utils/pricing.py-7-32 (1)
7-32: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winInclude
aws_instance_pricing.dbas installed package data.
pricing.pyreadsventis/controller/utils/aws_instance_pricing.db, and_load_cache()can keep both pricing globals asNoneif the file is absent. Add the database file to theventispackage-data bundle so installed deployments can load the pricing rows.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/controller/utils/pricing.py` around lines 7 - 32, The package configuration must include aws_instance_pricing.db as data for the ventis package so _load_cache() can load pricing rows after installation. Update the existing package-data configuration to bundle the database file located alongside pricing.py, without changing the cache-loading logic.tests/test_demo_obfuscation.py-76-107 (1)
76-107: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winRestore the process state that this fixture replaces.
setUpreplacesVENTIS_DATABASE_URL,session_store._engine,sqlmod._engine, andsqlmod._project_id.tearDownsets generic defaults instead of restoring the previous values. A later test can then run with the wrong database configuration or project state. Save each prior value, restore it during cleanup, and register cleanup before changing state.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_demo_obfuscation.py` around lines 76 - 107, Update the test fixture’s setUp and tearDown to preserve process state: capture the existing VENTIS_DATABASE_URL, session_store._engine, sqlmod._engine, and sqlmod._project_id, register cleanup before modifying them, and restore each original value during teardown rather than assigning generic defaults. Use the fixture’s cleanup mechanism so restoration also occurs when setup or the test fails.tests/test_demo_obfuscation.py-16-18 (1)
16-18: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winPreserve the instant for offset-aware timestamps.
datetime.replace(tzinfo=timezone.utc)relabels the local instant as UTC instead of converting it. For values like2025-01-01T00:00:00-04:00, this records an epoch four hours too early. Apply UTC only when_stored_epoch()receives a naive datetime.Proposed fix
def _stored_epoch(stored): """Unix epoch seconds held by a stored TIMESTAMPTZ.""" - return datetime.fromisoformat(str(stored)).replace(tzinfo=timezone.utc).timestamp() + value = datetime.fromisoformat(str(stored)) + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + return value.timestamp()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_demo_obfuscation.py` around lines 16 - 18, Update _stored_epoch so it converts offset-aware datetimes to UTC without changing their instant, while assigning timezone.utc only to naive datetimes. Preserve the existing ISO parsing and epoch timestamp behavior for both input forms.
🧹 Nitpick comments (8)
ventis/future.py (1)
53-71: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueMark the
parentparameter as ignored.Line 71 now derives the parent from the thread-local context. The
parentparameter is no longer read anywhere in__init__, but the docstring still documents it as "parent future". A reader cannot tell that passing a value has no effect.tests/test_future.pyconfirms the argument is intentionally ignored.Keep the parameter for stub compatibility, but state that it is ignored.
♻️ Proposed docstring update
def __init__(self, parent, service, method, args=None): """ - parent: parent future + parent: ignored. The parent future is taken from the currently + executing future on this thread (ventis_context). + Kept in the signature for generated-stub compatibility. service: service to be called method: method to be called in service args: arguments to be passed to the method """🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/future.py` around lines 53 - 71, Update the __init__ docstring in Future to explicitly state that the retained parent parameter is ignored, while keeping the parameter itself unchanged for stub compatibility.ventis/controller/local_controller_frontend.py (1)
43-44: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the now-unused parse in
Execute.The Redis write that consumed
datais gone, so line 43 only parses and discards. It also changes error behavior for the worse:Executehas no exception handler, so a malformed payload now raises out of the gRPC handler. If the line is removed, the payload reachesrun(), which already catchesjson.JSONDecodeErrorand logs it.Delete the parse and queue the raw payload.
♻️ Proposed fix
logger.info(f"Received request: {request.resonse}") - data = json.loads(request.resonse) self.request_queue.put(request.resonse)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/controller/local_controller_frontend.py` around lines 43 - 44, Remove the unused json.loads parse from Execute and enqueue the raw request.resonse directly with self.request_queue.put. Preserve run() as the point where malformed JSON is handled via its existing JSONDecodeError path.ventis/controller/local_controller.py (1)
493-505: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winDrop the metrics fields that no reader consumes.
This record writes
result,service,method, andargsintofuture:{future_id}:metrics.pull_runtime_informationandsend_runtime_informationread onlyid,request_id,agent,parent,created_at,finished_at,cpu_resource,gpu_resource,queue_time, the token fields,failed, andmodel. The four fields listed above are never read.All four already exist on
future:{future_id}, written byFuture.__init__.argsis the expensive one, becausejson.dumps(args)can hold a full workflow payload and is now stored twice per future.deploy.pyline 50 records that unbounded Redis growth already caused an OOM kill.Write only the fields the telemetry reader uses.
♻️ Proposed fix
self.redis.hset_multiple( f"future:{future_id}:metrics", { "id": future_id, "request_id": request_id or "", - "result": "", "parent": parent or "", - "service": service, - "method": function, - "args": json.dumps(args), "created_at": wall_start, }, )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/controller/local_controller.py` around lines 493 - 505, Update the metrics hash written by the local controller around the future metrics hset_multiple call to remove the unused result, service, method, and args fields, including the json.dumps(args) computation. Retain only fields consumed by pull_runtime_information and send_runtime_information, while leaving the corresponding future record unchanged.tests/test_instance_manager_runtime.py (1)
158-209: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd coverage for the workflow database environment variables.
Local/_runtime.pylines 89-95 now addVENTIS_DATABASE_URLandVENTIS_PROJECT_IDto workflow containers. This test uses a controller config that has nodatabasekey, so it only exercises the false branch of both conditions. No test asserts that the variables appear when the config supplies them.Add one test with
config={"poll_interval": 5, "database": {"url": ...}, "project_id": ...}and assert both variables land in the command.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_instance_manager_runtime.py` around lines 158 - 209, Add a test for the workflow container path in InstanceManager.ensure_instances using controller config with poll_interval, database.url, and project_id values. Assert the generated docker run command includes VENTIS_DATABASE_URL and VENTIS_PROJECT_ID with those configured values, while preserving the existing resource and workflow assertions.tests/test_local_controller_metrics.py (2)
61-66: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winReset the thread-local context between tests.
_execute_locallywrites process-wide thread-local state throughventis_context.set_request_id,set_current_future_id, andset_current_metrics_key. This test class never clears it, so the values survive into later tests in the same process.
tests/test_ventis_context.pyandtests/test_future.pyboth resetventis_context._localinsetUpandtearDown, which is what keeps them passing today. This file depends on that. Add the same reset here so the dependency is not implicit.The
ThreadPoolExecutor(max_workers=1)created at line 64 is also never shut down. Close it intearDown.💚 Proposed fix
class LocalControllerMetricsTests(unittest.TestCase): + def tearDown(self): + ventis_context._local = ventis_context.threading.local() + def test_collect_metrics_returns_expected_keys(self):Add the import at the top of the file:
from ventis import ventis_context🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_local_controller_metrics.py` around lines 61 - 66, Add ventis_context to the test module imports, reset ventis_context._local in setUp and tearDown for LocalControllerMetricsTests, and shut down the ThreadPoolExecutor created by each test in tearDown. Preserve the existing test setup while ensuring thread-local state and executor resources are cleaned up between tests.
31-59: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExtend the fake and cover the
request_idpath.
_FakeRedisimplements nosadd._execute_locallycallsself.redis.sadd(f"request:{request_id}:futures", future_id), but only whenrequest_idis truthy. Both tests pass norequest_id, so that branch never runs and the fake never needs the method. The result is that three new parameters are untested:request_id,submitted_at, andparent.Add
saddto_FakeRedis, then add one test that passes all three and asserts the future registration, thequeue_timefield, and theparentfield on the metrics hash.💚 Proposed fake addition
class _FakeRedis: def __init__(self): self.hashes = {} self.strings = {} + self.sets = {} self.client = _FakeRedisClient() + + def sadd(self, name, *values): + self.sets.setdefault(name, set()).update(values)Also applies to: 123-141
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_local_controller_metrics.py` around lines 31 - 59, Extend _FakeRedis with a sadd method that records set membership, then add a test exercising _execute_locally with request_id, submitted_at, and parent. Assert the request futures set contains the future ID and the resulting metrics hash includes the expected queue_time and parent fields.tests/test_deploy.py (1)
48-53: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a workflow that returns a non-dict value.
Both fixtures return a dict.
_execute_workflowwraps a non-dict result as{"value": result}and stores that wrapper in the Redisresultkey, which changes the/statusresponse shape for those workflows. No test covers it, so the change is invisible to this suite.Add a third fixture and assert the
/statuspayload.💚 Proposed test addition
def _failing_workflow(x=1): raise RuntimeError("workflow blew up") + + +def _scalar_workflow(x=1): + return f"answer-{x}"Then add the test to
DeployHandleWorkflowTests:def test_status_shape_for_a_non_dict_result(self): with _deployed_app(workflow_fn=_scalar_workflow) as app: client = app.test_client() request_id = client.post( "/_scalar_workflow", json={"x": 2} ).get_json()["request_id"] resp = client.get(f"/status/{request_id}") self.assertEqual(resp.status_code, 200) self.assertEqual(resp.get_json()["result"], "answer-2")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_deploy.py` around lines 48 - 53, Add a _scalar_workflow fixture alongside _noop_workflow and _failing_workflow that returns a non-dict value based on x. In DeployHandleWorkflowTests, add a test using _deployed_app with _scalar_workflow, submit the workflow, fetch its status, and assert a 200 response whose result field contains the original scalar value rather than a wrapper object.ventis/controller/instance_manager.py (1)
137-147: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsolidate the duplicated
controller:{endpoint}:agent_idwrite.Line 140 writes
controller:{endpoint}:agent_id. Both provider runtimes already write the same key during bootstrap:
ventis/controller/cloud_provider_logic/Local/_runtime.pyline 64 writes it to_require_controller().redis.ventis/controller/cloud_provider_logic/EC2/_runtime.pyline 226 writes it to the node Redis.The three writes target different Redis clients, so the key currently lands in the right place only because the writes overlap.
LocalController.__init__reads this key from its own node Redis. Keep one owner of this write._write_instanceis the better owner, because it already resolvesnode_redisby host and runs for every provider.Remove the runtime-level writes and let
_write_instancepublish the key.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/controller/instance_manager.py` around lines 137 - 147, Remove the duplicated controller:{endpoint}:agent_id writes from the Local and EC2 runtime bootstrap paths, specifically in their runtime initialization methods. Keep the existing write in InstanceManager._write_instance, which resolves node_redis by host and publishes the key for every provider; leave the separate agent_id-to-instance_type write unchanged.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 67236a14-8685-4c85-ae9c-b757ebfabc09
⛔ Files ignored due to path filters (3)
examples/portfolio.zipis excluded by!**/*.zipuv.lockis excluded by!**/*.lockventis/controller/utils/aws_instance_pricing.dbis excluded by!**/*.db
📒 Files selected for processing (63)
.dockerignore.gitignoreREADME.mdexamples/finance/agents/finance_agent.pyexamples/finance/agents/finance_agent.yamlexamples/finance/agents/market_agent.pyexamples/finance/agents/market_agent.yamlexamples/finance/agents/vllm_agent.pyexamples/finance/agents/vllm_agent.yamlexamples/finance/config/global_controller.ec2_smoke.yamlexamples/finance/config/global_controller.yamlexamples/finance/config/policy.yamlexamples/finance/workflow/example_workflow.pyexamples/helloworld/README.mdexamples/helloworld/agents/example_agent.pyexamples/helloworld/agents/example_agent.yamlexamples/helloworld/agents/vllm_agent.pyexamples/helloworld/agents/vllm_agent.yamlexamples/helloworld/config/global_controller.yamlexamples/helloworld/config/policy.yamlexamples/helloworld/workflow/example_workflow.pyexamples/portfolio/agents/advisor_agent.pyexamples/portfolio/agents/intent_agent.pyexamples/portfolio/agents/intent_agent.yamlexamples/portfolio/agents/metrics_agent.pyexamples/portfolio/config/global_controller.yamlexamples/portfolio/config/policy.yamlexamples/portfolio/workflow/portfolio_workflow.pyexamples/text2sql/agents/vllm_agent.pyexamples/text2sql/config/global_controller.yamlexamples/text2sql/workflow/text2sql_workflow.pypyproject.tomlrequirements.txttests/test_cli.pytests/test_demo_obfuscation.pytests/test_deploy.pytests/test_future.pytests/test_gpu_metrics.pytests/test_instance_manager_runtime.pytests/test_local_controller_metrics.pytests/test_runtime_ec2.pytests/test_runtime_sqlalchemy.pytests/test_session_store.pytests/test_ventis_context.pyventis/cli.pyventis/controller/cloud_provider_logic/EC2/_runtime.pyventis/controller/cloud_provider_logic/Local/_runtime.pyventis/controller/global_controller.pyventis/controller/instance_manager.pyventis/controller/local_controller.pyventis/controller/local_controller_frontend.pyventis/controller/utils/demo_obfuscation.pyventis/controller/utils/gpu_metrics.pyventis/controller/utils/pricing.pyventis/controller/utils/session_store.pyventis/controller/utils/sqlalchemy.pyventis/deploy.pyventis/future.pyventis/llm/__init__.pyventis/llm/bedrock.pyventis/stub_generator.pyventis/utils/redis_client.pyventis/ventis_context.py
| # Demo-only multipliers for scaling displayed costs; not real recorded costs. | ||
| token_cost_multiplier = 10000 | ||
| server_cost_multiplier = 100000 | ||
|
|
||
| with _get_engine(database_url).begin() as conn: | ||
| for raw in rows: | ||
| agent = raw.get("agent") | ||
| res = resources_by_agent.get(agent, {}) | ||
| agent_id = raw.get("agent") | ||
| fid = raw.get("future_id") | ||
| if not fid: | ||
| continue | ||
| session_id = raw.get("request_id") | ||
| if not session_id: | ||
| continue | ||
| workflow = ( | ||
| redis_client.get(f"request:{session_id}:workflow") | ||
| if redis_client is not None | ||
| else None | ||
| ) | ||
| # A future without finished_at is still executing. Now that metrics live | ||
| # entirely on the executing node (future:{future_id}:metrics is only ever | ||
| # written by the one process that runs it), "incomplete" genuinely means | ||
| # "still running" -- skip it and let a later poll, once it has actually | ||
| # finished, write the real measurements instead. | ||
| if not raw.get("finished_at"): | ||
| continue | ||
| start = float(raw.get("created_at") or 0) | ||
| end = float(raw.get("finished_at") or time.time()) | ||
| cpu_resource = float(raw.get("cpu_resource") or res.get("cpu", 0)) | ||
| gpu_resource = float(res.get("gpu", 0)) | ||
| end = float(raw.get("finished_at")) | ||
| shift = shift_for_session(session_id) | ||
| input_token_count = int(float(raw.get("input_token_count") or 0)) | ||
| output_token_count = int(float(raw.get("output_token_count") or 0)) | ||
| token_count = int(float(raw.get("token_count") or 0)) | ||
| cached_tokens = int(float(raw.get("input_cache_tokens") or 0)) | ||
| model = raw.get("model") | ||
| token_cost = pricing.compute_token_cost( | ||
| model, input_token_count, output_token_count | ||
| ) | ||
| server_cost = pricing.compute_server_cost( | ||
| redis_client.get(f"agent:{agent_id}:instance_type") | ||
| if redis_client is not None and agent_id | ||
| else None, | ||
| end - start, | ||
| ) | ||
|
|
||
| server_cost *= server_cost_multiplier | ||
| token_cost *= token_cost_multiplier |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
The hardcoded cost multipliers contradict the tests in this PR.
token_cost_multiplier and server_cost_multiplier are fixed at 10000 and 100000, and no logger.warning call exists in this module. The tests added in this PR expect environment-driven multipliers that default to 1 and a warning per active multiplier:
tests/test_runtime_sqlalchemy.pyLine 286-287 expectstotal_cost == 6.0andtoken_cost == 6.0with no multiplier configured. The current code stores60000.tests/test_runtime_sqlalchemy.pyLine 335-336 expectsserver_cost == 0.096. The current code stores9600.tests/test_runtime_sqlalchemy.pyLine 358-383 setsVENTIS_DEMO_TOKEN_COST_MULTIPLIERandVENTIS_DEMO_SERVER_COST_MULTIPLIER, asserts a WARNING record fromventis.controller.utils.sqlalchemy, and asserts the scaled values.
Read both multipliers from the environment, default them to 1, and log a warning when either is not 1. Recorded costs then stay real unless a demo explicitly opts in.
🐛 Proposed fix for the multiplier contract
- # Demo-only multipliers for scaling displayed costs; not real recorded costs.
- token_cost_multiplier = 10000
- server_cost_multiplier = 100000
+ # Demo-only multipliers for scaling displayed costs; not real recorded costs.
+ token_cost_multiplier = float(
+ os.environ.get("VENTIS_DEMO_TOKEN_COST_MULTIPLIER", 1)
+ )
+ server_cost_multiplier = float(
+ os.environ.get("VENTIS_DEMO_SERVER_COST_MULTIPLIER", 1)
+ )
+ if token_cost_multiplier != 1:
+ logger.warning(
+ "VENTIS_DEMO_TOKEN_COST_MULTIPLIER=%s is scaling recorded token costs; "
+ "stored values are not real costs.",
+ token_cost_multiplier,
+ )
+ if server_cost_multiplier != 1:
+ logger.warning(
+ "VENTIS_DEMO_SERVER_COST_MULTIPLIER=%s is scaling recorded server costs; "
+ "stored values are not real costs.",
+ server_cost_multiplier,
+ )📝 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.
| # Demo-only multipliers for scaling displayed costs; not real recorded costs. | |
| token_cost_multiplier = 10000 | |
| server_cost_multiplier = 100000 | |
| with _get_engine(database_url).begin() as conn: | |
| for raw in rows: | |
| agent = raw.get("agent") | |
| res = resources_by_agent.get(agent, {}) | |
| agent_id = raw.get("agent") | |
| fid = raw.get("future_id") | |
| if not fid: | |
| continue | |
| session_id = raw.get("request_id") | |
| if not session_id: | |
| continue | |
| workflow = ( | |
| redis_client.get(f"request:{session_id}:workflow") | |
| if redis_client is not None | |
| else None | |
| ) | |
| # A future without finished_at is still executing. Now that metrics live | |
| # entirely on the executing node (future:{future_id}:metrics is only ever | |
| # written by the one process that runs it), "incomplete" genuinely means | |
| # "still running" -- skip it and let a later poll, once it has actually | |
| # finished, write the real measurements instead. | |
| if not raw.get("finished_at"): | |
| continue | |
| start = float(raw.get("created_at") or 0) | |
| end = float(raw.get("finished_at") or time.time()) | |
| cpu_resource = float(raw.get("cpu_resource") or res.get("cpu", 0)) | |
| gpu_resource = float(res.get("gpu", 0)) | |
| end = float(raw.get("finished_at")) | |
| shift = shift_for_session(session_id) | |
| input_token_count = int(float(raw.get("input_token_count") or 0)) | |
| output_token_count = int(float(raw.get("output_token_count") or 0)) | |
| token_count = int(float(raw.get("token_count") or 0)) | |
| cached_tokens = int(float(raw.get("input_cache_tokens") or 0)) | |
| model = raw.get("model") | |
| token_cost = pricing.compute_token_cost( | |
| model, input_token_count, output_token_count | |
| ) | |
| server_cost = pricing.compute_server_cost( | |
| redis_client.get(f"agent:{agent_id}:instance_type") | |
| if redis_client is not None and agent_id | |
| else None, | |
| end - start, | |
| ) | |
| server_cost *= server_cost_multiplier | |
| token_cost *= token_cost_multiplier | |
| # Demo-only multipliers for scaling displayed costs; not real recorded costs. | |
| token_cost_multiplier = float( | |
| os.environ.get("VENTIS_DEMO_TOKEN_COST_MULTIPLIER", 1) | |
| ) | |
| server_cost_multiplier = float( | |
| os.environ.get("VENTIS_DEMO_SERVER_COST_MULTIPLIER", 1) | |
| ) | |
| if token_cost_multiplier != 1: | |
| logger.warning( | |
| "VENTIS_DEMO_TOKEN_COST_MULTIPLIER=%s is scaling recorded token costs; " | |
| "stored values are not real costs.", | |
| token_cost_multiplier, | |
| ) | |
| if server_cost_multiplier != 1: | |
| logger.warning( | |
| "VENTIS_DEMO_SERVER_COST_MULTIPLIER=%s is scaling recorded server costs; " | |
| "stored values are not real costs.", | |
| server_cost_multiplier, | |
| ) | |
| with _get_engine(database_url).begin() as conn: | |
| for raw in rows: | |
| agent_id = raw.get("agent") | |
| fid = raw.get("future_id") | |
| if not fid: | |
| continue | |
| session_id = raw.get("request_id") | |
| if not session_id: | |
| continue | |
| # A future without finished_at is still executing. Now that metrics live | |
| # entirely on the executing node (future:{future_id}:metrics is only ever | |
| # written by the one process that runs it), "incomplete" genuinely means | |
| # "still running" -- skip it and let a later poll, once it has actually | |
| # finished, write the real measurements instead. | |
| if not raw.get("finished_at"): | |
| continue | |
| start = float(raw.get("created_at") or 0) | |
| end = float(raw.get("finished_at")) | |
| shift = shift_for_session(session_id) | |
| input_token_count = int(float(raw.get("input_token_count") or 0)) | |
| output_token_count = int(float(raw.get("output_token_count") or 0)) | |
| token_count = int(float(raw.get("token_count") or 0)) | |
| cached_tokens = int(float(raw.get("input_cache_tokens") or 0)) | |
| model = raw.get("model") | |
| token_cost = pricing.compute_token_cost( | |
| model, input_token_count, output_token_count | |
| ) | |
| server_cost = pricing.compute_server_cost( | |
| redis_client.get(f"agent:{agent_id}:instance_type") | |
| if redis_client is not None and agent_id | |
| else None, | |
| end - start, | |
| ) | |
| server_cost *= server_cost_multiplier | |
| token_cost *= token_cost_multiplier |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/controller/utils/sqlalchemy.py` around lines 182 - 221, Update the
cost multiplier initialization in the metrics persistence flow to read
VENTIS_DEMO_TOKEN_COST_MULTIPLIER and VENTIS_DEMO_SERVER_COST_MULTIPLIER from
the environment, defaulting each to 1. Use the resulting values when scaling
token_cost and server_cost, and add logger.warning calls in the sqlalchemy
module whenever either configured multiplier differs from 1 so unscaled recorded
costs remain the default.
* added a failure field to the future that fails if it happens * fixed some issues * simplified logic and sent it to origin host * fixes
There was a problem hiding this comment.
Actionable comments posted: 9
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (4)
ventis/controller/utils/session_store.py (1)
65-80: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winReject stale session upserts before updating the durable status.
_SESSION_UPSERTapplies everyon conflict updatewithout checkingupdated_at, so the existing test for reverse timestamps storescompletedwithupdated_at = 999.0after a laterrunningwrite. Make the conflict clause write only whenexcluded.updated_at >= session.updated_at, and define precedence for equal timestamps so an out-of-orderfailedwrite cannot replacecompleted.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ventis/controller/utils/session_store.py` around lines 65 - 80, The _SESSION_UPSERT conflict clause currently applies stale writes unconditionally. Update its conditional update logic so writes occur only when excluded.updated_at is greater than session.updated_at, or equal with an explicit status precedence that prevents failed from replacing completed; preserve newer updates and define the equal-timestamp ordering consistently.examples/portfolio/agents/intent_agent.py (3)
101-111: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winReject non-finite weights before normalization.
float(weight)acceptsinf, andw > 0allows it intoholdings. If the total becomesinf, normalization producesnanweights that can corrupt portfolio risk calculations. Filter withmath.isfinite(w)and reject non-finite totals.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/portfolio/agents/intent_agent.py` around lines 101 - 111, Update the holdings parsing and normalization flow to accept weights only when w is finite and positive, using math.isfinite after float conversion. Also require the computed total to be finite and positive before normalizing, preventing inf or nan values from entering holdings.
85-95: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winDecode JSON instead of matching
{.*}greedily.
r"\{.*\}"can consume the first valid JSON object plus following brace-delimited text, sojson.loads()raisesJSONDecodeErrorfor valid model output. Try candidates from the first{forward and return the first decoded dict viajson.JSONDecoder.raw_decode().Suggested parser change
- match = re.search(r"\{.*\}", text, re.DOTALL) - if not match: - return None - try: - return json.loads(match.group(0)) - except (ValueError, TypeError): - return None + decoder = json.JSONDecoder() + for candidate in re.finditer(r"\{", text): + try: + value, _ = decoder.raw_decode(text[candidate.start():]) + except json.JSONDecodeError: + continue + if isinstance(value, dict): + return value + return None🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/portfolio/agents/intent_agent.py` around lines 85 - 95, The _extract_json method should stop using the greedy `{.*}` regex, which can include trailing brace-delimited text. Use json.JSONDecoder.raw_decode() to try JSON candidates beginning at each `{` position, returning the first successfully decoded dictionary and None when no valid object is found.
113-118: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winEnforce a maximum lookback period.
_sanitize()only caps nonpositive values, so a valid LLM output likelookback_days=999999999is passed toPriceAgent. The synthetic fallback iteratesrange(lookback_days), which can allocate unbounded CPU and memory. Define and enforce a documented upper bound before returning the lookup result.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/portfolio/agents/intent_agent.py` around lines 113 - 118, Update _sanitize() to define a documented maximum lookback constant and clamp valid lookback_days values above that limit before returning the result. Preserve DEFAULT_LOOKBACK_DAYS for invalid or nonpositive inputs, and ensure PriceAgent never receives an unbounded lookback period.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@examples/portfolio/agents/price_agent.py`:
- Around line 30-35: Update the history-building logic in PriceAgent.get_history
so invalid interior closes preserve their date positions instead of compacting
valid prices together. Carry validity or timestamp information into
MetricsAgent.compute, and ensure returns spanning missing closes are excluded
while risk_agent NaN closes remain excluded from covariance alignment.
- Around line 21-35: Update the close-value filtering in the price history
processing to convert each raw close once, retain only values passing
math.isfinite, and append the accepted value to closes while appending its
corresponding date to dates. Replace the separate list comprehensions over
raw_closes and hist.index, preserving date/value alignment and excluding NaN,
positive infinity, and negative infinity.
In `@tests/test_error_propagation.py`:
- Around line 69-74: Update the `future` mock in the test fixture to define an
`id` attribute with the expected future identifier before `Future.value()` is
exercised, preserving the existing `_key` and `_poll_redis` behavior so the test
reaches and asserts the intended `RuntimeError`.
In `@ventis/controller/local_controller.py`:
- Around line 699-701: Update the WriteResult failure handling around
_mark_future_failed so the originating controller can observe terminal callback
state: persist the failure in an idempotent callback outbox with retry delivery,
or write the terminal failure to Redis reachable by origin rather than only the
executing controller’s Redis. Ensure Future.value() receives either the result
or failure without waiting indefinitely.
- Around line 556-569: The early failure checks for a missing agent or method in
the execution flow must run inside the existing try/finally so every terminal
execution is finalized. Move the `self.agent` and `getattr(self.agent, function,
None)` validation into the try path, preserving their logging and
`_mark_future_failed` behavior, and ensure `requests_served` is incremented
consistently before the failure returns while letting finally record all
execution metrics.
- Around line 289-299: Update the request handling around _process_request to
reject non-object JSON payloads without terminating the request loop, and guard
future_id/origin extraction in the generic exception path with isinstance(data,
dict) before calling .get(). Preserve existing error logging and
_mark_future_failed behavior for valid object payloads.
- Around line 682-687: Update _send_result_callback and the destination-channel
setup in _get_remote_stub so remote failure callbacks are encrypted and
authenticated with grpc.secure_channel and mutual TLS credentials. Ensure the
callback payload containing error_message is sent only over that secured
channel; if mutual TLS cannot be configured, replace the raw failure text with a
safe error code instead.
In `@ventis/future.py`:
- Around line 156-162: Update Future.value and _notify_consumers so failed
futures propagate failed=1 and the stored error_message to every registered
consumer via WriteResult before raising. Make consumer notification idempotent
so the failure is sent at most once, and add an integration test covering a
registered consumer receiving a failed future instead of timing out.
- Around line 121-125: Make FutureExecute state transitions idempotent by adding
a per-future request status/ticket check in run or _process_request before
submitting work, and reject duplicate processing once the future has a terminal
result or failed state. Ensure late retries cannot overwrite the terminal state
recorded by Future._submit_request, including the failed/error metrics updated
in the shown path.
---
Outside diff comments:
In `@examples/portfolio/agents/intent_agent.py`:
- Around line 101-111: Update the holdings parsing and normalization flow to
accept weights only when w is finite and positive, using math.isfinite after
float conversion. Also require the computed total to be finite and positive
before normalizing, preventing inf or nan values from entering holdings.
- Around line 85-95: The _extract_json method should stop using the greedy
`{.*}` regex, which can include trailing brace-delimited text. Use
json.JSONDecoder.raw_decode() to try JSON candidates beginning at each `{`
position, returning the first successfully decoded dictionary and None when no
valid object is found.
- Around line 113-118: Update _sanitize() to define a documented maximum
lookback constant and clamp valid lookback_days values above that limit before
returning the result. Preserve DEFAULT_LOOKBACK_DAYS for invalid or nonpositive
inputs, and ensure PriceAgent never receives an unbounded lookback period.
In `@ventis/controller/utils/session_store.py`:
- Around line 65-80: The _SESSION_UPSERT conflict clause currently applies stale
writes unconditionally. Update its conditional update logic so writes occur only
when excluded.updated_at is greater than session.updated_at, or equal with an
explicit status precedence that prevents failed from replacing completed;
preserve newer updates and define the equal-timestamp ordering consistently.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 9f45907c-1b5e-432f-a262-77161005903c
📒 Files selected for processing (16)
examples/portfolio/agents/intent_agent.pyexamples/portfolio/agents/price_agent.pyexamples/portfolio/config/global_controller.yamlexamples/text2sql/config/global_controller.yamltests/test_error_propagation.pytests/test_future.pytests/test_local_controller_metrics.pytests/test_runtime_sqlalchemy.pytests/test_session_store.pyventis/controller/local_controller.pyventis/controller/local_controller_frontend.pyventis/controller/utils/session_store.pyventis/controller/utils/sqlalchemy.pyventis/future.pyventis/llm/bedrock.pyventis/stub_generator.py
💤 Files with no reviewable changes (2)
- examples/portfolio/config/global_controller.yaml
- ventis/stub_generator.py
🚧 Files skipped from review as they are similar to previous changes (6)
- tests/test_future.py
- ventis/llm/bedrock.py
- examples/text2sql/config/global_controller.yaml
- ventis/controller/utils/sqlalchemy.py
- tests/test_session_store.py
- tests/test_runtime_sqlalchemy.py
| future = SimpleNamespace( | ||
| redis=redis, | ||
| _key=lambda: "future:future-1", | ||
| _poll_redis=lambda: Future._poll_redis(future), | ||
| result=None, | ||
| ) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Set future.id in this mock.
Future.value() reads self.id before it calls _poll_redis(). This fixture has no id, so the test raises AttributeError instead of the expected RuntimeError.
Proposed fix
future = SimpleNamespace(
redis=redis,
_key=lambda: "future:future-1",
_poll_redis=lambda: Future._poll_redis(future),
+ id="future-1",
result=None,
)📝 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.
| future = SimpleNamespace( | |
| redis=redis, | |
| _key=lambda: "future:future-1", | |
| _poll_redis=lambda: Future._poll_redis(future), | |
| result=None, | |
| ) | |
| future = SimpleNamespace( | |
| redis=redis, | |
| _key=lambda: "future:future-1", | |
| _poll_redis=lambda: Future._poll_redis(future), | |
| id="future-1", | |
| result=None, | |
| ) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/test_error_propagation.py` around lines 69 - 74, Update the `future`
mock in the test fixture to define an `id` attribute with the expected future
identifier before `Future.value()` is exercised, preserving the existing `_key`
and `_poll_redis` behavior so the test reaches and asserts the intended
`RuntimeError`.
| data = None | ||
| try: | ||
| data = json.loads(raw) | ||
| self._process_request(data) | ||
| except json.JSONDecodeError: | ||
| logger.error("Invalid JSON in request: %s", raw) | ||
| except Exception as e: | ||
| logger.error("Error processing request: %s", e) | ||
| self._mark_future_failed( | ||
| data.get("future_id"), e, data.get("origin") | ||
| ) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Keep the request loop alive for non-object JSON.
Execute queues any valid JSON value. For null, a list, or a string, _process_request raises on .get(). Lines 297-299 then call .get() again, so run() exits and stops all request processing.
Validate that the payload is a JSON object. Guard failure-field extraction with isinstance(data, dict).
Proposed fix
data = json.loads(raw)
+if not isinstance(data, dict):
+ raise ValueError("Request payload must be a JSON object")
self._process_request(data)
...
except Exception as e:
logger.error("Error processing request: %s", e)
+ future_id = data.get("future_id") if isinstance(data, dict) else None
+ origin = data.get("origin") if isinstance(data, dict) else None
self._mark_future_failed(
- data.get("future_id"), e, data.get("origin")
+ future_id, e, origin
)📝 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.
| data = None | |
| try: | |
| data = json.loads(raw) | |
| self._process_request(data) | |
| except json.JSONDecodeError: | |
| logger.error("Invalid JSON in request: %s", raw) | |
| except Exception as e: | |
| logger.error("Error processing request: %s", e) | |
| self._mark_future_failed( | |
| data.get("future_id"), e, data.get("origin") | |
| ) | |
| data = None | |
| try: | |
| data = json.loads(raw) | |
| if not isinstance(data, dict): | |
| raise ValueError("Request payload must be a JSON object") | |
| self._process_request(data) | |
| except json.JSONDecodeError: | |
| logger.error("Invalid JSON in request: %s", raw) | |
| except Exception as e: | |
| logger.error("Error processing request: %s", e) | |
| future_id = data.get("future_id") if isinstance(data, dict) else None | |
| origin = data.get("origin") if isinstance(data, dict) else None | |
| self._mark_future_failed( | |
| future_id, e, origin | |
| ) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/controller/local_controller.py` around lines 289 - 299, Update the
request handling around _process_request to reject non-object JSON payloads
without terminating the request loop, and guard future_id/origin extraction in
the generic exception path with isinstance(data, dict) before calling .get().
Preserve existing error logging and _mark_future_failed behavior for valid
object payloads.
| if self.agent is None: | ||
| logger.error("No agent loaded, cannot execute %s.%s", service, function) | ||
| self._mark_future_failed(future_id, "No agent loaded", origin) | ||
| return | ||
|
|
||
| method = getattr(self.agent, function, None) | ||
| if method is None: | ||
| logger.error("Agent %s has no method '%s'", self.agent_name, function) | ||
| self._mark_future_failed( | ||
| future_id, f"Agent {self.agent_name} has no method '{function}'", origin | ||
| ) | ||
| return | ||
|
|
||
| self.redis.hincrby(self._metrics_key, "requests_served", 1) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Finalize metrics for early execution failures.
The missing-agent and missing-method returns occur before the try/finally. These terminal failures omit finished_at, resource usage, agent ID, queue time, and full_failures. They also do not increment requests_served, unlike agent-method failures.
Move these checks into the try path and let finally finalize every execution record.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/controller/local_controller.py` around lines 556 - 569, The early
failure checks for a missing agent or method in the execution flow must run
inside the existing try/finally so every terminal execution is finalized. Move
the `self.agent` and `getattr(self.agent, function, None)` validation into the
try path, preserving their logging and `_mark_future_failed` behavior, and
ensure `requests_served` is incremented consistently before the failure returns
while letting finally record all execution metrics.
| payload = json.dumps({ | ||
| "future_id": future_id, | ||
| "result": result, | ||
| "failed": int(bool(failed)), | ||
| "error_message": str(error_message or ""), | ||
| }) |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
git ls-files | rg '(^|/)local_controller\.py$|(^|/)local_controller_frontend\.py$|proto|grpc|requirements|pyproject|setup\.cfg|setup\.py' || true
echo
echo "== local_controller relevant symbols =="
wc -l ventis/controller/local_controller.py
ast-grep outline ventis/controller/local_controller.py --match LocalController --view expanded || true
rg -n "_get_remote_stub|_send_result_callback|grpc\.insecure_channel|secure_channel|result_callback|WriteResult|future_id|error_message|mark_future_failed" ventis/controller/local_controller.py ventis/controller ventis -S || true
echo
echo "== lines 630-710 =="
sed -n '630,710p' ventis/controller/local_controller.py
echo
echo "== lines 1-120 =="
sed -n '1,120p' ventis/controller/local_controller.pyRepository: CanyonCodeCoreAI/canyoncodecore
Length of output: 34605
Sensitive Data Exposure (CWE-319): Cleartext Transmission of Sensitive Information
Reachability: Internal
Reachability path
● Entry
ventis/future.py:99
_submit_request: Send the gRPC request to the local controller.
│
▼
● Hop
ventis/controller/local_controller_frontend.py:47
WriteResult: Accept a result or error from a remote controller and write it to local Redis.
│
▼
● Hop
tests/test_error_propagation.py
│
▼
● Sink
ventis/controller/local_controller.py
Encrypt and authenticate remote failure callbacks.
_send_result_callback sends error_message from _mark_future_failed, and _get_remote_stub uses grpc.insecure_channel for destination controller calls. A network observer on the controller-to-controller path can read raw failure text. Use grpc.secure_channel with mutual TLS for these callbacks, or limit the callback payload to safe error codes when encryption is not guaranteed.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/controller/local_controller.py` around lines 682 - 687, Update
_send_result_callback and the destination-channel setup in _get_remote_stub so
remote failure callbacks are encrypted and authenticated with
grpc.secure_channel and mutual TLS credentials. Ensure the callback payload
containing error_message is sent only over that secured channel; if mutual TLS
cannot be configured, replace the raw failure text with a safe error code
instead.
| except Exception as e: | ||
| logger.error("Failed to send result callback to %s: %s", origin, e) | ||
| self._mark_future_failed(future_id, f"Result callback failed: {e}") |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Persist callback delivery for the origin controller.
When WriteResult fails, this code records failure only in the executing controller’s Redis. The originating Future.value() polls its own Redis, so it receives neither the result nor this failure and can wait indefinitely.
Persist an idempotent callback outbox and retry delivery, or store terminal state in a Redis location reachable by the origin controller.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/controller/local_controller.py` around lines 699 - 701, Update the
WriteResult failure handling around _mark_future_failed so the originating
controller can observe terminal callback state: persist the failure in an
idempotent callback outbox with retry delivery, or write the terminal failure to
Redis reachable by origin rather than only the executing controller’s Redis.
Ensure Future.value() receives either the result or failure without waiting
indefinitely.
| self.redis.hset(f"future:{self.id}", "error", str(e)) | ||
| self.redis.hset_multiple(f"future:{self.id}:metrics", { | ||
| "failed": 1, | ||
| "error_message": str(e), | ||
| }) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C 12 -P '\bdef\s+Execute\s*\(|\bstub\.Execute\s*\(' ventis tests --glob '*.py' || true
rg -n -C 12 'future_id|request_queue|_process_request|dedup|idempot|retry' ventis tests --glob '*.py' || trueRepository: CanyonCodeCoreAI/canyoncodecore
Length of output: 50388
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,180p' ventis/future.py
printf '\n--- LocalControllerServicer Execute ---\n'
rg -n -C 35 'Execute|WriteResult|_process_request|future_id' ventis/controller/local_controller.py ventis/controller/local_controller_frontend.py ventis/controller/local_controller_servicer.py ventis/future.py ventis/controller/local_controller_frontend.py
printf '\n--- Request processing loop / future status checks ---\n'
rg -n -C 20 'request_queue|get\(|hexists|error|failed|WriteResult|_process_request' ventis/controller --glob '*.py'Repository: CanyonCodeCoreAI/canyoncodecore
Length of output: 50387
Make FutureExecute state transitions idempotent.
Execute only enqueues the request, and run() calls _process_request(data) for every dequeue without checking a terminal Redis state. A retry or duplicate request can still submit work for the same future_id, which can then write a result or retry error after Future._submit_request recorded failed=1. Use a per-future_id request ticket/status, or make WriteResult/_process_request reject transitions from terminal failure/result states.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/future.py` around lines 121 - 125, Make FutureExecute state
transitions idempotent by adding a per-future request status/ticket check in run
or _process_request before submitting work, and reject duplicate processing once
the future has a terminal result or failed state. Ensure late retries cannot
overwrite the terminal state recorded by Future._submit_request, including the
failed/error metrics updated in the shown path.
| failed = self.redis.hget(f"future:{self.id}:metrics", "failed") | ||
| if str(failed) == "1": | ||
| raise RuntimeError( | ||
| self.redis.hget(f"future:{self.id}:metrics", "error_message") | ||
| or "Unknown error" | ||
| ) | ||
|
|
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Propagate failed futures to registered consumers.
When failed == 1, value() raises before it reaches _notify_consumers() on Lines 179-180. _notify_consumers() sends only a success payload. A remote dependency handled by ventis/controller/local_controller.py::_resolve_future_args can therefore wait for its 300-second timeout instead of receiving the error through WriteResult.
Send failed=1 and error_message to each consumer. Make the notification idempotent. Add an integration test for a registered consumer and a failed future.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/future.py` around lines 156 - 162, Update Future.value and
_notify_consumers so failed futures propagate failed=1 and the stored
error_message to every registered consumer via WriteResult before raising. Make
consumer notification idempotent so the failure is sent at most once, and add an
integration test covering a registered consumer receiving a failed future
instead of timing out.
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
tests/test_telemetry_logging.py (3)
341-352: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winTest an unrecognized model value.
The fixture omits
model. It tests the missing-model default, not an unknown pricing key. Setmodelto an unrecognized value so this test covers a pricing lookup miss.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_telemetry_logging.py` around lines 341 - 352, Update the _FakeRedis fixture in test_total_cost_defaults_to_zero_for_unknown_model to include a model field with an unrecognized pricing value, while preserving the rest of the test data and assertions so the test exercises a pricing lookup miss rather than a missing-model default.
95-114: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winRestore process environment state after each test.
These fixtures overwrite process-global environment variables. They then delete values instead of restoring prior values. Later tests can receive a deleted SQLite database URL or lose configured demo multipliers.
tests/test_telemetry_logging.py#L95-L114: save and restore the priorVENTIS_DATABASE_URL.tests/test_telemetry_logging.py#L410-L423: restore prior multiplier values infinally.tests/test_session_logging.py#L21-L47: save and restore the priorVENTIS_DATABASE_URL.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_telemetry_logging.py` around lines 95 - 114, Restore process-global environment state instead of deleting or overwriting it. In tests/test_telemetry_logging.py lines 95-114, update RuntimeSqlalchemyTests.setUp/tearDown to save and restore the prior VENTIS_DATABASE_URL; in tests/test_telemetry_logging.py lines 410-423, preserve prior multiplier values and restore them in finally; in tests/test_session_logging.py lines 21-47, save and restore the prior VENTIS_DATABASE_URL in the fixture lifecycle.
319-323: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winHandle the canonical Claude Haiku 4.5 model ID in pricing.
compute_token_cost()uses exact model ID lookup, andbedrock.pypersists telemetry frommodelIdunchanged. AWS documents the canonical ID asanthropic.claude-haiku-4-5-20251001-v1:0; update the priced model ID and add test coverage so telemetry using the canonical ID does not fall back to the zero-cost unknown-model path.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_telemetry_logging.py` around lines 319 - 323, Update the Claude Haiku 4.5 priced model entry used by compute_token_cost() to the canonical ID anthropic.claude-haiku-4-5-20251001-v1:0, and revise the telemetry test fixture in the pricing coverage to use that ID. Ensure the test verifies canonical-ID telemetry receives its configured token cost rather than the unknown-model zero-cost fallback.Source: MCP tools
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@ventis/controller/utils/session_logging.py`:
- Around line 27-31: The session upsert in _SESSION_UPSERT must conflict on the
project-scoped key (project_id, session_id) instead of session_id alone, while
preserving project ownership during updates. Update the test DDL in
tests/test_session_logging.py:32-34 to define the matching unique constraint,
and add a same-session-ID/different-project collision test near
tests/test_session_logging.py:150-158 verifying each project retains and reads
its own row.
- Around line 91-100: Normalize the JSONB fields in get_session() before
returning the row so PostgreSQL dictionaries and SQLite text values produce the
same payload type; decode text values within session_logging.py and return the
normalized dictionary. Update tests/test_session_logging.py at lines 84-110 and
127-141 to decode only where needed and assert the normalized get_session() type
directly.
- Around line 45-51: Update _get_engine to require authenticated TLS for
PostgreSQL connections by ensuring the URL passed to create_engine uses
sslmode=verify-full, preserving any existing query parameters and avoiding
duplicate sslmode values. Apply this to the dialect-rewritten
VENTIS_DATABASE_URL/database_url before initializing _engine.
---
Outside diff comments:
In `@tests/test_telemetry_logging.py`:
- Around line 341-352: Update the _FakeRedis fixture in
test_total_cost_defaults_to_zero_for_unknown_model to include a model field with
an unrecognized pricing value, while preserving the rest of the test data and
assertions so the test exercises a pricing lookup miss rather than a
missing-model default.
- Around line 95-114: Restore process-global environment state instead of
deleting or overwriting it. In tests/test_telemetry_logging.py lines 95-114,
update RuntimeSqlalchemyTests.setUp/tearDown to save and restore the prior
VENTIS_DATABASE_URL; in tests/test_telemetry_logging.py lines 410-423, preserve
prior multiplier values and restore them in finally; in
tests/test_session_logging.py lines 21-47, save and restore the prior
VENTIS_DATABASE_URL in the fixture lifecycle.
- Around line 319-323: Update the Claude Haiku 4.5 priced model entry used by
compute_token_cost() to the canonical ID
anthropic.claude-haiku-4-5-20251001-v1:0, and revise the telemetry test fixture
in the pricing coverage to use that ID. Ensure the test verifies canonical-ID
telemetry receives its configured token cost rather than the unknown-model
zero-cost fallback.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 69001493-e3e3-4862-bc1e-cd70523ef305
⛔ Files ignored due to path filters (1)
ventis/controller/utils/aws_pricing_chart.dbis excluded by!**/*.db
📒 Files selected for processing (10)
examples/portfolio/config/global_controller.yamltests/test_session_logging.pytests/test_telemetry_logging.pyventis/controller/global_controller.pyventis/controller/utils/pricing.pyventis/controller/utils/session_logging.pyventis/controller/utils/telemetry_logging.pyventis/deploy.pyventis/llm/bedrock.pyventis/stub_generator.py
💤 Files with no reviewable changes (3)
- ventis/llm/bedrock.py
- examples/portfolio/config/global_controller.yaml
- ventis/controller/utils/telemetry_logging.py
🚧 Files skipped from review as they are similar to previous changes (4)
- ventis/controller/global_controller.py
- ventis/deploy.py
- ventis/stub_generator.py
- ventis/controller/utils/pricing.py
| ON CONFLICT (session_id) DO UPDATE SET | ||
| status = excluded.status, | ||
| updated_at = excluded.updated_at, | ||
| input = COALESCE(excluded.input, {SESSION_TABLE_NAME}.input), | ||
| output = COALESCE(excluded.output, {SESSION_TABLE_NAME}.output) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
ast-grep outline ventis/deploy.py --items all --type function
rg -n -C 5 --glob '*.py' '\b(uuid4|request_id|session_id|upsert_session)\b' ventis tests
fd -HI -t f -E .git -e py -e sql -e yml -e yaml . |
xargs -r rg -n -C 5 'CREATE TABLE[[:space:]]+session|PRIMARY KEY.*session_id|UNIQUE.*session_id|ON CONFLICT'Repository: CanyonCodeCoreAI/canyoncodecore
Length of output: 50387
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== session logging implementation =="
sed -n '1,120p' ventis/controller/utils/session_logging.py
echo "== deploy request/session flow =="
sed -n '80,240p' ventis/deploy.py
echo "== session logging tests =="
sed -n '1,220p' tests/test_session_logging.py
echo "== direct request_id generation/usages =="
rg -n -C 3 --glob '*.py' 'uuid\.uuid4\(\)\.hex|get_request_id|request_id\s*=' ventis/deploy.py tests/test_session_logging.pyRepository: CanyonCodeCoreAI/canyoncodecore
Length of output: 16414
🌐 Web query:
PostgreSQL ON CONFLICT DO UPDATE retains current row column value if excluded column value not used documentation
💡 Result:
In PostgreSQL, when using an ON CONFLICT DO UPDATE clause, any column not explicitly included in the SET assignment list will retain its original, pre-existing value from the row that conflicted [1][2][3]. The ON CONFLICT DO UPDATE operation functions as a standard UPDATE statement for the conflicting row [1][2]. The SET clause determines which columns are modified and what their new values will be [2][3]. If a column is omitted from the SET clause, PostgreSQL does not modify that column, meaning it keeps its current value [2][4]. You can reference the existing row's current values by using the table's name (or an alias), and you can reference the values proposed for insertion using the special EXCLUDED table [1][2]. If you do not explicitly assign a value—either from the EXCLUDED table, a literal, or an expression involving the existing row—the column remains unchanged [1][2][5]. For example, if you have a table users with columns (id, name, email) and run the following: INSERT INTO users (id, name, email) VALUES (1, 'Alice', 'alice@example.com') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name; In this case, if a conflict on the id column occurs, the name will be updated to 'Alice', but the email column will retain whatever value was already stored in that row because it was not mentioned in the SET clause [1][2][5].
Citations:
- 1: https://www.postgresql.org/docs/19/sql-insert.html
- 2: https://www.postgresql.org/docs/current/sql-insert.html
- 3: https://www.postgresql.org/docs/17/sql-insert.html
- 4: https://www.postgresql.org/docs/18/sql-insert.html
- 5: https://stackoverflow.com/questions/36359440/postgresql-insert-on-conflict-update-upsert-use-all-excluded-values
Make the session upsert match the project-scoped read.
get_session() filters by (project_id, session_id), but _SESSION_UPSERT conflicts on session_id only and preserves the existing project_id. A different project uploading the same ID can update the first project’s status/output while the first project can still read that row through its scoped lookup. Use (project_id, session_id) as the conflict target, mirror that unique key in the test DDL, and add a same-ID, different-project collision test.
📍 Affects 2 files
ventis/controller/utils/session_logging.py#L27-L31(this comment)tests/test_session_logging.py#L32-L34tests/test_session_logging.py#L150-L158
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/controller/utils/session_logging.py` around lines 27 - 31, The session
upsert in _SESSION_UPSERT must conflict on the project-scoped key (project_id,
session_id) instead of session_id alone, while preserving project ownership
during updates. Update the test DDL in tests/test_session_logging.py:32-34 to
define the matching unique constraint, and add a
same-session-ID/different-project collision test near
tests/test_session_logging.py:150-158 verifying each project retains and reads
its own row.
| def _get_engine(database_url): | ||
| global _engine | ||
| if _engine is None: | ||
| url = os.environ.get("VENTIS_DATABASE_URL", str(database_url)) | ||
| if url.startswith("postgresql://"): | ||
| url = "postgresql+psycopg://" + url[len("postgresql://"):] | ||
| _engine = create_engine(url) |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C 3 \
--glob '*.py' --glob '*.yml' --glob '*.yaml' --glob '*.toml' \
'VENTIS_DATABASE_URL|PGSSLMODE|sslmode|postgresql(\+psycopg)?://' .Repository: CanyonCodeCoreAI/canyoncodecore
Length of output: 10781
🌐 Web query:
PostgreSQL libpq sslmode prefer default verify-full authenticated TLS documentation
💡 Result:
In PostgreSQL libpq, sslmode is a connection parameter that determines the priority and security level of SSL/TLS negotiations [1][2]. The current default value for sslmode is prefer [3][1]. The sslmode settings are summarized below: - prefer (default): libpq first attempts an SSL connection; if that fails, it falls back to a non-SSL connection [1][4]. It does not provide protection against Man-in-the-Middle (MITM) attacks and provides only opportunistic encryption [3][5]. - verify-full: libpq only attempts an SSL connection, verifies that the server certificate is issued by a trusted Certificate Authority (CA), and confirms that the requested server hostname matches the identity stored in the server certificate [1][2]. This mode provides both encryption and protection against MITM attacks, and is recommended for security-sensitive environments [3][6]. Key Security Considerations: - The default prefer mode is considered insecure because it is susceptible to MITM attacks and is only maintained for backward compatibility [3][7]. - When using sslrootcert=system, the default sslmode is automatically elevated to verify-full, and any weaker setting will result in an error [1][2]. - verify-full is the most secure option, as it ensures both encryption and proper server identity verification [3][5]. For secure deployments, it is recommended to explicitly set sslmode to verify-full [3][7].
Citations:
- 1: https://www.postgresql.org/docs/18/libpq-connect.html
- 2: https://www.postgresql.org/docs/current/libpq-connect.html
- 3: https://www.postgresql.org/docs/current/libpq-ssl.html
- 4: https://ongres.com/blog/ssl_mode_behavior_in_authentication_hooked_extensions/
- 5: https://www.postgresql.org/docs/18/libpq-ssl.html
- 6: https://www.postgresql.org/docs/19/libpq-ssl.html
- 7: https://aembit.io/blog/the-strange-world-of-postgres-tls/
Security Misconfiguration (CWE-319): Cleartext Transmission of Sensitive Information
Reachability: Internal
Reachability path
● Entry
ventis/deploy.py:85
_expire_request_keys: Let a finished request's Redis keys age out instead of living forever.
│
▼
● Sink
ventis/controller/utils/session_logging.py
Require authenticated TLS for PostgreSQL session logging.
VENTIS_DATABASE_URL is passed through create_engine() after a dialect rewrite. For remote PostgreSQL deployments, sslmode=prefer is the libpq default and can fall back to plaintext; use sslmode=verify-full for authenticated hostname validation.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/controller/utils/session_logging.py` around lines 45 - 51, Update
_get_engine to require authenticated TLS for PostgreSQL connections by ensuring
the URL passed to create_engine uses sslmode=verify-full, preserving any
existing query parameters and avoiding duplicate sslmode values. Apply this to
the dialect-rewritten VENTIS_DATABASE_URL/database_url before initializing
_engine.
Source: MCP tools
| with _get_engine(database_url).connect() as conn: | ||
| row = ( | ||
| conn.execute( | ||
| _SESSION_SELECT, | ||
| {"session_id": session_id, "project_id": project_id}, | ||
| ) | ||
| .mappings() | ||
| .first() | ||
| ) | ||
| return dict(row) if row is not None else None |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
git ls-files | rg '(^|/)(session_logging.py|test_session_logging.py|requirements.*|pyproject.toml|setup.cfg|Pipfile|poetry.lock)$' || true
echo "== session_logging outline =="
ast-grep outline ventis/controller/utils/session_logging.py || true
echo "== tests outline =="
sed -n '1,180p' tests/test_session_logging.py
echo "== relevant source =="
cat -n ventis/controller/utils/session_logging.py | sed -n '1,170p'
echo "== session logging calls =="
rg -n "get_session|upsert_session|session_logging" -S . || trueRepository: CanyonCodeCoreAI/canyoncodecore
Length of output: 15493
🌐 Web query:
psycopg3 JSONB default Python type loads returns json object documentation
💡 Result:
In Psycopg 3, JSONB fields are automatically deserialized into Python objects (such as dicts or lists) when queried, using the standard library json.loads by default [1][2]. You do not need to perform manual conversion when reading these values [1]. To insert Python objects into JSONB fields, you must explicitly wrap them using the psycopg.types.json.Jsonb wrapper [1][3]. Psycopg 3 does not automatically convert Python dictionaries to JSONB to prevent ambiguity, as many different Python types could potentially be represented as JSON [1]. If you wish to customize how JSON is loaded or dumped (e.g., to use a different library like ujson or to change serialization parameters), you can use the following configuration functions from the psycopg.types.json module: 1. set_json_loads(func, context=None): Configures the function used to parse JSON data when fetching from the database [1][2]. 2. set_json_dumps(func, context=None): Configures the function used to serialize Python objects for storage [1][2]. These functions can be applied globally or restricted to a specific connection or cursor by passing the context parameter [1][2]. For fine-grained control at the individual query level, you can also pass a dumps function directly to the Json or Jsonb wrapper [1].
Citations:
- 1: https://www.psycopg.org/psycopg3/docs/basic/adapt.html
- 2: https://access.crunchydata.com/documentation/psycopg3/latest/api/types.html
- 3: https://stackoverflow.com/questions/77130229/psycopg3-inserting-dict-into-jsonb-field
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== dependency specs =="
rg -n "psycopg|sqlalchemy|sqlite" pyproject.toml requirements.txt || true
echo "== /status usage in deploy =="
sed -n '180,225p' ventis/deploy.py
echo "== python sqlite jsonb probe =="
python3 - <<'PY'
import json
import sqlite3
from pathlib import Path
with sqlite3.connect(":memory:") as conn:
conn.execute("CREATE TABLE session (output JSONB);")
conn.execute("INSERT INTO session (output) VALUES (?)", (json.dumps({"result": 42}),))
row = conn.execute("SELECT output FROM session").fetchone()
print(type(row[0]), row[0])
print("parse-able:", isinstance(row[0], str), row[0] == '{"result": 42}' or row[0] == {'result': 42})
PY
echo "== python postgres probe unavailable? =="
python3 - <<'PY'
try:
import psycopg
print("psycopg found", psycopg.__version__ if hasattr(psycopg, "__version__") else "")
except Exception as exc:
print("psycopg unavailable:", type(exc).__name__, exc)
PYRepository: CanyonCodeCoreAI/canyoncodecore
Length of output: 2156
Normalize session JSON payloads before returning them.
get_session() currently lets psycopg 3 return decoded JSONB values as dictionaries, while SQLite tests expose JSON stored as text. That makes the durable status contract backend-specific. Return a normalized payload type in session_logging.py; decode in the tests only when needed, and assert the normalized get_session() type instead of parsing strings.
📍 Affects 2 files
ventis/controller/utils/session_logging.py#L91-L100(this comment)tests/test_session_logging.py#L84-L110tests/test_session_logging.py#L127-L141
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@ventis/controller/utils/session_logging.py` around lines 91 - 100, Normalize
the JSONB fields in get_session() before returning the row so PostgreSQL
dictionaries and SQLite text values produce the same payload type; decode text
values within session_logging.py and return the normalized dictionary. Update
tests/test_session_logging.py at lines 84-110 and 127-141 to decode only where
needed and assert the normalized get_session() type directly.
Source: MCP tools
| "disk_percent": str(psutil.disk_usage("/").percent), | ||
| "memory_percent": str(psutil.virtual_memory().percent), | ||
| "uptime_seconds": str(max(time.time() - psutil.boot_time(), 0.0)), | ||
| "queue_length": str(self._executor._work_queue.qsize()), |
There was a problem hiding this comment.
Reevaluate if we need a thread
| # ---- requirements.txt ------------------------------------------------ | ||
| requirements = "grpcio\nredis\npyyaml\nboto3\nyfinance\n" | ||
| # psutil is required unconditionally -- local_controller.py imports it at | ||
| # module level for CPU/disk/memory metrics reporting on every agent. |
There was a problem hiding this comment.
Eventually we need a dependency file from the yaml
2 participants
Telemetry Feature Addition
Summary by CodeRabbit
New Features
Bug Fixes
Documentation