diff --git a/examples/finance/workflow/example_workflow.py b/examples/finance/workflow/example_workflow.py index cac18ad..e4b6ce5 100644 --- a/examples/finance/workflow/example_workflow.py +++ b/examples/finance/workflow/example_workflow.py @@ -18,8 +18,8 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "grpc_stubs")) from deploy import deploy -from finance_agent import FinanceAgent -from market_agent import MarketResearchAgent +from agents.finance_agent import FinanceAgent +from agents.market_agent import MarketResearchAgent def main(ticker: str = "AAPL"): diff --git a/examples/helloworld/workflow/example_workflow.py b/examples/helloworld/workflow/example_workflow.py index 842fe80..6bafff3 100644 --- a/examples/helloworld/workflow/example_workflow.py +++ b/examples/helloworld/workflow/example_workflow.py @@ -15,7 +15,7 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "grpc_stubs")) from deploy import deploy -from example_agent import ExampleAgent +from agents.example_agent import ExampleAgent def main(name: str = "World"): diff --git a/examples/portfolio/workflow/portfolio_workflow.py b/examples/portfolio/workflow/portfolio_workflow.py index 44a1484..619c4bd 100644 --- a/examples/portfolio/workflow/portfolio_workflow.py +++ b/examples/portfolio/workflow/portfolio_workflow.py @@ -31,10 +31,10 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "grpc_stubs")) from deploy import deploy -from intent_agent import IntentAgent -from metrics_agent import MetricsAgent -from risk_agent import RiskAgent -from advisor_agent import AdvisorAgent +from agents.intent_agent import IntentAgent +from agents.metrics_agent import MetricsAgent +from agents.risk_agent import RiskAgent +from agents.advisor_agent import AdvisorAgent def main( diff --git a/examples/text2sql/workflow/text2sql_workflow.py b/examples/text2sql/workflow/text2sql_workflow.py index d2f32eb..ac9801d 100644 --- a/examples/text2sql/workflow/text2sql_workflow.py +++ b/examples/text2sql/workflow/text2sql_workflow.py @@ -25,11 +25,11 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "grpc_stubs")) from deploy import deploy -from schema_agent import SchemaRetrievalAgent -from sql_generator_agent import SQLGeneratorAgent -from sql_validator_agent import SQLValidatorAgent -from sandbox_agent import SandboxExecutorAgent -from production_agent import ProductionExecutorAgent +from agents.schema_agent import SchemaRetrievalAgent +from agents.sql_generator_agent import SQLGeneratorAgent +from agents.sql_validator_agent import SQLValidatorAgent +from agents.sandbox_agent import SandboxExecutorAgent +from agents.production_agent import ProductionExecutorAgent def main(question: str = "total order amount per customer region", n_candidates: int = 3): diff --git a/tests/run_tests.sh b/tests/run_tests.sh index e9f8386..c5556ec 100755 --- a/tests/run_tests.sh +++ b/tests/run_tests.sh @@ -30,13 +30,10 @@ cd "$TEST_DIR" echo ">> 1. Generating new project..." ventis new-project $PROJECT_NAME cd $PROJECT_NAME -grep -v 'gpu:' config/global_controller.yaml > config/global_controller.yaml.tmp -mv config/global_controller.yaml.tmp config/global_controller.yaml +grep -v 'gpu:' .car/config/global_controller.yaml > .car/config/global_controller.yaml.tmp +mv .car/config/global_controller.yaml.tmp .car/config/global_controller.yaml -echo ">> 2. Building agents (ventis build)..." -ventis build - -echo ">> 3. Deploying workflow (ventis deploy)..." +echo ">> 2. Building and deploying workflow (ventis deploy)..." ventis deploy & DEPLOY_PID=$! diff --git a/tests/test_cli.py b/tests/test_cli.py index 406b95d..44b9270 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -23,12 +23,14 @@ def _fake_controller_module(self, controller): @patch("atexit.register") @patch("signal.signal") + @patch("ventis.cli._run_build") @patch("ventis.cli._ensure_grpc_stubs_importable") @patch("ventis.cli._preflight_ec2_deploy") def test_deploy_skips_ec2_preflight_for_local_config( self, preflight, ensure_grpc, + _run_build, _signal_patch, _atexit_patch, ): @@ -54,12 +56,14 @@ def test_deploy_skips_ec2_preflight_for_local_config( @patch("atexit.register") @patch("signal.signal") + @patch("ventis.cli._run_build") @patch("ventis.cli._ensure_grpc_stubs_importable") @patch("ventis.cli._preflight_ec2_deploy") def test_deploy_runs_ec2_preflight_for_ec2_config( self, preflight, ensure_grpc, + _run_build, _signal_patch, _atexit_patch, ): @@ -81,6 +85,36 @@ def test_deploy_runs_ec2_preflight_for_ec2_config( preflight.assert_called_once_with(config, os.getcwd()) controller.run.assert_called_once_with() + @patch("atexit.register") + @patch("signal.signal") + @patch("ventis.cli._run_build") + @patch("ventis.cli._ensure_grpc_stubs_importable") + @patch("ventis.cli._preflight_ec2_deploy") + def test_deploy_uses_car_when_present( + self, preflight, ensure_grpc, _run_build, _signal_patch, _atexit_patch + ): + controller = MagicMock() + controller_module = self._fake_controller_module(controller) + args = SimpleNamespace(config=".car/config/global_controller.yaml") + + with tempfile.TemporaryDirectory() as tmpdir, patch( + "ventis.cli.os.path.isfile", return_value=True + ), patch( + "ventis.cli._load_config", return_value={"agents": []} + ), patch.dict( + sys.modules, {"ventis.controller.global_controller": controller_module} + ): + Path(tmpdir, ".car").mkdir() + cwd = os.getcwd() + os.chdir(tmpdir) + try: + cli.cmd_deploy(args) + finally: + os.chdir(cwd) + + ensure_grpc.assert_called_once_with(os.path.join(os.path.realpath(tmpdir), ".car")) + preflight.assert_not_called() + @patch("ventis.cli._ensure_grpc_stubs_importable") @patch("ventis.cli._require_docker_for_ec2") def test_preflight_does_not_require_ssh_fields(self, require_docker, ensure_grpc): @@ -103,28 +137,43 @@ class CliBuildTests(unittest.TestCase): def _run_build( self, project_dir, agent_yaml_paths, buildx_available, platform="linux/amd64" ): - """Run cmd_build against project_dir with docker/subprocess calls mocked. + """Run _run_build against project_dir with docker/subprocess calls mocked. Returns (docker_calls, generate_docker_mock, generate_workflow_docker_mock). """ - config_path = project_dir / "config" / "global_controller.yaml" - args = SimpleNamespace(config=str(config_path)) + artifact_root = ( + project_dir / ".car" if (project_dir / ".car").is_dir() else project_dir + ) + config_path = artifact_root / "config" / "global_controller.yaml" docker_calls = [] def fake_run(cmd, check): docker_calls.append(cmd) return SimpleNamespace(returncode=0) + def fake_glob(pattern): + if pattern.endswith("*.proto"): + return ["proto/a.proto"] + if agent_yaml_paths: + self.assertEqual( + os.path.realpath(Path(pattern).parent), + os.path.realpath(Path(agent_yaml_paths[0]).parent), + ) + return agent_yaml_paths + + def fake_generate_stub(yaml_path, _output_path): + with open(yaml_path) as f: + self.assertIn("agent", yaml.safe_load(f)) + with ( patch( "ventis.cli._get_package_dir", return_value=str(project_dir / "package"), ), + patch("ventis.cli.glob.glob", side_effect=fake_glob), patch( - "ventis.cli.glob.glob", - side_effect=[agent_yaml_paths, ["proto/a.proto"]], + "ventis.stub_generator.generate_stub", side_effect=fake_generate_stub ), - patch("ventis.stub_generator.generate_stub"), patch("ventis.stub_generator.generate_docker") as generate_docker, patch( "ventis.stub_generator.generate_workflow_docker" @@ -136,7 +185,7 @@ def fake_run(cmd, check): cwd = os.getcwd() os.chdir(project_dir) try: - cli.cmd_build(args) + cli._run_build(str(config_path)) finally: os.chdir(cwd) @@ -240,6 +289,32 @@ def test_build_uses_buildx_bake_when_available(self): ) self.assertEqual(targets["workflow"]["tags"], ["ventis-workflow"]) + def test_build_uses_car_when_present(self): + with tempfile.TemporaryDirectory() as tmpdir: + project_dir = Path(tmpdir) + artifact_root = project_dir / ".car" + source_root = artifact_root / "app" + source_root.mkdir(parents=True) + source_yaml = self._write_agent_and_workflow_config(source_root) + source_root.joinpath("config").rename(artifact_root / "config") + agent_yaml = artifact_root / "config" / source_yaml.name + source_yaml.rename(agent_yaml) + + manifest = artifact_root / "config" / "global_controller.yaml" + _, generate_docker, generate_workflow_docker = self._run_build( + project_dir, [str(manifest), str(agent_yaml)], buildx_available=True + ) + + for call in (generate_docker, generate_workflow_docker): + self.assertEqual( + os.path.realpath(call.call_args.kwargs["project_dir"]), + os.path.realpath(source_root), + ) + self.assertEqual( + os.path.realpath(generate_docker.call_args.kwargs["output_dir"]), + os.path.realpath(artifact_root / "docker_container" / "ExampleAgent"), + ) + def test_build_with_no_agents_builds_nothing(self): with tempfile.TemporaryDirectory() as tmpdir: project_dir = Path(tmpdir) @@ -252,7 +327,7 @@ def test_build_with_no_agents_builds_nothing(self): self.assertFalse(any(call[0] == "docker" for call in docker_calls)) - def test_build_skips_agent_without_entrypoint(self): + def test_build_fails_when_stub_cannot_be_generated(self): with tempfile.TemporaryDirectory() as tmpdir: project_dir = Path(tmpdir) (project_dir / "config").mkdir() @@ -268,9 +343,8 @@ def test_build_skips_agent_without_entrypoint(self): ) ) - docker_calls, _, _ = self._run_build(project_dir, [], buildx_available=True) - - self.assertFalse(any(call[0] == "docker" for call in docker_calls)) + with self.assertRaises(SystemExit): + self._run_build(project_dir, [], buildx_available=True) def _write_requirements_config(self, project_dir): """Scaffold one plain agent, one agent with `requirements`, one workflow with `requirements`.""" @@ -373,5 +447,22 @@ def test_build_ignores_non_list_requirements(self): self.assertEqual(generate_docker.call_args.kwargs["requirements"], []) +class CliCleanTests(unittest.TestCase): + def test_clean_uses_car_when_present(self): + with tempfile.TemporaryDirectory() as tmpdir: + project_dir = Path(tmpdir) + (project_dir / ".car" / "stubs").mkdir(parents=True) + (project_dir / "stubs").mkdir() + cwd = os.getcwd() + os.chdir(project_dir) + try: + cli.cmd_clean(SimpleNamespace()) + finally: + os.chdir(cwd) + + self.assertFalse((project_dir / ".car" / "stubs").exists()) + self.assertTrue((project_dir / "stubs").exists()) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_deploy.py b/tests/test_deploy.py index 3f029cb..3c02008 100644 --- a/tests/test_deploy.py +++ b/tests/test_deploy.py @@ -6,7 +6,7 @@ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) -import ventis.deploy as deploy_module +import ventis.controller.deploy as deploy_module class _FakeRedis: diff --git a/tests/test_error_propagation.py b/tests/test_error_propagation.py index 82da262..59c6eba 100644 --- a/tests/test_error_propagation.py +++ b/tests/test_error_propagation.py @@ -17,7 +17,7 @@ from ventis.controller.local_controller import LocalController from ventis.controller.local_controller_frontend import LocalControllerServicer -from ventis.future import Future +from ventis.controller.future import Future import local_controler_pb2 diff --git a/tests/test_future.py b/tests/test_future.py index e190426..4914b29 100644 --- a/tests/test_future.py +++ b/tests/test_future.py @@ -13,8 +13,8 @@ ), ) -import ventis.future as future_module -import ventis.ventis_context as ventis_context +import ventis.controller.future as future_module +import ventis.controller.ventis_context as ventis_context class _FakeRedis: diff --git a/tests/test_global_controller_identity.py b/tests/test_global_controller_identity.py index e4337f7..76a6689 100644 --- a/tests/test_global_controller_identity.py +++ b/tests/test_global_controller_identity.py @@ -103,14 +103,16 @@ def test_a_second_call_with_a_new_config_overwrites_the_published_value(self): }, ) - def test_missing_project_id_or_database_publishes_safe_defaults(self): - controller = _bare_controller({}) + def test_missing_database_publishes_safe_default(self): + # project_id is always populated by _load_config() by the time _write_identity() + # runs -- only database_url has a real "unset" case to default here. + controller = _bare_controller({"project_id": "11111111-1111-1111-1111-111111111111"}) controller._write_identity() self.assertEqual( controller.redis.hgetall(GlobalController.IDENTITY_KEY), - {"project_id": "0", "database_url": ""}, + {"project_id": "11111111-1111-1111-1111-111111111111", "database_url": ""}, ) diff --git a/tests/test_global_controller_project_id.py b/tests/test_global_controller_project_id.py new file mode 100644 index 0000000..1feebc8 --- /dev/null +++ b/tests/test_global_controller_project_id.py @@ -0,0 +1,70 @@ +"""_load_config() must mint a project_id when a config file omits one, and persist it back +to the file so the same value survives a reload_config() or process restart -- not a fresh +uuid on every load. +""" + +import os +import re +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +import yaml + +from ventis.controller.global_controller import GlobalController + +UUID_HEX_RE = re.compile(r"^[0-9a-f]{32}$") + + +def _write_config(body): + f = tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) + f.write(body) + f.close() + return f.name + + +class LoadConfigProjectIdTests(unittest.TestCase): + def test_generates_and_persists_project_id_when_missing(self): + config_path = _write_config("agents: []\npoll_interval: 5\n") + try: + config = GlobalController._load_config(config_path) + + self.assertTrue(UUID_HEX_RE.match(config["project_id"])) + + with open(config_path) as f: + on_disk = yaml.safe_load(f) + self.assertEqual(on_disk["project_id"], config["project_id"]) + finally: + os.unlink(config_path) + + def test_reload_reuses_the_persisted_project_id_instead_of_minting_a_new_one(self): + config_path = _write_config("agents: []\npoll_interval: 5\n") + try: + first = GlobalController._load_config(config_path) + second = GlobalController._load_config(config_path) + + self.assertEqual(first["project_id"], second["project_id"]) + finally: + os.unlink(config_path) + + def test_existing_project_id_is_left_untouched(self): + config_path = _write_config( + 'agents: []\nproject_id: "11111111-1111-1111-1111-111111111111"\n' + ) + try: + config = GlobalController._load_config(config_path) + + self.assertEqual(config["project_id"], "11111111-1111-1111-1111-111111111111") + + with open(config_path) as f: + contents = f.read() + # No second project_id line got appended alongside the existing one. + self.assertEqual(contents.count("project_id"), 1) + finally: + os.unlink(config_path) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_otel_exporter_fields.py b/tests/test_otel_exporter_fields.py index b74177d..f6c2074 100644 --- a/tests/test_otel_exporter_fields.py +++ b/tests/test_otel_exporter_fields.py @@ -29,7 +29,7 @@ def test_init_db_creates_waiting_table_with_full_schema(self): def test_fields_are_normalized_and_added_to_span(self): db.init_db(self.db_path) raw = { - "future_id": "00112233445566778899aabbccddeeff", + "future_id": "0011223344556677", # 64-bit (16 hex chars), matches Future.id's format "request_id": "ffeeddccbbaa99887766554433221100", "service": "PriceAgent", "method": "get_history", @@ -61,7 +61,7 @@ def test_fields_are_normalized_and_added_to_span(self): def test_error_message_is_wired_from_redis_error_field(self): db.init_db(self.db_path) raw = { - "future_id": "11112222333344445555666677778888", + "future_id": "1111222233334444", # 64-bit (16 hex chars), matches Future.id's format "request_id": "88887777666655554444333322221111", "service": "AdvisorAgent", "method": "summarize", diff --git a/tests/test_stub_generator.py b/tests/test_stub_generator.py index fb01f2e..916bf80 100644 --- a/tests/test_stub_generator.py +++ b/tests/test_stub_generator.py @@ -11,6 +11,7 @@ from ventis.stub_generator import ( BASE_AGENT_REQUIREMENTS, BASE_WORKFLOW_REQUIREMENTS, + _stub_destination, generate_docker, generate_workflow_docker, ) @@ -86,5 +87,57 @@ def test_per_workflow_requirements_are_appended_to_base(self): self.assertEqual(requirements, BASE_WORKFLOW_REQUIREMENTS + ["yfinance"]) +class StubDestinationTests(unittest.TestCase): + """A stub replaces the real module at its entrypoint path, so it is written + to exactly that one location. Flat is only a fallback for a stub with no + entrypoint mapping, or one whose mapping escapes the build context. + """ + + def test_unmapped_stub_falls_back_to_flat(self): + self.assertEqual(_stub_destination("/stubs/split_agent.py", {}), "split_agent.py") + + def test_entrypoint_mapping_is_the_only_destination(self): + destination = _stub_destination( + "/stubs/split_agent.py", {"split_agent.py": "agents/split_agent.py"} + ) + self.assertEqual(destination, "agents/split_agent.py") + + def test_flat_entrypoint_stays_flat(self): + destination = _stub_destination( + "/stubs/split_agent.py", {"split_agent.py": "split_agent.py"} + ) + self.assertEqual(destination, "split_agent.py") + + def test_unsafe_entrypoint_falls_back_to_flat(self): + destination = _stub_destination( + "/stubs/split_agent.py", {"split_agent.py": "../../etc/passwd"} + ) + self.assertEqual(destination, "split_agent.py") + + +class GenerateWorkflowDockerStubPlacementTests(unittest.TestCase): + def test_stub_lands_only_at_its_entrypoint_path(self): + with tempfile.TemporaryDirectory() as tmpdir: + workflow_file = Path(tmpdir) / "workflow.py" + workflow_file.write_text("from agents.split_agent import SplitAgent\n") + + stub_file = Path(tmpdir) / "stubs" / "split_agent.py" + stub_file.parent.mkdir() + stub_file.write_text("class SplitAgent:\n pass\n") + + output_dir = os.path.join(tmpdir, "out") + generate_workflow_docker( + str(workflow_file), + [str(stub_file)], + output_dir=output_dir, + stub_entrypoints={"split_agent.py": "agents/split_agent.py"}, + ) + + nested_path = Path(output_dir) / "agents" / "split_agent.py" + flat_path = Path(output_dir) / "split_agent.py" + self.assertIn("class SplitAgent", nested_path.read_text()) + self.assertFalse(flat_path.exists(), "stub must not be duplicated flat") + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_ventis_context.py b/tests/test_ventis_context.py index bec4122..f860d1f 100644 --- a/tests/test_ventis_context.py +++ b/tests/test_ventis_context.py @@ -4,7 +4,7 @@ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) -import ventis.ventis_context as ventis_context +import ventis.controller.ventis_context as ventis_context class VentisContextTests(unittest.TestCase): diff --git a/ventis/Dockerfile b/ventis/Dockerfile new file mode 100644 index 0000000..1814548 --- /dev/null +++ b/ventis/Dockerfile @@ -0,0 +1,20 @@ +FROM python:3.11-slim + +RUN apt-get update && apt-get install -y docker.io && rm -rf /var/lib/apt/lists/* + +COPY . /ventis +RUN pip install /ventis + +# global_controller.py bare-imports these; pip install only ships the .proto source. +RUN python -m grpc_tools.protoc \ + -I/ventis/ventis/controller/proto \ + --python_out=/usr/local/lib/python3.11/site-packages \ + --grpc_python_out=/usr/local/lib/python3.11/site-packages \ + /ventis/ventis/controller/proto/local_controler.proto + +EXPOSE 8000 + +ENTRYPOINT ["python", "-m", "ventis.server"] + + +# to run: docker build -f ventis/Dockerfile -t saakeths/canyonos:latest . diff --git a/ventis/OTLP_Exporter/convert.py b/ventis/OTLP_Exporter/convert.py index 5e6ac74..72e2342 100644 --- a/ventis/OTLP_Exporter/convert.py +++ b/ventis/OTLP_Exporter/convert.py @@ -29,11 +29,12 @@ def waiting_row_to_span(row): row = dict(row) trace_id = int(row["session_id"], 16) - span_id = int.from_bytes(bytes.fromhex(row["future_id"])[:8], "big") + # future_id/parent_id are already 64-bit (Future.id is generated at that + # width directly -- see ventis/controller/future.py), matching OTel's + # span_id, so no truncation is needed here. + span_id = int(row["future_id"], 16) parent_id = row.get("parent_id") - parent_span_id = ( - int.from_bytes(bytes.fromhex(parent_id)[:8], "big") if parent_id else None - ) + parent_span_id = int(parent_id, 16) if parent_id else None context = SpanContext( trace_id=trace_id, span_id=span_id, is_remote=False, trace_flags=_SAMPLED diff --git a/ventis/OTLP_Exporter/db.py b/ventis/OTLP_Exporter/db.py index 005ba71..f1438f7 100644 --- a/ventis/OTLP_Exporter/db.py +++ b/ventis/OTLP_Exporter/db.py @@ -134,21 +134,30 @@ def write_waiting_rows(rows, redis_client=None, project_id=None, db_path=DB_PATH # Cost figures are only meaningful once the future has finished, so skip # computing them until then rather than recomputing on every poll. if finished_at is not None: - token_cost = ( - pricing.compute_token_cost( - raw.get("model"), input_token_count, output_token_count + # Cost lookups can fail independently of the telemetry itself (e.g. + # no aws_instance_pricing table on a local-provider deployment) -- + # don't let that drop the whole row, just cost it at 0. + try: + token_cost = ( + pricing.compute_token_cost( + raw.get("model"), input_token_count, output_token_count + ) + * _TOKEN_COST_MULTIPLIER ) - * _TOKEN_COST_MULTIPLIER - ) - 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, - finished_at - started_at, + except Exception: + token_cost = 0.0 + try: + 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, + finished_at - started_at, + ) + * _SERVER_COST_MULTIPLIER ) - * _SERVER_COST_MULTIPLIER - ) + except Exception: + server_cost = 0.0 else: token_cost = 0.0 server_cost = 0.0 diff --git a/ventis/OTLP_Exporter/otel_exporter.py b/ventis/OTLP_Exporter/otel_exporter.py index 7e365b0..1ed210a 100644 --- a/ventis/OTLP_Exporter/otel_exporter.py +++ b/ventis/OTLP_Exporter/otel_exporter.py @@ -20,7 +20,7 @@ import time sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from ventis.utils.redis_client import RedisClient +from ventis.controller.utils.redis_client import RedisClient from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( OTLPSpanExporter as GrpcOTLPSpanExporter, @@ -243,7 +243,10 @@ def main(): signal.signal(signal.SIGTERM, _handle_shutdown) signal.signal(signal.SIGINT, _handle_shutdown) db.init_db() - _redis = RedisClient() # localhost:6379, db 0 -- same host as GlobalController + # GC reaches its own Redis via host.docker.internal (a sibling container, + # not the same network namespace, since GC runs on bridge networking) -- + # match that instead of plain localhost. + _redis = RedisClient(host="host.docker.internal") _last_destinations_raw = _redis.get(DESTINATIONS_KEY) _processors = _build_processors(_last_destinations_raw) logger.info("OTel exporter process started with %d destination(s).", len(_processors)) diff --git a/ventis/README.md b/ventis/README.md new file mode 100644 index 0000000..6b5675a --- /dev/null +++ b/ventis/README.md @@ -0,0 +1,8 @@ +# CanyonOS Platform + +Every folder in here is a separate process to be run. + +- controller: The control plane and manager +- OTLP_Exporter: The OTel Data Exporter +- server.py: Flask server that CLI connects to +- (soon) Instance_Manager: Responsible for scaling (currently in controller) \ No newline at end of file diff --git a/ventis/cli.py b/ventis/cli.py index c5a2b68..c66a106 100644 --- a/ventis/cli.py +++ b/ventis/cli.py @@ -1,10 +1,10 @@ """ Ventis CLI -Entry point for the `ventis` command. Provides three subcommands: +Entry point for the `ventis` command. Provides these subcommands: ventis new-project — Scaffold a new Ventis project - ventis build — Generate stubs and build Docker images - ventis deploy — Launch agents via the Global Controller + ventis deploy — Build (stubs + Docker images) then launch + agents via the Global Controller """ import argparse @@ -21,7 +21,8 @@ logging.basicConfig(level=logging.INFO) logger = logging.getLogger("ventis") DEFAULT_DOCKER_PLATFORM = "linux/amd64" -DEFAULT_CONFIG_PATH = "config/global_controller.yaml" +ARTIFACT_DIR_NAME = ".car" +SOURCE_DIR_NAME = "app" EC2_REQUIRED_CONFIG_KEYS = ( "ami_id", "subnet_id", @@ -53,6 +54,10 @@ def _load_config(config_path): return yaml.safe_load(f) +def _artifact_prefix(root): + return ARTIFACT_DIR_NAME if os.path.isdir(os.path.join(root, ARTIFACT_DIR_NAME)) else "" + + def _normalize_requirements(agent_cfg): """Return an agent's `requirements` list, or [] if absent/null/malformed.""" requirements = agent_cfg.get("requirements") or [] @@ -175,17 +180,35 @@ def cmd_new_project(args): logger.error("Templates directory not found at %s", templates_dir) sys.exit(1) - # Copy the entire templates tree into the new project - shutil.copytree(templates_dir, project_dir) + # Copy the entire templates tree into .car/app, then pull config and agent + # declarations up into .car/config, keeping generated artifacts (stubs, + # grpc_stubs, docker_container) siblings of the source under .car/. + artifact_root = os.path.join(project_dir, ARTIFACT_DIR_NAME) + source_root = os.path.join(artifact_root, SOURCE_DIR_NAME) + shutil.copytree(templates_dir, source_root) + + source_config = os.path.join(source_root, "config") + artifact_config = os.path.join(artifact_root, "config") + if os.path.isdir(source_config): + shutil.move(source_config, artifact_root) + else: + os.makedirs(artifact_config) + + source_agents = os.path.join(source_root, "agents") + for declaration in glob.glob(os.path.join(source_agents, "*.yaml")): + shutil.move(declaration, artifact_config) + + readme = os.path.join(source_root, "README.md") + if os.path.isfile(readme): + shutil.move(readme, project_dir) # Create empty output directories - os.makedirs(os.path.join(project_dir, "stubs"), exist_ok=True) - os.makedirs(os.path.join(project_dir, "grpc_stubs"), exist_ok=True) + os.makedirs(os.path.join(artifact_root, "stubs"), exist_ok=True) + os.makedirs(os.path.join(artifact_root, "grpc_stubs"), exist_ok=True) logger.info("Created new Ventis project: %s", project_dir) logger.info("") logger.info(" cd %s", project_name) - logger.info(" ventis build") logger.info(" ventis deploy") @@ -194,28 +217,31 @@ def cmd_new_project(args): # ------------------------------------------------------------------ # -def cmd_build(args): +def _run_build(config_path): """ Generate stubs, compile gRPC protos, generate Docker contexts, and build Docker images. - Must be run from the project root (where config/ lives). + Must be run from the project root (where config/ lives). Invoked as the + first phase of `ventis deploy`. """ - config_path = args.config if not os.path.isfile(config_path): logger.error("Config file not found: %s", config_path) sys.exit(1) config = _load_config(config_path) agents = config.get("agents", []) - project_dir = os.getcwd() + project_dir = os.path.abspath(os.getcwd()) + prefix = _artifact_prefix(project_dir) + artifact_root = os.path.join(project_dir, prefix) if prefix else project_dir + source_root = os.path.join(artifact_root, SOURCE_DIR_NAME) if prefix else project_dir package_dir = _get_package_dir() # -------------------------------------------------------------- # # Step 1: Discover agent YAML files and generate Python stubs # # -------------------------------------------------------------- # - agents_dir = os.path.join(project_dir, "agents") - stubs_dir = os.path.join(project_dir, "stubs") + declarations_dir = os.path.join(artifact_root, "config" if prefix else "agents") + stubs_dir = os.path.join(artifact_root, "stubs") os.makedirs(stubs_dir, exist_ok=True) from ventis.stub_generator import ( @@ -224,9 +250,9 @@ def cmd_build(args): generate_workflow_docker, ) - yaml_files = glob.glob(os.path.join(agents_dir, "*.yaml")) + yaml_files = glob.glob(os.path.join(declarations_dir, "*.yaml")) if not yaml_files: - logger.warning("No agent YAML files found in %s", agents_dir) + logger.warning("No agent YAML files found in %s", declarations_dir) import yaml @@ -238,9 +264,19 @@ def cmd_build(args): if name: yaml_by_name[name] = yaml_path - # Maps each generated stub's basename to its agent's entrypoint path, so a - # stub can also be placed at its nested, entrypoint-mirrored location. + # Maps each generated stub's basename to its agent's entrypoint path, which + # is the single location the stub is written to and copied to. entrypoints_by_name = {a["name"]: a.get("entrypoint") for a in agents} + missing_stubs = [ + a["name"] + for a in agents + if a.get("type", "agent") != "workflow" + and (a["name"] not in yaml_by_name or not a.get("entrypoint")) + ] + if missing_stubs: + logger.error("Cannot generate stubs for agents: %s", ", ".join(missing_stubs)) + sys.exit(1) + stub_entrypoints = { f"{os.path.splitext(os.path.basename(p))[0]}.py": entrypoints_by_name[n] for n, p in yaml_by_name.items() @@ -248,9 +284,12 @@ 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}.py") + for agent_name, yaml_path in yaml_by_name.items(): + entrypoint = entrypoints_by_name.get(agent_name) + if not entrypoint: + continue + output_path = os.path.join(stubs_dir, entrypoint) + os.makedirs(os.path.dirname(output_path), exist_ok=True) logger.info("Generating stub: %s -> %s", yaml_path, output_path) generate_stub(yaml_path, output_path) stub_paths.append(output_path) @@ -258,7 +297,7 @@ def cmd_build(args): # -------------------------------------------------------------- # # Step 2: Compile gRPC protobuf stubs # # -------------------------------------------------------------- # - grpc_stubs_dir = os.path.join(project_dir, "grpc_stubs") + grpc_stubs_dir = os.path.join(artifact_root, "grpc_stubs") os.makedirs(grpc_stubs_dir, exist_ok=True) proto_dir = os.path.join(package_dir, "controller", "proto") @@ -296,12 +335,12 @@ def cmd_build(args): ) continue - workflow_path = os.path.join(project_dir, workflow_file) + workflow_path = os.path.join(source_root, workflow_file) if not os.path.isfile(workflow_path): logger.error("Workflow file not found: %s", workflow_path) continue - docker_context = os.path.join(project_dir, "docker_container", "Workflow") + docker_context = os.path.join(artifact_root, "docker_container", "Workflow") logger.info("Generating workflow Docker context for '%s'", agent_name) generate_workflow_docker( workflow_path, @@ -309,10 +348,8 @@ def cmd_build(args): output_dir=docker_context, grpc_stubs_dir=grpc_stubs_dir, api_port=agent_cfg.get("api_port", 8080), - project_dir=project_dir, - stub_entrypoints=stub_entrypoints, + project_dir=source_root, requirements=_normalize_requirements(agent_cfg), - project_dir=project_dir, # Stubs are placed both flat and at their entrypoint-mirrored path, # so both flat and nested import styles resolve to the stub. stub_entrypoints=stub_entrypoints, @@ -327,7 +364,7 @@ def cmd_build(args): ) continue - agent_file = os.path.join(project_dir, entrypoint) + agent_file = os.path.join(source_root, entrypoint) if not os.path.isfile(agent_file): logger.error("Agent file not found: %s", agent_file) continue @@ -341,7 +378,7 @@ def cmd_build(args): ) continue - docker_context = os.path.join(project_dir, "docker_container", agent_name) + docker_context = os.path.join(artifact_root, "docker_container", agent_name) logger.info("Generating Docker context for '%s'", agent_name) generate_docker( matching_yaml, @@ -349,10 +386,8 @@ def cmd_build(args): output_dir=docker_context, grpc_stubs_dir=grpc_stubs_dir, stub_files=stub_paths, - project_dir=project_dir, - stub_entrypoints=stub_entrypoints, + project_dir=source_root, requirements=_normalize_requirements(agent_cfg), - project_dir=project_dir, # Same reasoning as the workflow call above: stubs are placed both # flat and at their entrypoint-mirrored path. stub_entrypoints=stub_entrypoints, @@ -372,7 +407,7 @@ def cmd_build(args): if not bake_targets: logger.info("No Docker images to build.") elif _docker_available() and _docker_available(("docker", "buildx", "version")): - docker_container_dir = os.path.join(project_dir, "docker_container") + docker_container_dir = os.path.join(artifact_root, "docker_container") os.makedirs(docker_container_dir, exist_ok=True) bake_file_path = os.path.join(docker_container_dir, "docker-bake.json") _write_bake_file(bake_targets, bake_file_path, _docker_platform()) @@ -418,24 +453,31 @@ def cmd_deploy(args): logger.error("Config file not found: %s", config_path) sys.exit(1) + # Build first (stubs, protos, Docker contexts, images), then deploy them. + # `ventis build` was merged into `ventis deploy`. + _run_build(config_path) + config = _load_config(config_path) - project_dir = os.getcwd() + project_dir = os.path.abspath(os.getcwd()) + prefix = _artifact_prefix(project_dir) + artifact_root = os.path.join(project_dir, prefix) if prefix else project_dir # Fail here rather than after a fleet of containers is already up without - # the API keys they need. + # the API keys they need. base_dir matches GlobalController, which resolves + # env_file against its cwd -- any other base rejects a file it would find. try: resolve_env_file(config, base_dir=project_dir) except ValueError as e: logger.error("%s", e) sys.exit(1) - _ensure_grpc_stubs_importable(project_dir) + _ensure_grpc_stubs_importable(artifact_root) if any( agent.get("provider", "local").upper() == "EC2" for agent in config.get("agents", []) ): - _preflight_ec2_deploy(config, project_dir) + _preflight_ec2_deploy(config, artifact_root) from ventis.controller.global_controller import GlobalController @@ -476,12 +518,14 @@ def cmd_clean(args): """ Remove generated stubs, gRPC files, and Docker build contexts. """ - project_dir = os.getcwd() + project_dir = os.path.abspath(os.getcwd()) + prefix = _artifact_prefix(project_dir) + artifact_root = os.path.join(project_dir, prefix) if prefix else project_dir paths_to_clean = [ - os.path.join(project_dir, "stubs"), - os.path.join(project_dir, "grpc_stubs"), - os.path.join(project_dir, "docker_container"), + os.path.join(artifact_root, "stubs"), + os.path.join(artifact_root, "grpc_stubs"), + os.path.join(artifact_root, "docker_container"), ] for path in paths_to_clean: @@ -503,6 +547,9 @@ def cmd_clean(args): def main(): + default_config_path = os.path.join( + _artifact_prefix(os.getcwd()), "config", "global_controller.yaml" + ) parser = argparse.ArgumentParser( prog="ventis", description="Ventis — Distributed Agent Orchestration Framework", @@ -517,29 +564,16 @@ def main(): new_proj.add_argument("name", help="Name of the project directory to create") new_proj.set_defaults(func=cmd_new_project) - # ventis build - build = subparsers.add_parser( - "build", - help="Generate stubs, compile protos, and build Docker images", - ) - build.add_argument( - "-c", - "--config", - default=DEFAULT_CONFIG_PATH, - help=f"Path to global controller config (default: {DEFAULT_CONFIG_PATH})", - ) - build.set_defaults(func=cmd_build) - # ventis deploy deploy = subparsers.add_parser( "deploy", - help="Launch agents via the Global Controller", + help="Build stubs/images, then launch agents via the Global Controller", ) deploy.add_argument( "-c", "--config", - default=DEFAULT_CONFIG_PATH, - help=f"Path to global controller config (default: {DEFAULT_CONFIG_PATH})", + default=default_config_path, + help=f"Path to global controller config (default: {default_config_path})", ) deploy.set_defaults(func=cmd_deploy) diff --git a/ventis/llm/bedrock.py b/ventis/controller/bedrock.py similarity index 94% rename from ventis/llm/bedrock.py rename to ventis/controller/bedrock.py index f350b69..97c6a3f 100644 --- a/ventis/llm/bedrock.py +++ b/ventis/controller/bedrock.py @@ -1,8 +1,8 @@ import os try: - from ventis.utils.redis_client import RedisClient - import ventis.ventis_context as ventis_context + from ventis.controller.utils.redis_client import RedisClient + import ventis.controller.ventis_context as ventis_context except ImportError: from redis_client import RedisClient import ventis_context diff --git a/ventis/controller/cloud_provider_logic/EC2/_runtime.py b/ventis/controller/cloud_provider_logic/EC2/_runtime.py index 7e4e9ab..a1cbb4d 100644 --- a/ventis/controller/cloud_provider_logic/EC2/_runtime.py +++ b/ventis/controller/cloud_provider_logic/EC2/_runtime.py @@ -25,7 +25,7 @@ from ventis.controller.utils.env_file import env_file_args from ventis.controller.utils.redis_utils import _wait_for_redis -from ventis.utils.redis_client import RedisClient +from ventis.controller.utils.redis_client import RedisClient logger = logging.getLogger(__name__) diff --git a/ventis/controller/cloud_provider_logic/Local/_runtime.py b/ventis/controller/cloud_provider_logic/Local/_runtime.py index a387f7b..1a84e56 100644 --- a/ventis/controller/cloud_provider_logic/Local/_runtime.py +++ b/ventis/controller/cloud_provider_logic/Local/_runtime.py @@ -15,6 +15,7 @@ DEFAULT_HOST = "localhost" CONTAINER_PORT = 50051 PROVIDER = "local" +MAX_PORT_ATTEMPTS = 50 _controller = None @@ -62,9 +63,6 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id): redis_host = provisioned["redis_host"] runtime_id = provisioned["runtime_id"] - endpoint = routing_endpoint_for(provisioned) - _require_controller().redis.set(f"controller:{endpoint}:agent_id", agent_id) - inspect = _require_controller()._run_cmd( ["docker", "inspect", "-f", "{{.State.Running}}", runtime_id], host, user ) @@ -76,53 +74,72 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id): ) _require_controller()._run_cmd(["docker", "rm", "-f", runtime_id], host, user) - cmd = [ - "docker", - "run", - "-d", - "-it", - "--add-host=host.docker.internal:host-gateway", - "--name", - runtime_id, - "-p", - f"{host_port}:{CONTAINER_PORT}", - "-e", - f"VENTIS_AGENT_PORT={host_port}", - "-e", - f"VENTIS_AGENT_HOST={redis_host}", - "-e", - f"VENTIS_REDIS_HOST={redis_host}", - "-e", - f"VENTIS_REDIS_PORT={spec.get('redis_port', 6379)}", - "-e", - f"VENTIS_POLL_INTERVAL={_require_controller().config.get('poll_interval', 5)}", - ] - if ctrl_type == "workflow": - cmd.extend(["-p", f"{spec.get('api_port', 8080)}:8080"]) - config = _require_controller().config - db_url = config.get("database", {}).get("url") - project_id = config.get("project_id") - if db_url: - cmd.extend(["-e", f"VENTIS_DATABASE_URL={db_url}"]) - if project_id: - cmd.extend(["-e", f"VENTIS_PROJECT_ID={project_id}"]) - if resources.get("cpu"): - cmd.extend(["--cpus", str(resources["cpu"])]) - if resources.get("memory"): - cmd.extend(["--memory", f"{resources['memory']}m"]) - if resources.get("gpu"): - cmd.extend(["--gpus", str(resources["gpu"])]) - - # User secrets from `env_file`. Explicit -e flags above still win, so a - # stray VENTIS_* line in someone's .env cannot break agent wiring. - with env_file_args( - _require_controller(), host, user, runtime_id, _is_local_host(host) - ) as env_args: - cmd.extend(env_args) - cmd.append(image) - result = _require_controller()._run_cmd(cmd, host, user) - if result.returncode != 0: - raise RuntimeError(f"Failed to launch {runtime_id}") + for attempt in range(MAX_PORT_ATTEMPTS): + cmd = [ + "docker", + "run", + "-d", + "-it", + "--add-host=host.docker.internal:host-gateway", + "--name", + runtime_id, + "-p", + f"{host_port}:{CONTAINER_PORT}", + "-e", + f"VENTIS_AGENT_PORT={host_port}", + "-e", + f"VENTIS_AGENT_HOST={redis_host}", + "-e", + f"VENTIS_REDIS_HOST={redis_host}", + "-e", + f"VENTIS_REDIS_PORT={spec.get('redis_port', 6379)}", + "-e", + f"VENTIS_POLL_INTERVAL={_require_controller().config.get('poll_interval', 5)}", + ] + if ctrl_type == "workflow": + cmd.extend(["-p", f"{spec.get('api_port', 8080)}:8080"]) + config = _require_controller().config + db_url = config.get("database", {}).get("url") + project_id = config.get("project_id") + if db_url: + cmd.extend(["-e", f"VENTIS_DATABASE_URL={db_url}"]) + if project_id: + cmd.extend(["-e", f"VENTIS_PROJECT_ID={project_id}"]) + if resources.get("cpu"): + cmd.extend(["--cpus", str(resources["cpu"])]) + if resources.get("memory"): + cmd.extend(["--memory", f"{resources['memory']}m"]) + if resources.get("gpu"): + cmd.extend(["--gpus", str(resources["gpu"])]) + + # User secrets from `env_file`. Explicit -e flags above still win, so a + # stray VENTIS_* line in someone's .env cannot break agent wiring. + with env_file_args( + _require_controller(), host, user, runtime_id, _is_local_host(host) + ) as env_args: + cmd.extend(env_args) + cmd.append(image) + result = _require_controller()._run_cmd(cmd, host, user) + + if result.returncode == 0: + break + if "port is already allocated" in (result.stderr or ""): + # `docker run` leaves a `Created`-but-never-started container behind + # under this name when the port bind fails. Remove it before + # retrying with a new port, or the retry hits a name conflict + # instead of the port conflict we're trying to work around. + _require_controller()._run_cmd(["docker", "rm", "-f", runtime_id], host, user) + host_port += 1 + continue + raise RuntimeError(f"Failed to launch {runtime_id}: {result.stderr}") + else: + raise RuntimeError( + f"Failed to launch {runtime_id}: no free port found after " + f"{MAX_PORT_ATTEMPTS} attempts" + ) + + endpoint = f"{_container_routing_host(host)}:{host_port}" + _require_controller().redis.set(f"controller:{endpoint}:agent_id", agent_id) instance = { "agent_name": agent_name, diff --git a/ventis/deploy.py b/ventis/controller/deploy.py similarity index 98% rename from ventis/deploy.py rename to ventis/controller/deploy.py index d197342..47de634 100644 --- a/ventis/deploy.py +++ b/ventis/controller/deploy.py @@ -17,7 +17,7 @@ def my_workflow(query: str): """ try: - import ventis.ventis_context as ventis_context + import ventis.controller.ventis_context as ventis_context except ImportError: import ventis_context import json @@ -33,7 +33,7 @@ def my_workflow(query: str): # Try to import from absolute package (local install) or fallback to flat file (Docker container) try: - from ventis.utils.redis_client import RedisClient + from ventis.controller.utils.redis_client import RedisClient except ImportError: from redis_client import RedisClient diff --git a/ventis/future.py b/ventis/controller/future.py similarity index 94% rename from ventis/future.py rename to ventis/controller/future.py index 68615f1..04b050a 100644 --- a/ventis/future.py +++ b/ventis/controller/future.py @@ -1,6 +1,6 @@ import time import json -import uuid +import secrets import sys import os import logging @@ -8,12 +8,12 @@ import grpc try: - import ventis.ventis_context as ventis_context + import ventis.controller.ventis_context as ventis_context except ImportError: import ventis_context try: - from ventis.utils.grpc_options import GRPC_CHANNEL_OPTIONS + from ventis.controller.utils.grpc_options import GRPC_CHANNEL_OPTIONS except ImportError: from grpc_options import GRPC_CHANNEL_OPTIONS @@ -23,7 +23,7 @@ sys.path.insert(0, os.path.abspath("grpc_stubs")) try: - from ventis.utils.redis_client import RedisClient + from ventis.controller.utils.redis_client import RedisClient except ImportError: from redis_client import RedisClient import local_controler_pb2 @@ -63,8 +63,11 @@ def __init__(self, parent, service, method, args=None): args: arguments to be passed to the method """ - # initial value of future object - self.id = uuid.uuid4().hex + # initial value of future object. 64-bit (8 bytes / 16 hex chars) -- + # this doubles as the OTel span_id (convert.py), which is defined as + # 64-bit, so it's generated at that width directly instead of a + # 128-bit uuid4 that would need truncating later. + self.id = secrets.token_hex(8) # Grab the request_id from the thread-local context (set by deploy) self.request_id = ventis_context.get_request_id() diff --git a/ventis/controller/global_controller.py b/ventis/controller/global_controller.py index fc5dd38..c85dcbf 100644 --- a/ventis/controller/global_controller.py +++ b/ventis/controller/global_controller.py @@ -13,6 +13,7 @@ import sys import threading import time +import uuid from concurrent.futures import ThreadPoolExecutor import yaml @@ -28,11 +29,14 @@ send_runtime_information, send_agent_information, ) -from ventis.utils.redis_client import RedisClient -from ventis.utils.grpc_options import GRPC_CHANNEL_OPTIONS - -# Add generated grpc_stubs from the local project to the path -sys.path.insert(0, os.path.abspath("grpc_stubs")) +from ventis.controller.utils.redis_client import RedisClient +from ventis.controller.utils.grpc_options import GRPC_CHANNEL_OPTIONS + +# Add generated grpc_stubs from the local project to the path. Projects using +# the .car artifact layout keep grpc_stubs under .car/; older/plain layouts +# keep it at the project root. +_artifact_prefix = ".car" if os.path.isdir(".car") else "" +sys.path.insert(0, os.path.abspath(os.path.join(_artifact_prefix, "grpc_stubs"))) import local_controler_pb2 import local_controler_pb2_grpc import grpc @@ -92,7 +96,7 @@ def __init__(self, config_path): self._last_metrics_poll_time = {} # (host, port) -> time.time() of last metrics read self._lc_stubs = {} # endpoint -> gRPC stub self.instance_manager = InstanceManager(self) - assign_project_id(self.config.get("project_id",0)) + assign_project_id(self.config.get("project_id")) # Clean up any stale containers from previous runs self._cleanup_stale_containers() @@ -188,12 +192,27 @@ def _cleanup_stale_containers(self): def _load_config(config_path): """Load the YAML config file after importing root .env values.""" project_root = os.path.abspath(os.path.join(os.path.dirname(config_path), "..")) + # Under the .car layout, config lives at /.car/config, so the + # naive parent-of-parent lands on .car itself -- go up one more level + # to reach the actual project root where .env lives. + if os.path.basename(project_root) == ".car": + project_root = os.path.dirname(project_root) GlobalController._load_dotenv(os.path.join(project_root, ".env")) with open(config_path, "r") as f: config = yaml.safe_load(f) + if not config.get("project_id"): + config["project_id"] = GlobalController._assign_new_project_id(config_path) config = GlobalController._expand_env_value(config) return config + @staticmethod + def _assign_new_project_id(config_path): + """Generate a project_id and append it to the config file so it stays stable across reloads/restarts.""" + project_id = str(uuid.uuid4()) + with open(config_path, "a") as f: + f.write(f'project_id: "{project_id}"\n') + return project_id + @staticmethod def _load_dotenv(path): """Load simple KEY=VALUE entries without overriding existing environment values.""" @@ -261,7 +280,7 @@ def reload_config(self): self.env_file_path = resolve_env_file(self.config) self.controllers = self.config.get("agents", []) self.poll_interval = self.config.get("poll_interval", 5) - assign_project_id(self.config.get("project_id", 0)) + assign_project_id(self.config.get("project_id")) self._write_identity() self.instance_manager.publish_routing_snapshot(self.controllers) @@ -324,7 +343,7 @@ def _load_and_write_policies(self): def _write_identity(self): """Publish the current project/database identity to every node's Redis.""" payload = { - "project_id": str(self.config.get("project_id", 0)), + "project_id": str(self.config.get("project_id")), "database_url": self.config.get("database", {}).get("url") or "", } targets = list(self.node_redis.values()) or [self.redis] @@ -377,8 +396,11 @@ def _launch_redis_containers(self): redis_port = node_cfg["redis_port"] user = node_cfg["user"] container_name = f"ventis-redis-{host.replace('.', '-')}" - # For localhost, connect directly; for remote, connect via host IP - connect_host = "localhost" if host in ("localhost", "127.0.0.1") else host + # VENTIS_REDIS_HOST overrides the localhost case for a containerized GC; host/remote paths unchanged. + if host in ("localhost", "127.0.0.1"): + connect_host = os.environ.get("VENTIS_REDIS_HOST", "localhost") + else: + connect_host = host if self._redis_container_healthy(container_name, host, user, connect_host, redis_port): logger.info("Reusing existing Redis container %s on %s", container_name, host) @@ -562,7 +584,7 @@ def _poll_one_instance(self, instance): try: future_rows = pull_runtime_information(node_redis) self._otel_db.write_waiting_rows( - future_rows, node_redis, self.config.get("project_id", 0) + future_rows, node_redis, self.config.get("project_id") ) # This is now legacy, keeping it for now, but will remove this later send_runtime_information( @@ -893,9 +915,9 @@ def stop(self): if __name__ == "__main__": - script_dir = os.path.dirname(os.path.abspath(__file__)) - project_root = os.path.join(script_dir, "..", "..") - default_config = os.path.join(project_root, "config", "global_controller.yaml") + default_config = os.path.join( + _artifact_prefix, "config", "global_controller.yaml" + ) import argparse @@ -904,7 +926,7 @@ def stop(self): "-c", "--config", default=default_config, - help="Path to the YAML config file (default: config/global_controller.yaml)", + help=f"Path to the YAML config file (default: {default_config})", ) args = parser.parse_args() diff --git a/ventis/controller/local_controller.py b/ventis/controller/local_controller.py index 8b5942e..8d9b525 100644 --- a/ventis/controller/local_controller.py +++ b/ventis/controller/local_controller.py @@ -18,8 +18,8 @@ try: from ventis.controller.local_controller_frontend import start_server from ventis.controller.utils.gpu_metrics import read_gpu_percent - from ventis.utils.redis_client import RedisClient - from ventis.utils.grpc_options import GRPC_CHANNEL_OPTIONS + from ventis.controller.utils.redis_client import RedisClient + from ventis.controller.utils.grpc_options import GRPC_CHANNEL_OPTIONS except ImportError: from gpu_metrics import read_gpu_percent from local_controller_frontend import start_server @@ -32,7 +32,7 @@ sys.path.insert(0, os.path.abspath("grpc_stubs")) try: - import ventis.ventis_context as ventis_context + import ventis.controller.ventis_context as ventis_context except ImportError: import ventis_context import local_controler_pb2 diff --git a/ventis/controller/local_controller_frontend.py b/ventis/controller/local_controller_frontend.py index a5fcc25..722bd16 100644 --- a/ventis/controller/local_controller_frontend.py +++ b/ventis/controller/local_controller_frontend.py @@ -32,7 +32,7 @@ def __init__(self, my_endpoint="unknown"): redis_host = os.environ.get("VENTIS_REDIS_HOST", "localhost") redis_port = int(os.environ.get("VENTIS_REDIS_PORT", 6379)) try: - from ventis.utils.redis_client import RedisClient + from ventis.controller.utils.redis_client import RedisClient except ImportError: from redis_client import RedisClient self.redis = RedisClient(host=redis_host, port=redis_port) @@ -152,7 +152,7 @@ def _cleanup_request(self, request_id): def start_server(port=50051, my_endpoint="unknown"): """Start the gRPC server.""" try: - from ventis.utils.grpc_options import GRPC_SERVER_OPTIONS + from ventis.controller.utils.grpc_options import GRPC_SERVER_OPTIONS except ImportError: from grpc_options import GRPC_SERVER_OPTIONS diff --git a/ventis/utils/grpc_options.py b/ventis/controller/utils/grpc_options.py similarity index 100% rename from ventis/utils/grpc_options.py rename to ventis/controller/utils/grpc_options.py diff --git a/ventis/utils/redis_client.py b/ventis/controller/utils/redis_client.py similarity index 100% rename from ventis/utils/redis_client.py rename to ventis/controller/utils/redis_client.py diff --git a/ventis/controller/utils/telemetry_logging.py b/ventis/controller/utils/telemetry_logging.py index 7d911e2..503f3e1 100644 --- a/ventis/controller/utils/telemetry_logging.py +++ b/ventis/controller/utils/telemetry_logging.py @@ -7,7 +7,7 @@ from sqlalchemy import create_engine, text from ventis.controller.utils import pricing -from ventis.utils.redis_client import RedisClient +from ventis.controller.utils.redis_client import RedisClient logger = logging.getLogger(__name__) diff --git a/ventis/ventis_context.py b/ventis/controller/ventis_context.py similarity index 100% rename from ventis/ventis_context.py rename to ventis/controller/ventis_context.py diff --git a/ventis/llm/__init__.py b/ventis/llm/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/ventis/server.py b/ventis/server.py new file mode 100644 index 0000000..8df8b0f --- /dev/null +++ b/ventis/server.py @@ -0,0 +1,70 @@ +import os +import signal +import subprocess +import sys + +from flask import Flask, jsonify, request + +app = Flask("ventis-server") + +# The project files are copied here (into a named volume) by `canyonos sync` / +# `canyonos deploy`. Deploy builds and launches against this path. +WORKSPACE_DIR = "/workspace" + +_gc_process = None + + +def _gc_running(): + return _gc_process is not None and _gc_process.poll() is None + + +@app.route("/new-project", methods=["POST"]) +def new_project(): + return jsonify({"error": "new-project runs locally via the CLI"}), 400 + + +@app.route("/deploy", methods=["POST"]) +def deploy(): + global _gc_process + + if _gc_running(): + return jsonify({"error": "already running"}), 409 + + data = request.get_json(force=True, silent=True) or {} + config_path = data.get("config_path", "config/global_controller.yaml") + full_path = os.path.join(WORKSPACE_DIR, config_path) + + if not os.path.isfile(full_path): + return jsonify({"error": f"config file not found: {full_path}"}), 400 + + # `ventis deploy` builds (stubs/protos/images) then launches the Global + # Controller. cwd is the workspace so build outputs land alongside the + # project files and the controller finds them. Build+deploy output streams + # to the container logs, which `canyonos deploy` tails. + _gc_process = subprocess.Popen( + [sys.executable, "-m", "ventis.cli", "deploy", "-c", config_path], + cwd=WORKSPACE_DIR, + ) + return jsonify({"status": "started", "pid": _gc_process.pid}), 200 + + +@app.route("/clean", methods=["POST"]) +def clean(): + global _gc_process + + if not _gc_running(): + return jsonify({"error": "not running"}), 409 + + _gc_process.send_signal(signal.SIGTERM) + _gc_process.wait() + _gc_process = None + return jsonify({"status": "stopped"}), 200 + + +@app.route("/status", methods=["GET"]) +def status(): + return jsonify({"running": _gc_running()}), 200 + + +if __name__ == "__main__": + app.run(host="0.0.0.0", port=8000) diff --git a/ventis/stub_generator.py b/ventis/stub_generator.py index 33824ff..1108be1 100644 --- a/ventis/stub_generator.py +++ b/ventis/stub_generator.py @@ -378,8 +378,8 @@ def generate_docker( # Copy general agent files files_to_copy += [ # (source_path, destination_filename) - (os.path.join(script_dir, "future.py"), "future.py"), - (os.path.join(script_dir, "ventis_context.py"), "ventis_context.py"), + (os.path.join(script_dir, "controller", "future.py"), "future.py"), + (os.path.join(script_dir, "controller", "ventis_context.py"), "ventis_context.py"), ( os.path.join(script_dir, "controller", "local_controller.py"), "local_controller.py", @@ -388,26 +388,20 @@ def generate_docker( os.path.join(script_dir, "controller", "local_controller_frontend.py"), "local_controller_frontend.py", ), - (os.path.join(script_dir, "utils", "redis_client.py"), "redis_client.py"), - (os.path.join(script_dir, "utils", "grpc_options.py"), "grpc_options.py"), + (os.path.join(script_dir, "controller", "utils", "redis_client.py"), "redis_client.py"), + (os.path.join(script_dir, "controller", "utils", "grpc_options.py"), "grpc_options.py"), ( os.path.join(script_dir, "controller", "utils", "gpu_metrics.py"), "gpu_metrics.py", ), - (os.path.join(script_dir, "llm", "bedrock.py"), "bedrock.py"), + (os.path.join(script_dir, "controller", "bedrock.py"), "bedrock.py"), ] # Copy provided agent stubs, overwriting the swept real file at the same path if stub_files: for stub_file in stub_files: - files_to_copy.append( - ( - os.path.abspath(stub_file), - _stub_destination(stub_file, stub_entrypoints or {}), - ) - ) - - files_to_copy.append((os.path.abspath(agent_file), os.path.basename(agent_file))) + destination = _stub_destination(stub_file, stub_entrypoints or {}) + files_to_copy.append((os.path.abspath(stub_file), destination)) # Copy gRPC generated stubs if they exist if os.path.isdir(grpc_stubs_dir): @@ -417,6 +411,12 @@ def generate_docker( _copy_files(output_dir, files_to_copy) + # Copy the real agent entrypoint to the context root. + shutil.copy2( + os.path.abspath(agent_file), + os.path.join(output_dir, os.path.basename(agent_file)), + ) + # Copy the YAML definition too shutil.copy2( os.path.abspath(yaml_path), @@ -500,10 +500,9 @@ def generate_workflow_docker( files_to_copy = _sweep_py_files(project_dir) if project_dir else [] files_to_copy += [ - (os.path.abspath(workflow_file), workflow_basename), - (os.path.join(script_dir, "future.py"), "future.py"), - (os.path.join(script_dir, "ventis_context.py"), "ventis_context.py"), - (os.path.join(script_dir, "deploy.py"), "deploy.py"), + (os.path.join(script_dir, "controller", "future.py"), "future.py"), + (os.path.join(script_dir, "controller", "ventis_context.py"), "ventis_context.py"), + (os.path.join(script_dir, "controller", "deploy.py"), "deploy.py"), ( os.path.join(script_dir, "controller", "local_controller.py"), "local_controller.py", @@ -512,8 +511,8 @@ def generate_workflow_docker( os.path.join(script_dir, "controller", "local_controller_frontend.py"), "local_controller_frontend.py", ), - (os.path.join(script_dir, "utils", "redis_client.py"), "redis_client.py"), - (os.path.join(script_dir, "utils", "grpc_options.py"), "grpc_options.py"), + (os.path.join(script_dir, "controller", "utils", "redis_client.py"), "redis_client.py"), + (os.path.join(script_dir, "controller", "utils", "grpc_options.py"), "grpc_options.py"), *[ (os.path.join(script_dir, "controller", "utils", name), name) for name in ("gpu_metrics.py", "session_logging.py") @@ -522,12 +521,8 @@ def generate_workflow_docker( # Copy stub files, overwriting the swept real file at the same path for stub_file in stub_files: - files_to_copy.append( - ( - os.path.abspath(stub_file), - _stub_destination(stub_file, stub_entrypoints or {}), - ) - ) + destination = _stub_destination(stub_file, stub_entrypoints or {}) + files_to_copy.append((os.path.abspath(stub_file), destination)) # Copy gRPC generated stubs if they exist if os.path.isdir(grpc_stubs_dir): @@ -537,6 +532,12 @@ def generate_workflow_docker( _copy_files(output_dir, files_to_copy) + # Copy the real workflow entrypoint to the context root. + shutil.copy2( + os.path.abspath(workflow_file), + os.path.join(output_dir, workflow_basename), + ) + # ---- workflow_launcher.py -------------------------------------------- launcher = f"""import threading import time diff --git a/ventis/utils/__init__.py b/ventis/utils/__init__.py deleted file mode 100644 index 7863cb0..0000000 --- a/ventis/utils/__init__.py +++ /dev/null @@ -1 +0,0 @@ -# empty proxy module for packaging