Skip to content

Fix stub/real-file collision in per-agent Docker builds #19

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
iidsample merged 2 commits into from
Jul 22, 2026
Merged

Fix stub/real-file collision in per-agent Docker builds #19

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
52d3bcb
added some fixes
Saaketh0 Jul 22, 2026
8bada23
fixed threading count
Saaketh0 Jul 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ventis/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,7 +188,7 @@ def cmd_build(args):
stub_paths = []
for yaml_path in yaml_files:
base_name = os.path.splitext(os.path.basename(yaml_path))[0]
output_path = os.path.join(stubs_dir, f"{base_name}_stub.py")
output_path = os.path.join(stubs_dir, f"{base_name}.py")
logger.info("Generating stub: %s -> %s", yaml_path, output_path)
generate_stub(yaml_path, output_path)
stub_paths.append(output_path)
Expand Down
2 changes: 1 addition & 1 deletion ventis/controller/instance_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ def ensure_instances(self, agent_specs):
}
)

max_workers = int(os.environ.get("VENTIS_MAX_AGENT_INSTANCES", 8))
max_workers = min(len(jobs),os.cpu_count()*100)

@coderabbitai coderabbitai Bot Jul 22, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== file map =="
git ls-files 'ventis/controller/instance_manager.py' 'ventis/**' | sed -n '1,120p'

echo
echo "== outline =="
ast-grep outline ventis/controller/instance_manager.py --view expanded || true

echo
echo "== relevant excerpt =="
nl -ba ventis/controller/instance_manager.py | sed -n '1,140p'

echo
echo "== search for cpu_count / max_workers / provisioning caps =="
rg -n "cpu_count|max_workers|VENTIS_MAX_AGENT_INSTANCES|ThreadPoolExecutor|ProcessPoolExecutor|executor" ventis -S || true

echo
echo "== runtime probe: None * 100 behavior =="
python3 - <<'PY'
import os
try:
    print(min(3, os.cpu_count()*100))
except Exception as e:
    print(type(e).__name__, e)
PY

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 2516

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== relevant excerpt =="
sed -n '31,115p' ventis/controller/instance_manager.py | cat -n

echo
echo "== runtime probe: cpu_count None behavior =="
python3 - <<'PY'
import os
vals = [None, 0, 1, 8]
for v in vals:
    try:
        result = min(3, v * 100)
        print(f"{v!r}: {result}")
    except Exception as e:
        print(f"{v!r}: {type(e).__name__}: {e}")
PY

echo
echo "== search for max_workers and executor usage =="
rg -n "max_workers|ThreadPoolExecutor|ProcessPoolExecutor|cpu_count|ensure_instances|_provision_one" ventis/controller/instance_manager.py -n -S || true

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 4685

Handle os.cpu_count() returning None. os.cpu_count() * 100 raises TypeError on hosts where the CPU count is unavailable, so ensure_instances() crashes before provisioning starts. Use a fallback like os.cpu_count() or 1 and add a regression test for the None case.

🤖 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` at line 74, Update ensure_instances()
to normalize os.cpu_count() with a fallback of 1 before multiplying by 100,
preserving the existing max_workers calculation. Add a regression test that
mocks os.cpu_count() to return None and verifies provisioning proceeds without a
TypeError.

🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Map candidate file structure
ast-grep outline ventis/controller/instance_manager.py --view expanded || true

# Show relevant section with line numbers
sed -n '1,220p' ventis/controller/instance_manager.py

# Search for max_workers / executor usage in repo
rg -n "max_workers|ThreadPoolExecutor|ProcessPoolExecutor|concurrent\.futures|executor" ventis -S

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 10485

🏁 Script executed:

#!/bin/bash
set -euo pipefail

ast-grep outline ventis/controller/instance_manager.py --view expanded || true
sed -n '1,220p' ventis/controller/instance_manager.py
rg -n "max_workers|ThreadPoolExecutor|ProcessPoolExecutor|concurrent\.futures|executor" ventis -S

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 10485

🏁 Script executed:

#!/bin/bash
set -euo pipefail

pwd
git ls-files | rg '^ventis/controller/instance_manager\.py$|^ventis/controller/'
sed -n '1,220p' ventis/controller/instance_manager.py
rg -n "max_workers|ThreadPoolExecutor|ProcessPoolExecutor|concurrent\.futures|executor" ventis -S

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 9948

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Locate the file and print a small window around the target line.
file=$(git ls-files | rg '^ventis/controller/instance_manager\.py$' || true)
printf 'FILE=%s\n' "$file"
if [ -n "$file" ]; then
  nl -ba "$file" | sed -n '1,180p'
fi

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 252

🏁 Script executed:

#!/bin/bash
set -euo pipefail

python3 - <<'PY'
import os
print("cwd:", os.getcwd())
print("has ventis/controller/instance_manager.py:", os.path.exists("ventis/controller/instance_manager.py"))
PY

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 241

🏁 Script executed:

#!/bin/bash
set -euo pipefail

rg -n "def provision_instance|def bootstrap_instance|terminate_instance|validate_config" ventis/controller/cloud_provider_logic -S
sed -n '1,220p' ventis/controller/cloud_provider_logic/Local/_runtime.py
sed -n '1,260p' ventis/controller/cloud_provider_logic/EC2/_runtime.py

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 13069

Add a hard cap for provisioning concurrency. This pool can still grow to 100 * CPU count, which means thousands of simultaneous Docker/AWS calls on large hosts. Keep the limit separately configurable and fall back safely when CPU count is unavailable.

🤖 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` at line 74, Update the worker-count
calculation around max_workers to apply a separately configurable hard cap for
provisioning concurrency, while retaining the CPU-based limit as the lower
bound. Handle os.cpu_count() returning None safely by using an appropriate
fallback before calculating the limit.

provisioned = []
if jobs:
with ThreadPoolExecutor(max_workers=max_workers) as executor:
Expand Down
19 changes: 19 additions & 0 deletions ventis/controller/utils/sqlalchemy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,12 +32,31 @@
)


_CREATE_TABLE = text(
f"""
CREATE TABLE IF NOT EXISTS {TABLE_NAME} (
future_id TEXT PRIMARY KEY,
session_id TEXT,
workflow TEXT,
agent TEXT,
execution_time REAL,
cpu_resource REAL,
gpu_resource REAL,
created_at TEXT,
updated_at TEXT
)
"""
)
Comment on lines +35 to +49

@coderabbitai coderabbitai Bot Jul 22, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Map the target file and nearby symbols first.
ast-grep outline ventis/controller/utils/sqlalchemy.py --view expanded || true

# Show the file with line numbers.
cat -n ventis/controller/utils/sqlalchemy.py

# Find where runtime_information is written/read.
rg -n "created_at|updated_at|send_data|runtime_information|_engine|CREATE TABLE" ventis/controller -S

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 6625

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Read only the relevant sections if the file is long.
sed -n '1,220p' ventis/controller/utils/sqlalchemy.py

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 3658

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# If tests or examples reference persistence, inspect them too.
rg -n "runtime_information|created_at|updated_at|session_id|future_id" -S .

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 7769

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Check for line counts before reading more if needed.
wc -l ventis/controller/utils/sqlalchemy.py

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 211

🏁 Script executed:

#!/bin/bash
set -euo pipefail

cat -n tests/test_runtime_sqlalchemy.py

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 5069

🏁 Script executed:

#!/bin/bash
set -euo pipefail

python3 - <<'PY'
import sqlite3

for decl in ["TEXT", "REAL", "NUMERIC", "INTEGER"]:
    conn = sqlite3.connect(":memory:")
    conn.execute(f"CREATE TABLE t (v {decl})")
    conn.execute("INSERT INTO t(v) VALUES (?)", (1.5,))
    value, typ = conn.execute("SELECT v, typeof(v) FROM t").fetchone()
    print(decl, repr(value), typ)
PY

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 233

🏁 Script executed:

#!/bin/bash
set -euo pipefail

sed -n '1,220p' tests/test_runtime_sqlalchemy.py

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 4166

🏁 Script executed:

#!/bin/bash
set -euo pipefail

sed -n '1,180p' tests/test_runtime_sqlalchemy.py

Repository: CanyonCodeCoreAI/canyoncodecore

Length of output: 4166

Keep the timestamp columns numeric. send_data writes epoch floats into created_at and updated_at, so declaring them as TEXT stores string values in SQLite and leaves the schema contract inconsistent. Use REAL here, or serialize timestamps to text everywhere.

Proposed schema fix 
-        created_at TEXT,
-        updated_at TEXT
+        created_at REAL,
+        updated_at REAL
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
_CREATE_TABLE = text(
f"""
CREATE TABLE IF NOT EXISTS {TABLE_NAME} (
future_id TEXT PRIMARY KEY,
session_id TEXT,
workflow TEXT,
agent TEXT,
execution_time REAL,
cpu_resource REAL,
gpu_resource REAL,
created_at TEXT,
updated_at TEXT
)
"""
)
_CREATE_TABLE = text(
f"""
CREATE TABLE IF NOT EXISTS {TABLE_NAME} (
future_id TEXT PRIMARY KEY,
session_id TEXT,
workflow TEXT,
agent TEXT,
execution_time REAL,
cpu_resource REAL,
gpu_resource REAL,
created_at REAL,
updated_at REAL
)
"""
)
🤖 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 35 - 49, Update the
CREATE TABLE schema in _CREATE_TABLE so the created_at and updated_at columns
use REAL instead of TEXT, matching the epoch-float values written by send_data.
Leave the remaining column definitions unchanged.



def _get_engine(database_url):
global _engine
if _engine is None:
_engine = create_engine(
os.environ.get("VENTIS_DATABASE_URL", str(database_url))
)
with _engine.begin() as conn:
conn.execute(_CREATE_TABLE)
Comment on lines 52 to +59

@coderabbitai coderabbitai Bot Jul 22, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Publish the engine only after table initialization succeeds.

_engine is assigned before CREATE TABLE completes. A concurrent caller can observe the non-None engine and execute an upsert before the table exists; if creation fails, the failed engine remains cached and initialization is never retried. Build the engine locally, initialize it, then assign it under a lock.

Proposed initialization fix 
 def _get_engine(database_url):
     global _engine
     if _engine is None:
-        _engine = create_engine(
-            os.environ.get("VENTIS_DATABASE_URL", str(database_url))
-        )
-        with _engine.begin() as conn:
-            conn.execute(_CREATE_TABLE)
+        with _engine_lock:
+            if _engine is None:
+                engine = create_engine(
+                    os.environ.get("VENTIS_DATABASE_URL", str(database_url))
+                )
+                with engine.begin() as conn:
+                    conn.execute(_CREATE_TABLE)
+                _engine = engine
     return _engine
🤖 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 52 - 59, Update
_get_engine to build the SQLAlchemy engine in a local variable, complete the
_CREATE_TABLE initialization, and only then publish it to _engine while holding
an initialization lock. Ensure concurrent callers cannot observe a partially
initialized engine, and leave _engine unset when initialization fails so later
calls can retry.

return _engine


Expand Down
3 changes: 2 additions & 1 deletion ventis/stub_generator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -302,7 +302,6 @@ def generate_docker(
# Copy general agent files
files_to_copy = [
# (source_path, destination_filename)
(os.path.abspath(agent_file), os.path.basename(agent_file)),
(os.path.join(script_dir, "future.py"), "future.py"),
(os.path.join(script_dir, "ventis_context.py"), "ventis_context.py"),
(
Expand All @@ -322,6 +321,8 @@ def generate_docker(
files_to_copy.append(
(os.path.abspath(stub_file), os.path.basename(stub_file))
)

files_to_copy.append((os.path.abspath(agent_file), os.path.basename(agent_file)))

# Copy gRPC generated stubs if they exist
if os.path.isdir(grpc_stubs_dir):
Expand Down
Loading