From b9d82b7da95a544a9c41ff3033698c2c6d944c0d Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Thu, 10 Sep 2026 11:38:06 -0700 Subject: [PATCH 01/10] Linear Problem 1 fix Portfolio workflow indexed a Future instead of blocking on its value. Co-Authored-By: Claude Sonnet 5 --- examples/portfolio/workflow/portfolio_workflow.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/examples/portfolio/workflow/portfolio_workflow.py b/examples/portfolio/workflow/portfolio_workflow.py index 772f935..a55139d 100644 --- a/examples/portfolio/workflow/portfolio_workflow.py +++ b/examples/portfolio/workflow/portfolio_workflow.py @@ -46,7 +46,9 @@ def main( advisor = AdvisorAgent() # Stage 0: parse the free-text request into structured holdings + window. - intent = intent_agent.parse(query=query) + # parse() returns a Future in deployment -- .value() blocks for the result, + # which comes back as a JSON string like the other stage calls below. + intent = json.loads(intent_agent.parse(query=query).value()) holdings = intent["holdings"] lookback_days = intent["lookback_days"] From 9f86d75ca10482fb265b5034400b5d6ffa4c6f31 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Thu, 10 Sep 2026 11:57:16 -0700 Subject: [PATCH 02/10] Linear Problem 5 fix Place generated agent stubs both flat and at their entrypoint-mirrored path in agent and workflow Docker contexts, so both 'from price_agent import PriceAgent' and 'from agents.price_agent import PriceAgent' style peer imports resolve. _stub_destination() only ever wrote one destination, contradicting the two comments in cli.py that already claimed dual placement -- this is why MetricsAgent's container logged 'No module named price_agent'. Updated the one test that asserted the old (wrong) single-destination behavior. This same class of bug and fix (ac9a75e, 01a70f2, 137e1db, 0b9546c) has recurred across several unmerged branches; none reached main. Co-Authored-By: Claude Sonnet 5 --- canyonos_core/stub_generator.py | 22 ++++++++++++++++------ tests/test_stub_generator.py | 4 ++-- 2 files changed, 18 insertions(+), 8 deletions(-) diff --git a/canyonos_core/stub_generator.py b/canyonos_core/stub_generator.py index a830735..ef3f4dc 100644 --- a/canyonos_core/stub_generator.py +++ b/canyonos_core/stub_generator.py @@ -411,11 +411,16 @@ def generate_docker( ), ] - # Copy provided agent stubs, overwriting the swept real file at the same path + # Copy provided agent stubs both flat (for `from price_agent import ...` style + # peer imports) and at their entrypoint-mirrored path (overwriting the swept + # real agent file there, as before), so both import styles resolve. if stub_files: for stub_file in stub_files: - destination = _stub_destination(stub_file, stub_entrypoints or {}) - files_to_copy.append((os.path.abspath(stub_file), destination)) + flat_dest = os.path.basename(stub_file) + entrypoint_dest = _stub_destination(stub_file, stub_entrypoints or {}) + files_to_copy.append((os.path.abspath(stub_file), flat_dest)) + if entrypoint_dest != flat_dest: + files_to_copy.append((os.path.abspath(stub_file), entrypoint_dest)) # Copy gRPC generated stubs if they exist if os.path.isdir(grpc_stubs_dir): @@ -540,10 +545,15 @@ def generate_workflow_docker( ], ] - # Copy stub files, overwriting the swept real file at the same path + # Copy stub files both flat (for `from price_agent import ...` style imports + # in the workflow) and at their entrypoint-mirrored path (overwriting the + # swept real agent file there, as before), so both import styles resolve. for stub_file in stub_files: - destination = _stub_destination(stub_file, stub_entrypoints or {}) - files_to_copy.append((os.path.abspath(stub_file), destination)) + flat_dest = os.path.basename(stub_file) + entrypoint_dest = _stub_destination(stub_file, stub_entrypoints or {}) + files_to_copy.append((os.path.abspath(stub_file), flat_dest)) + if entrypoint_dest != flat_dest: + files_to_copy.append((os.path.abspath(stub_file), entrypoint_dest)) # Copy gRPC generated stubs if they exist if os.path.isdir(grpc_stubs_dir): diff --git a/tests/test_stub_generator.py b/tests/test_stub_generator.py index d3af7f2..2dd013c 100644 --- a/tests/test_stub_generator.py +++ b/tests/test_stub_generator.py @@ -116,7 +116,7 @@ def test_unsafe_entrypoint_falls_back_to_flat(self): class GenerateWorkflowDockerStubPlacementTests(unittest.TestCase): - def test_stub_lands_only_at_its_entrypoint_path(self): + def test_stub_lands_both_flat_and_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") @@ -136,7 +136,7 @@ def test_stub_lands_only_at_its_entrypoint_path(self): 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") + self.assertIn("class SplitAgent", flat_path.read_text()) if __name__ == "__main__": From 993e889f12dbfc2c694a8f8fc18f95a513a38443 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Thu, 10 Sep 2026 12:02:12 -0700 Subject: [PATCH 03/10] Fixed Issue #10 --- canyonos_core/controller/local_controller.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/canyonos_core/controller/local_controller.py b/canyonos_core/controller/local_controller.py index a3406c2..b4e5adb 100644 --- a/canyonos_core/controller/local_controller.py +++ b/canyonos_core/controller/local_controller.py @@ -467,7 +467,7 @@ def _process_request(self, data): for key, value in args.items(): if ( isinstance(value, str) - and len(value) == 32 + and len(value) == 16 and all(c in "0123456789abcdefABCDEF" for c in value) ): future_key = f"future:{value}" @@ -531,7 +531,7 @@ def _resolve_future_args(self, args, poll_interval=0.01, timeout=300): # Check if this arg value is a UUID hex string identifying a future if ( isinstance(value, str) - and len(value) == 32 + and len(value) == 16 and all(c in "0123456789abcdefABCDEF" for c in value) ): future_key = f"future:{value}" From bb7f053036751a100ccc63a080eb03ea325e48a8 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Thu, 10 Sep 2026 14:12:24 -0700 Subject: [PATCH 04/10] Issue #2 --- canyonos_core/OTLP_Exporter/DESIGN.md | 135 ++-- canyonos_core/OTLP_Exporter/convert.py | 42 +- canyonos_core/OTLP_Exporter/db.py | 78 ++- canyonos_core/OTLP_Exporter/otel_exporter.py | 481 +++++++++++--- tests/test_otel_exporter_fanout.py | 625 ++++++++++++++++--- 5 files changed, 1145 insertions(+), 216 deletions(-) diff --git a/canyonos_core/OTLP_Exporter/DESIGN.md b/canyonos_core/OTLP_Exporter/DESIGN.md index 15f0163..b33e457 100644 --- a/canyonos_core/OTLP_Exporter/DESIGN.md +++ b/canyonos_core/OTLP_Exporter/DESIGN.md @@ -2,10 +2,10 @@ Status: **implemented (single-table design; multi-destination fan-out in progress)**. `GlobalController` writes futures into a `waiting` table (SQLite); a GC-supervised, GC-restarted OTel Exporter process reads -finished/unsent rows, converts each to an OTel span, and hands it to a real -`BatchSpanProcessor`/`OTLPSpanExporter`. Batching, serialization, and sending are all -OTel SDK code — the only custom pieces are the row→span conversion and durable -sent-tracking. This doc is a design/rationale reference; the actual files +finished/unsent rows, converts them to OTel spans, and exports them synchronously via +`OTLPSpanExporter.export()`, marking rows sent only on a `SpanExportResult.SUCCESS` from +every destination. Serialization, transport, and transient-failure retry are all OTel SDK +code — the only custom pieces are the row→span conversion and durable sent-tracking. This doc is a design/rationale reference; the actual files (`otel_exporter.py`, `db.py`, `convert.py`, `canyonos/controller/utils/process_supervisor.py`) are the source of truth for current behavior. @@ -13,7 +13,7 @@ are the source of truth for current behavior. CanyonOS futures need to reach an external OTLP-compatible tracing backend. Design: a separate OTLP Exporter process, spawned and supervised by GlobalController, that reads unsent finished future rows from a local SQLite DB, converts them into OTel spans, and -hands them to the OTel SDK's own batching/export machinery, which ships them to an +exports them through the OTel SDK's own OTLP exporters, which ship them to an external OTLP Receiver (out of scope here — assumed to be a separate, already-addressable service). @@ -31,7 +31,7 @@ Decisions (final status): config-aware. `GlobalController` serializes that list to JSON and passes it to the exporter subprocess as a single `CANYONOS_OTEL_DESTINATIONS` env var via `ProcessSupervisor.register(..., env=...)`. The exporter builds one independent - exporter/`BatchSpanProcessor` pair per destination, picking the gRPC vs HTTP + OTLP exporter per destination, picking the gRPC vs HTTP exporter class from each destination's `protocol` field. gRPC and HTTP destinations may be mixed in the same list. Deliberately vendor-neutral: no backend name (Postgres, Langfuse, or otherwise) appears anywhere in `otel_exporter.py`; the @@ -41,19 +41,23 @@ Decisions (final status): own idiomatic mechanism, so no exporter-side config plumbing was added, only a GC-side YAML→env-var translation. If `otel.destinations` is absent, GlobalController logs that no OTel metrics collection will happen and skips starting the exporter - subprocess entirely. Configuration is read at exporter startup; changing it requires - a GlobalController/exporter restart. + subprocess entirely. Configuration now reaches the exporter through the + `otel:destinations` Redis key rather than the `CANYONOS_OTEL_DESTINATIONS` env var + described above (kept for history): GlobalController writes that key, and every poll + tick re-reads it and rebuilds the exporters if it changed, so a config reload (SIGHUP) + reaches this process without a restart. An invalid value is logged and ignored, + leaving the previous working exporters in place. - **Data source**: NOT `runtime_information` — a dedicated `waiting` table in its own SQLite file (`canyonos/OTLP_Exporter/otel_queue.db`, see `db.py`), written by GC's existing `_poll_controllers` *alongside* (not instead of) the existing `send_runtime_information` write. Keeps this pipeline's schema/state fully decoupled from the dashboard/cost table. - **Two tables collapsed into one**: an earlier version of this design had a second - `queue` table (`waiting` → promote → `queue` → drain → send). Collapsed once it became - clear `BatchSpanProcessor` already provides its own in-memory queue — the only thing a - second table added was durability across the exporter's own process restarts, which a - `sent` column on `waiting` alone provides just as well, with less code. See `db.py`'s - module docstring. + `queue` table (`waiting` → promote → `queue` → drain → send). Collapsed to a single + `sent` column on `waiting`, which provides the same durability with less code. (The + original rationale cited `BatchSpanProcessor`'s in-memory queue; that processor has + since been removed — see "Synchronous export" below — and `waiting` is now the only + queue in the pipeline, which is what makes the durability property hold at all.) - **Span construction**: settled — spans are built as `ReadableSpan` objects directly (bypassing `Tracer`/`TracerProvider` entirely, no `IdGenerator` workaround needed for either `trace_id` or `span_id`). Confirmed working via `ConsoleSpanExporter` during @@ -89,26 +93,78 @@ and raises if invoked directly without `CANYONOS_OTEL_DESTINATIONS` set). `otel_exporter.py` parses the destination configuration at startup and constructs the appropriate OTLP exporter for each entry (gRPC or HTTP), passing that destination's -endpoint and headers to the SDK. `BatchSpanProcessor(..., schedule_delay_millis=1000)` -— the flush delay is explicitly overridden from the SDK default (5000ms) to 1000ms; -`max_export_batch_size` is left at the SDK default (512), which already approximates the -original "500 spans" batching ask without any override needed. +endpoint, headers, and timeout to the SDK. Each destination's `timeout` bounds how long a +single failing export blocks the poll loop, so it is worth setting explicitly rather than +leaning on the SDK default of 10s. -### 2. `canyonos/OTLP_Exporter/otel_exporter.py` +### 2. `canyonos/OTLP_Exporter/otel_exporter.py` — synchronous export A plain loop, polling every `POLL_INTERVAL_SECONDS` (5s, checked every 1s so SIGTERM stays responsive), calling `_send_pending()` each tick. At startup it constructs one -independent OTLP exporter and `BatchSpanProcessor` for each configured destination; -each pair may use a different protocol, endpoint, and headers: -- `SELECT * FROM waiting WHERE finished_at IS NOT NULL AND (sent IS NULL OR sent = 0)`. -- Per row, each isolated in its own try/except (one malformed row is logged and skipped, - never blocks the rest of the batch): `convert.waiting_row_to_span(row)` → - `on_end(span)` on every configured processor → `db.mark_sent(future_id)` immediately. - The row is marked after it has been queued to all processors. `sent` therefore means - **queued to every configured destination**, not remotely acknowledged; this is the - initial best-effort delivery contract and retains the existing single boolean schema. -- Each processor is constructed once at startup; no `TracerProvider` is used at all, - since spans are hand-built and handed straight to the processors via `on_end()`. -- Every processor is shut down on exit, flushing its pending batch independently. +independent OTLP exporter per configured destination; each may use a different protocol, +endpoint, headers, and timeout: +- `SELECT * FROM waiting WHERE finished_at IS NOT NULL AND (sent IS NULL OR sent = 0)`, + bounded by `MAX_SPANS_PER_POLL` so one tick's export request stays a sane size; a + backlog is drained over successive polls. +- Conversion is per row, isolated in its own try/except, and each span is then + test-encoded individually (`_reject_unexportable`). Encoding is what actually rejects + a bad row — an out-of-range id, an unencodable attribute — and it would otherwise + happen inside the batched `export()` call, where one row's failure discards every + other span in the batch and the error names only the destination. Rejecting per row + keeps one bad future from blocking everything behind it and names the offending + `future_id`. +- After `MAX_ROW_EXPORT_ATTEMPTS` (5) such rejections, that row is replaced by a + placeholder span (`convert.invalid_row_placeholder_span`) which exports normally and + marks the row sent, so it leaves the pending set instead of being retried forever and + occupying part of the `LIMIT` window. The placeholder keeps the row's real `trace_id` + where the `session_id` allows it, and masks out-of-range ids into valid ones. It is + named `canyonos.invalid_span`, carries `canyonos.export.invalid` and the real + `future_id` as attributes, and deliberately has no exception event: it records that + telemetry could not be represented, **not** that the agent failed, and a dashboard must + be able to tell those apart. The counter is in-memory only — the placeholder is what + makes the outcome durable, so no schema change and no attempt column are needed. + Only per-row rejections count toward it; an export failure is shared by the whole + batch, and counting those would replace the entire queue with placeholders after a + spell of receiver downtime. +- The resulting spans are exported as one batch per destination via + `exporter.export(spans)`, which is **synchronous** and returns a `SpanExportResult`. + Rows are marked sent only when every destination returned SUCCESS *and* reported no + partial rejection. +- Retry needs no machinery: a failed batch simply leaves those rows at `sent = 0`, and + the next poll picks them up. `waiting` is the durable queue. The SDK already retries + transient failures internally with exponential backoff, bounded by the destination's + `timeout`, so a returned FAILURE means it genuinely gave up. +- Marking is all-or-nothing across destinations, so one destination failing re-delivers + to destinations that already accepted the batch. Span ids are deterministic, so those + duplicates collapse at the backend. +- **`partial_success` is checked, for HTTP destinations only.** A receiver may return + 200/OK while rejecting individual spans; the SDK exporters discard the response body + and report `SUCCESS` regardless, so `sent` would otherwise mean "the receiver accepted + the request", not "every span was stored". HTTP exporters are therefore built with a + `requests.Session` carrying a response hook (`_PartialSuccessRecorder`) that reads + `partial_success.rejected_spans`, and a non-zero count fails the batch. gRPC exposes no + equivalent public seam, so gRPC destinations cannot detect partial rejection — the + exporter logs a warning once when one is configured. +- `protocol` is validated against `SUPPORTED_PROTOCOLS` (`grpc`, `http`, + `http/protobuf`). It was previously read but never checked, so any other value — + including a typo or an absent field — silently selected the HTTP exporter. +- A queue that never yields a row is reported. `sqlite3.connect()` creates a missing + file instead of refusing, so a misdirected `DB_PATH` reads as a permanently idle + queue rather than an error. After `EMPTY_QUEUE_WARNING_POLLS` consecutive empty + polls the exporter checks the table's total row count and, if it is still zero, + warns once naming the path. A queue whose rows are all already exported is a + normal idle state and stays silent. +- No `TracerProvider` and no `BatchSpanProcessor` are used at all; spans are hand-built + and handed straight to the exporters. +- Every exporter is shut down on exit. Nothing is buffered in an OTLP exporter (its own + `force_flush()` is a documented no-op), so there is nothing to lose on shutdown — + anything unacknowledged is still `sent = 0` and resumes on the next start. + +**Why not `BatchSpanProcessor`** (the original design, removed): `on_end()` only enqueues +onto an in-memory queue and returns `None`, so the real send happened later on an SDK +background thread with no way to report back. Rows were marked sent immediately and a +failed export was lost silently and permanently, with the HTTP error stranded in the GC +container log. The in-memory queue also silently dropped spans when full, and lost its +entire contents on SIGKILL — while those rows already read `sent = 1`. ### 3. Future row → OTel span conversion (`canyonos/OTLP_Exporter/convert.py`) `future_id` maps to OTel `span_id`, not `trace_id` — `session_id` (== `request_id`) is @@ -174,23 +230,26 @@ config work, since `protocol: http` now needs that package importable). ## Known gaps (not yet built) - `WriteResult()` passes an undefined `error_message` variable to its fan-out callback, which can interrupt remote consumer propagation after the callback hash is persisted. -- Rows are marked `sent` immediately after `BatchSpanProcessor.on_end()` accepts them, +- ~~Rows are marked `sent` immediately after `BatchSpanProcessor.on_end()` accepts them, before the asynchronous OTLP export is confirmed; a later delivery failure can lose a - span while leaving `sent = 1`. + span while leaving `sent = 1`.~~ **Fixed** — export is now synchronous and `sent` is + written only on a SUCCESS from every destination. - Spans carry no explicit `resource`/`instrumentation_scope` — would show as `service.name=unknown_service` at a real backend. -- Destination-specific delivery acknowledgement/retry state is not tracked yet: - `sent` only records that the span was queued to all configured processors, so an - asynchronous export failure can still lose a span until a later delivery-state design - is added. +- Per-destination delivery state is still not tracked: `sent` is one boolean across all + destinations, so if one of several destinations fails the whole batch is retried + everywhere and the healthy destinations receive duplicates. Harmless for tracing + backends (ids are deterministic), but a per-destination table would avoid it. - `waiting` grows unboundedly: sent rows are never pruned, and futures that never finish (`finished_at` never arrives) also stay forever, invisible and un-expiring. - `error_name` is always `NULL` — CanyonOS's own Redis writer never records a distinct exception-type field, only a message string. - Test coverage is still limited; the waiting-field migration/normalization/conversion path is covered, but the exporter process and live OTLP delivery are not. -- Never verified against a live OTLP receiver — only against a refused connection - (confirmed the SDK's real retry/error-handling path is exercised correctly). +- Live-receiver coverage is now basic but real: the synchronous-export change was + verified against a local HTTP receiver (span accepted, real OTLP protobuf received, + row marked sent) as well as a refused connection (row left unsent and retried). Still + never verified against a production-grade OTLP backend. - No retry-limit/quarantine for a permanently malformed row — it logs an error every poll forever rather than being given up on. diff --git a/canyonos_core/OTLP_Exporter/convert.py b/canyonos_core/OTLP_Exporter/convert.py index fc749c8..c0c97c2 100644 --- a/canyonos_core/OTLP_Exporter/convert.py +++ b/canyonos_core/OTLP_Exporter/convert.py @@ -50,13 +50,15 @@ def waiting_row_to_span(row): events = [] status = Status(StatusCode.UNSET) if row["failed"]: + # An absent error_name means the producer never recorded one; naming a + # type here would invent an exception class the code never raised. + exception_attributes = {EXCEPTION_MESSAGE: row.get("error_message") or ""} + if row.get("error_name"): + exception_attributes[EXCEPTION_TYPE] = row["error_name"] events.append( Event( name="exception", - attributes={ - EXCEPTION_TYPE: row.get("error_name") or "RuntimeError", - EXCEPTION_MESSAGE: row.get("error_message") or "", - }, + attributes=exception_attributes, timestamp=to_epoch_nanos(row.get("finished_at")), ) ) @@ -103,3 +105,35 @@ def waiting_row_to_span(row): start_time=to_epoch_nanos(row.get("started_at")), end_time=to_epoch_nanos(row.get("finished_at")), ) + + +def invalid_row_placeholder_span(row, reason): + """Stand-in span for a future whose own telemetry cannot be encoded. + + Deliberately not named after the agent and not carrying an exception event: + this records that telemetry could not be represented, not that the agent + failed, and the two must stay distinguishable in a dashboard. + """ + row = dict(row) + context = SpanContext( + trace_id=(int(row["session_id"], 16) & (2**128 - 1)) or 1, + span_id=(int(row["future_id"], 16) & (2**64 - 1)) or 1, + is_remote=False, + trace_flags=_SAMPLED, + ) + return ReadableSpan( + name="canyonos.invalid_span", + context=context, + parent=None, + attributes={ + "canyonos.export.invalid": True, + "canyonos.export.error": reason, + "canyonos.future_id": row["future_id"], + }, + status=Status( + StatusCode.ERROR, description=f"span could not be exported: {reason}" + ), + kind=SpanKind.INTERNAL, + start_time=to_epoch_nanos(row.get("started_at")), + end_time=to_epoch_nanos(row.get("finished_at")), + ) diff --git a/canyonos_core/OTLP_Exporter/db.py b/canyonos_core/OTLP_Exporter/db.py index 16c02ae..c612190 100644 --- a/canyonos_core/OTLP_Exporter/db.py +++ b/canyonos_core/OTLP_Exporter/db.py @@ -1,14 +1,7 @@ -"""SQLite schema and writes for the OTel export pipeline's waiting table. - -`waiting` holds future rows as GlobalController observes them (including still-running -ones). There's no separate queue table -- OTel's own BatchSpanProcessor already queues -and batches spans in memory, so the only thing we need to track durably is which rows -have already been sent, which the `sent` column on this same table provides. (An earlier -version of this pipeline had a second `queue` table for that; collapsed away since it -wasn't doing anything BatchSpanProcessor doesn't already do -- see DESIGN.md.) -""" +"""SQLite schema and writes for the OTel export pipeline's waiting table.""" import json +import logging import os import sqlite3 @@ -17,8 +10,28 @@ # It is currently stored here for backcompat with the old telemetry collecting +logger = logging.getLogger(__name__) + DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "otel_queue.db") +# Cost lookups can fail on every row of every poll, so each kind is reported once. +_cost_failures_logged = set() + + +def _log_cost_failure(kind, exc): + """Report the first failure of each cost-lookup kind; suppress the rest.""" + if kind in _cost_failures_logged: + return + _cost_failures_logged.add(kind) + logger.warning( + "%s lookup failed; affected rows are recorded with a cost of 0. Further " + "%s failures are suppressed for the life of this process: %s", + kind, + kind, + exc, + exc_info=True, + ) + # Demo-only multipliers for scaling displayed costs, DELETE FOR MORE ACCURATE METRICS _TOKEN_COST_MULTIPLIER = 10000 _SERVER_COST_MULTIPLIER = 100000 @@ -88,7 +101,7 @@ def init_db(db_path=DB_PATH): def _normalize_json_text(value): """Return JSON text, encoding legacy scalar strings that are not valid JSON.""" - if value is None or value == "": + if value is None: return None try: json.loads(value) @@ -111,6 +124,18 @@ def write_waiting_rows(rows, redis_client=None, project_id=None, db_path=DB_PATH fid = raw.get("future_id") session_id = raw.get("request_id") if not fid or not session_id: + missing = ", ".join( + field + for field, present in (("future_id", fid), ("request_id", session_id)) + if not present + ) + logger.warning( + "Dropping future row missing %s; it will never be exported " + "(future_id=%r, request_id=%r)", + missing, + fid, + session_id, + ) continue agent_id = raw.get("agent") started_at = float(raw.get("created_at") or 0) @@ -144,7 +169,8 @@ def write_waiting_rows(rows, redis_client=None, project_id=None, db_path=DB_PATH ) * _TOKEN_COST_MULTIPLIER ) - except Exception: + except Exception as e: + _log_cost_failure("Token cost", e) token_cost = 0.0 try: server_cost = ( @@ -156,7 +182,8 @@ def write_waiting_rows(rows, redis_client=None, project_id=None, db_path=DB_PATH ) * _SERVER_COST_MULTIPLIER ) - except Exception: + except Exception as e: + _log_cost_failure("Server cost", e) server_cost = 0.0 else: token_cost = 0.0 @@ -211,3 +238,30 @@ def mark_sent(future_id, db_path=DB_PATH): conn.commit() finally: conn.close() + + +def mark_sent_many(future_ids, db_path=DB_PATH): + """Mark every listed waiting row sent in one transaction.""" + if not future_ids: + return + try: + conn = sqlite3.connect(db_path) + except Exception as e: + # Re-raised, not swallowed: the caller reports these rows as delivered + # but unmarked, which is what tells an operator to expect duplicates. + logger.error( + "Failed to open %s to mark %d row(s) sent: %s", + db_path, + len(future_ids), + e, + exc_info=True, + ) + raise + try: + conn.executemany( + "UPDATE waiting SET sent = 1 WHERE future_id = ?", + [(future_id,) for future_id in future_ids], + ) + conn.commit() + finally: + conn.close() diff --git a/canyonos_core/OTLP_Exporter/otel_exporter.py b/canyonos_core/OTLP_Exporter/otel_exporter.py index 2d1cab1..2903cbc 100644 --- a/canyonos_core/OTLP_Exporter/otel_exporter.py +++ b/canyonos_core/OTLP_Exporter/otel_exporter.py @@ -1,14 +1,4 @@ -"""Entrypoint for the OTLP exporter process. - -Each poll tick reads finished, not-yet-sent rows from ``waiting``, converts each to a -span, hands it to every configured BatchSpanProcessor, and marks it sent only after -all processors accept it. Batching, OTLP serialization, and sending remain the SDK's -responsibility (see DESIGN.md). - -Destinations come from the ``otel:destinations`` Redis key (GlobalController writes it), -not env -- every poll tick re-reads it and rebuilds processors if it changed, so a config -reload (SIGHUP) reaches this process without a restart. -""" +"""Entrypoint for the OTLP exporter process.""" import json import logging @@ -28,7 +18,12 @@ from opentelemetry.exporter.otlp.proto.http.trace_exporter import ( OTLPSpanExporter as HttpOTLPSpanExporter, ) -from opentelemetry.sdk.trace.export import BatchSpanProcessor +from opentelemetry.exporter.otlp.proto.common.trace_encoder import encode_spans +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( + ExportTraceServiceResponse, +) +from opentelemetry.sdk.trace.export import SpanExportResult +import requests import convert import db @@ -37,9 +32,25 @@ logger = logging.getLogger(__name__) _running = True -_processors = [] +_exporters = [] _last_destinations_raw = None POLL_INTERVAL_SECONDS = 5 +# Bounds one export request, since a backlog is drained by repeated polls. +MAX_SPANS_PER_POLL = 512 +MAX_ROW_EXPORT_ATTEMPTS = 5 +# Counted in memory only: a placeholder marks the row sent, so it leaves the +# pending set for good and the count never needs to survive a restart. +_row_export_failures = {} +EMPTY_QUEUE_WARNING_POLLS = 12 +EMPTY_QUEUE_REWARN_POLLS = 720 +SUPPORTED_PROTOCOLS = ("grpc", "http", "http/protobuf") +_consecutive_empty_polls = 0 +_empty_queue_warned_at = None +# Keyed by destination name; only HTTP destinations can report partial success. +_partial_success_recorders = {} +_grpc_partial_success_warned = False +# Destination name -> whether its last export succeeded, so recovery is reported. +_destination_healthy = {} DESTINATIONS_KEY = "otel:destinations" # keep in sync with GlobalController.OTEL_DESTINATIONS_KEY _redis = None @@ -52,7 +63,13 @@ def _validate_destination(destination, index): if not isinstance(name, str) or not name.strip(): raise ValueError(f"destination {index} name must be a non-empty string") - protocol = destination.get("protocol") # must be exactly "grpc" or "http" + protocol = destination.get("protocol") + protocol = protocol.lower() if isinstance(protocol, str) else protocol + if protocol not in SUPPORTED_PROTOCOLS: + raise ValueError( + f"destination {name!r} protocol must be one of " + f"{list(SUPPORTED_PROTOCOLS)}; got {protocol!r}" + ) endpoint = destination.get("endpoint") if not isinstance(endpoint, str) or not endpoint.strip(): raise ValueError(f"destination {name!r} endpoint must be a non-empty string") @@ -116,8 +133,41 @@ def _configured_destinations(raw): return validated +class _PartialSuccessRecorder: + """Records rejected_spans from OTLP responses, which the SDK exporters discard.""" + + def __init__(self, destination_name): + self.destination_name = destination_name + self.rejected_spans = 0 + self.error_message = "" + self._unparseable_logged = False + + def reset(self): + self.rejected_spans = 0 + self.error_message = "" + + def __call__(self, response, *args, **kwargs): + if not response.ok or not response.content: + return + try: + parsed = ExportTraceServiceResponse.FromString(response.content) + except Exception as e: + if not self._unparseable_logged: + self._unparseable_logged = True + logger.warning( + "Destination %s returned a body that is not an " + "ExportTraceServiceResponse, so partial rejections cannot be " + "detected there: %s", + self.destination_name, + e, + ) + return + self.rejected_spans = parsed.partial_success.rejected_spans + self.error_message = parsed.partial_success.error_message + + def _build_exporter(destination): - """Construct one OTLP exporter.""" + """Construct one OTLP exporter, and its partial-success recorder when supported.""" kwargs = { "endpoint": destination["endpoint"], } @@ -126,7 +176,15 @@ def _build_exporter(destination): if destination["protocol"] == "grpc": if destination["insecure"] is not None: kwargs["insecure"] = destination["insecure"] # fmt: skip - return GrpcOTLPSpanExporter(**kwargs) + global _grpc_partial_success_warned + if not _grpc_partial_success_warned: + _grpc_partial_success_warned = True + logger.warning( + "gRPC destinations cannot report partial rejections: the SDK " + "discards the response body, so spans this receiver rejects " + "individually will still be marked sent." + ) + return GrpcOTLPSpanExporter(**kwargs), None if destination["insecure"] is not None: logger.warning( @@ -134,35 +192,82 @@ def _build_exporter(destination): destination["name"], destination["insecure"], ) - return HttpOTLPSpanExporter(**kwargs) + recorder = _PartialSuccessRecorder(destination["name"]) + session = requests.Session() + session.hooks["response"].append(recorder) + kwargs["session"] = session + return HttpOTLPSpanExporter(**kwargs), recorder + + +def _probe_destination(destination_name, exporter): + """Report whether a destination actually answers, without failing startup. + + An empty export is a real OTLP request, so this exercises the endpoint, + path, TLS and auth rather than just proving a port is open. A destination + that is merely down yet is not fatal -- rows stay queued until it returns. + """ + try: + reachable = exporter.export([]) is SpanExportResult.SUCCESS + detail = "" + except Exception as e: + reachable = False + detail = f": {e}" + if reachable: + logger.info("OTel destination %s answered a connectivity check.", destination_name) + else: + logger.warning( + "OTel destination %s did not answer a connectivity check, so nothing " + "will reach it until that is fixed; spans stay queued meanwhile%s", + destination_name, + detail, + ) -def _build_processors(raw): - """Build one exporter/BatchSpanProcessor pair per configured destination.""" +def _build_exporters(raw): + """Build one OTLP exporter per configured destination.""" destinations = _configured_destinations(raw) if destinations is None: raise RuntimeError(f"{DESTINATIONS_KEY} is not set; otel.destinations is required") - processors = [] + exporters = [] + recorders = {} try: for destination in destinations: - exporter = _build_exporter(destination) - processors.append( - ( - destination["name"], - BatchSpanProcessor(exporter, schedule_delay_millis=1000), - ) - ) + exporter, recorder = _build_exporter(destination) + exporters.append((destination["name"], exporter)) + if recorder is not None: + recorders[destination["name"]] = recorder logger.info( "Configured OTel destination %s (%s).", destination["name"], destination["protocol"], ) + _probe_destination(destination["name"], exporter) except Exception: - for _, processor in processors: - processor.shutdown() + _shutdown_exporters( + exporters, + f"discarding destinations already built before {destination['name']!r} " + f"failed to build", + ) raise - return processors + _partial_success_recorders.clear() + _partial_success_recorders.update(recorders) + return exporters + + +def _shutdown_exporters(exporters, reason): + """Shut down each exporter, logging rather than propagating individual failures.""" + for destination_name, exporter in exporters: + try: + exporter.shutdown() + except Exception as e: + logger.error( + "Failed to shut down OTel destination %s while %s: %s", + destination_name, + reason, + e, + exc_info=True, + ) def _handle_shutdown(signum, frame): @@ -171,85 +276,292 @@ def _handle_shutdown(signum, frame): def _reload_destinations_if_changed(): - # Invalid Redis values are logged and ignored -- keep the previous processors + # Invalid Redis values are logged and ignored -- keep the previous exporters # running rather than tearing down a working config over a bad update. - global _processors, _last_destinations_raw - raw = _redis.get(DESTINATIONS_KEY) + global _exporters, _last_destinations_raw + try: + raw = _redis.get(DESTINATIONS_KEY) + except Exception as e: + logger.error( + "Failed to read %s from Redis; keeping the current %d destination(s): %s", + DESTINATIONS_KEY, + len(_exporters), + e, + exc_info=True, + ) + return if raw == _last_destinations_raw: return try: - new_processors = _build_processors(raw) + new_exporters = _build_exporters(raw) except Exception as e: logger.warning("Ignoring invalid %s update: %s", DESTINATIONS_KEY, e) return - for _, processor in _processors: - processor.shutdown() - _processors = new_processors + _shutdown_exporters(_exporters, "replacing it after a config reload") + _exporters = new_exporters _last_destinations_raw = raw - logger.info("Reloaded %d OTel destination(s) from Redis.", len(_processors)) - + logger.info("Reloaded %d OTel destination(s) from Redis.", len(_exporters)) -def _send_pending(): - """Convert and send each finished, not-yet-sent waiting row.""" - processors = _processors - if not processors: - raise RuntimeError("OTel exporter has no configured processors") - conn = sqlite3.connect(db.DB_PATH) +def _read_pending_rows(): + """Read one poll's worth of finished, not-yet-sent waiting rows.""" + try: + conn = sqlite3.connect(db.DB_PATH) + except Exception as e: + logger.error( + "Failed to open the waiting database at %s; no spans exported this poll: %s", + db.DB_PATH, + e, + exc_info=True, + ) + return [] conn.row_factory = sqlite3.Row try: - rows = conn.execute( + return conn.execute( "SELECT * FROM waiting WHERE finished_at IS NOT NULL " - "AND (sent IS NULL OR sent = 0)" + "AND (sent IS NULL OR sent = 0) LIMIT ?", + (MAX_SPANS_PER_POLL,), ).fetchall() + except Exception as e: + logger.error( + "Failed to read pending rows from %s; no spans exported this poll: %s", + db.DB_PATH, + e, + exc_info=True, + ) + return [] finally: conn.close() + + +def _reject_unexportable(span): + """Raise if the OTLP encoder cannot serialize this span. + + Encoding is what actually rejects a bad row (an out-of-range id, an + unencodable attribute), and it happens inside the batched export() call -- + where one row's failure discards every other span in the batch. Doing it per + row here keeps a single bad future from blocking everything behind it. + """ + encode_spans([span]).SerializePartialToString() + + +def _placeholder_after_repeated_failure(row, error): + """Return a placeholder span once a row has failed too often, else None. + + Only per-row conversion/encoding failures count here. An export failure is + shared by the whole batch, so counting those would replace every row in the + queue with a placeholder after a spell of receiver downtime. + """ + future_id = row["future_id"] + attempts = _row_export_failures.get(future_id, 0) + 1 + _row_export_failures[future_id] = attempts + if attempts < MAX_ROW_EXPORT_ATTEMPTS: + logger.error( + "Skipping waiting row %s -- cannot be exported (attempt %d of %d): %s", + future_id, + attempts, + MAX_ROW_EXPORT_ATTEMPTS, + error, + ) + return None + try: + placeholder = convert.invalid_row_placeholder_span(row, str(error)) + _reject_unexportable(placeholder) + except Exception as e: + logger.error( + "Waiting row %s cannot be exported and no placeholder could be built " + "for it either, so it stays in the queue: %s", + future_id, + e, + exc_info=True, + ) + return None + logger.warning( + "Waiting row %s failed %d export attempts; sending a placeholder span in " + "its place so the future is not lost silently: %s", + future_id, + attempts, + error, + ) + _row_export_failures.pop(future_id, None) + return placeholder + + +def _waiting_row_count(): + """Total rows in waiting, or None when the table cannot be counted.""" + try: + conn = sqlite3.connect(db.DB_PATH) + try: + return conn.execute("SELECT COUNT(*) FROM waiting").fetchone()[0] + finally: + conn.close() + except Exception as e: + logger.error( + "Failed to count rows in %s: %s", db.DB_PATH, e, exc_info=True + ) + return None + + +def _note_queue_state(found_pending): + """Warn once if no row ever appears, which sqlite cannot report as an error. + + A missing database file is created rather than refused, so a misdirected + DB_PATH looks exactly like an idle queue until someone compares the two paths. + """ + global _consecutive_empty_polls, _empty_queue_warned_at + if found_pending: + _consecutive_empty_polls = 0 + return + _consecutive_empty_polls += 1 + if _consecutive_empty_polls < EMPTY_QUEUE_WARNING_POLLS: + return + if ( + _empty_queue_warned_at is not None + and _consecutive_empty_polls - _empty_queue_warned_at < EMPTY_QUEUE_REWARN_POLLS + ): + return + # Only latch once the condition is confirmed, so a failed count re-checks + # next poll instead of silencing the warning for the life of the process. + if _waiting_row_count() != 0: + return + _empty_queue_warned_at = _consecutive_empty_polls + logger.warning( + "No rows have ever appeared in %s after %d consecutive polls. Spans are " + "only exported from this file, so GlobalController may be writing futures " + "to a different otel_queue.db than this process is reading.", + db.DB_PATH, + _consecutive_empty_polls, + ) + + +def _send_pending(): + """Export finished, not-yet-sent waiting rows and mark them only once delivered.""" + exporters = _exporters + if not exporters: + raise RuntimeError("OTel exporter has no configured destinations") + + rows = _read_pending_rows() + _note_queue_state(bool(rows)) if not rows: return - sent_count = 0 + + spans = [] + future_ids = [] for row in rows: try: span = convert.waiting_row_to_span(row) + _reject_unexportable(span) + except Exception as e: + span = _placeholder_after_repeated_failure(row, e) + if span is None: + continue + else: + _row_export_failures.pop(row["future_id"], None) + spans.append(span) + future_ids.append(row["future_id"]) + if not spans: + return + + failed_destinations = [] + for destination_name, exporter in exporters: + recorder = _partial_success_recorders.get(destination_name) + if recorder is not None: + recorder.reset() + delivered = False + try: + result = exporter.export(spans) except Exception as e: logger.error( - "Skipping waiting row %s -- failed to convert: %s", row["future_id"], e + "Destination %s raised while exporting %d span(s): %s", + destination_name, + len(spans), + e, ) - continue - - failed_destinations = [] - for destination_name, processor in processors: - try: - processor.on_end(span) - except Exception as e: - # Still offer the span to the remaining processors. The row is only - # acknowledged when every destination accepted it, so a failed - # destination will be retried by the next poll. - failed_destinations.append(destination_name) + else: + if result is not SpanExportResult.SUCCESS: logger.error( - "Destination %s rejected waiting row %s: %s", + "Destination %s failed to export %d span(s).", destination_name, - row["future_id"], - e, + len(spans), + ) + elif recorder is not None and recorder.rejected_spans: + logger.error( + "Destination %s accepted the request but rejected %d of %d " + "span(s), so the batch is not acknowledged: %s", + destination_name, + recorder.rejected_spans, + len(spans), + recorder.error_message or "no reason given", ) - if failed_destinations: - continue - db.mark_sent(row["future_id"]) - sent_count += 1 - logger.info("Queued %d span(s) for all configured OTel destinations.", sent_count) + else: + delivered = True + if not delivered: + failed_destinations.append(destination_name) + elif _destination_healthy.get(destination_name) is False: + logger.info( + "OTel destination %s is accepting spans again.", destination_name + ) + _destination_healthy[destination_name] = delivered + + # A partial success still leaves every row unsent, so the retry re-delivers to + # destinations that already accepted the batch; span ids are deterministic, so + # the duplicates collapse at the backend. + if failed_destinations: + logger.warning( + "Leaving %d span(s) unsent for retry; failed destination(s): %s", + len(spans), + ", ".join(failed_destinations), + ) + return + + try: + db.mark_sent_many(future_ids, db.DB_PATH) + except Exception as e: + logger.error( + "Exported %d span(s) but failed to mark them sent in %s -- they will be " + "re-exported and duplicated on the next poll: %s", + len(future_ids), + db.DB_PATH, + e, + exc_info=True, + ) + return + logger.info("Exported %d span(s) to all configured OTel destinations.", len(spans)) def main(): - global _processors, _redis, _last_destinations_raw + global _exporters, _redis, _last_destinations_raw signal.signal(signal.SIGTERM, _handle_shutdown) signal.signal(signal.SIGINT, _handle_shutdown) - db.init_db() - # 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)) + try: + db.init_db() + except Exception as e: + logger.error( + "Fatal: cannot initialize the waiting database at %s: %s", + db.DB_PATH, + e, + exc_info=True, + ) + raise + try: + _redis = RedisClient(host="host.docker.internal") + _last_destinations_raw = _redis.get(DESTINATIONS_KEY) + except Exception as e: + logger.error( + "Fatal: cannot reach Redis to read %s: %s", DESTINATIONS_KEY, e, exc_info=True + ) + raise + try: + _exporters = _build_exporters(_last_destinations_raw) + except Exception as e: + logger.error( + "Fatal: cannot build OTel destinations from %s: %s", + DESTINATIONS_KEY, + e, + exc_info=True, + ) + raise + logger.info("OTel exporter process started with %d destination(s).", len(_exporters)) try: last_poll = 0 while _running: @@ -258,17 +570,16 @@ def main(): _reload_destinations_if_changed() _send_pending() except Exception as e: - logger.warning("Poll cycle failed (non-fatal): %s", e) + logger.error( + "Unexpected error in OTel export poll cycle (non-fatal, " + "retrying next tick): %s", + e, + exc_info=True, + ) last_poll = time.time() time.sleep(1) finally: - for destination_name, processor in _processors: - try: - processor.shutdown() - except Exception as e: - logger.error( - "Failed to shut down OTel destination %s: %s", destination_name, e - ) + _shutdown_exporters(_exporters, "shutting the exporter process down") logger.info("OTel exporter process exiting.") diff --git a/tests/test_otel_exporter_fanout.py b/tests/test_otel_exporter_fanout.py index 38d2cd4..8e47ad8 100644 --- a/tests/test_otel_exporter_fanout.py +++ b/tests/test_otel_exporter_fanout.py @@ -15,6 +15,12 @@ # therefore imports ``convert`` and ``db`` as top-level modules. sys.path.insert(0, os.path.join(ROOT, "canyonos_core", "OTLP_Exporter")) +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( # noqa: E402 + ExportTraceServiceResponse, +) +from opentelemetry.sdk.trace.export import SpanExportResult # noqa: E402 + +import convert # noqa: E402 import db # noqa: E402 import otel_exporter # noqa: E402 @@ -62,11 +68,9 @@ def _destination_config(): }, ] - def test_build_processors_constructs_mixed_exporters_with_explicit_args(self): + def test_build_exporters_constructs_mixed_exporters_with_explicit_args(self): grpc_exporter = object() http_exporter = object() - grpc_processor = MagicMock(name="grpc_processor") - http_processor = MagicMock(name="http_processor") destinations = self._destination_config() with patch.object( @@ -77,15 +81,11 @@ def test_build_processors_constructs_mixed_exporters_with_explicit_args(self): otel_exporter, "HttpOTLPSpanExporter", return_value=http_exporter, - ) as http_constructor, patch.object( - otel_exporter, - "BatchSpanProcessor", - side_effect=[grpc_processor, http_processor], - ) as processor_constructor: - processors = otel_exporter._build_processors(json.dumps(destinations)) + ) as http_constructor: + exporters = otel_exporter._build_exporters(json.dumps(destinations)) self.assertEqual( - processors, [("railway", grpc_processor), ("langfuse", http_processor)] + exporters, [("railway", grpc_exporter), ("langfuse", http_exporter)] ) grpc_constructor.assert_called_once_with( endpoint="receiver.example:4317", @@ -97,18 +97,12 @@ def test_build_processors_constructs_mixed_exporters_with_explicit_args(self): endpoint="https://langfuse.example/api/public/otel", headers={"authorization": "Basic secret"}, timeout=7, - ) - self.assertEqual( - processor_constructor.call_args_list, - [ - unittest.mock.call(grpc_exporter, schedule_delay_millis=1000), - unittest.mock.call(http_exporter, schedule_delay_millis=1000), - ], + session=unittest.mock.ANY, ) - def test_build_processors_raises_when_destinations_raw_is_none(self): + def test_build_exporters_raises_when_destinations_raw_is_none(self): with self.assertRaisesRegex(RuntimeError, "otel.destinations is required"): - otel_exporter._build_processors(None) + otel_exporter._build_exporters(None) def test_configured_destinations_rejects_malformed_empty_and_duplicate_values(self): invalid_values = [ @@ -179,7 +173,7 @@ def _insert_pending_row(self): ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0) """, ( - "00112233445566778899aabbccddeeff", + "0011223344556677", "ffeeddccbbaa99887766554433221100", 1.0, 2.0, @@ -193,72 +187,86 @@ def _insert_pending_row(self): finally: conn.close() - def test_send_pending_delivers_the_same_span_to_every_processor(self): + def _assert_row_unsent(self): + conn = sqlite3.connect(self.db_path) + try: + self.assertEqual(conn.execute("SELECT sent FROM waiting").fetchone()[0], 0) + finally: + conn.close() + + def test_send_pending_delivers_the_same_spans_to_every_destination(self): self._insert_pending_row() first = MagicMock(name="first") + first.export.return_value = SpanExportResult.SUCCESS second = MagicMock(name="second") + second.export.return_value = SpanExportResult.SUCCESS with patch.object(otel_exporter.db, "DB_PATH", self.db_path), patch.object( - otel_exporter.db, "mark_sent" - ) as mark_sent: - otel_exporter._processors = [("railway", first), ("langfuse", second)] + otel_exporter.db, "mark_sent_many" + ) as mark_sent_many: + otel_exporter._exporters = [("railway", first), ("langfuse", second)] otel_exporter._send_pending() - first.on_end.assert_called_once() - second.on_end.assert_called_once() - self.assertIs(first.on_end.call_args.args[0], second.on_end.call_args.args[0]) - mark_sent.assert_called_once_with("00112233445566778899aabbccddeeff") + first.export.assert_called_once() + second.export.assert_called_once() + self.assertEqual( + first.export.call_args.args[0], second.export.call_args.args[0] + ) + mark_sent_many.assert_called_once_with(["0011223344556677"], self.db_path) - def test_send_pending_attempts_remaining_processors_and_leaves_row_unsent_on_failure(self): + def test_send_pending_leaves_rows_unsent_when_a_destination_returns_failure(self): + self._insert_pending_row() + rejecting = MagicMock(name="rejecting") + rejecting.export.return_value = SpanExportResult.FAILURE + with patch.object(otel_exporter.db, "DB_PATH", self.db_path): + otel_exporter._exporters = [("railway", rejecting)] + otel_exporter._send_pending() + + rejecting.export.assert_called_once() + self._assert_row_unsent() + + def test_send_pending_tries_every_destination_and_leaves_rows_unsent_on_error(self): self._insert_pending_row() failed = MagicMock(name="failed") - failed.on_end.side_effect = RuntimeError("destination unavailable") + failed.export.side_effect = RuntimeError("destination unavailable") remaining = MagicMock(name="remaining") + remaining.export.return_value = SpanExportResult.SUCCESS with patch.object(otel_exporter.db, "DB_PATH", self.db_path), patch.object( - otel_exporter.db, "mark_sent" - ) as mark_sent: - otel_exporter._processors = [("railway", failed), ("langfuse", remaining)] + otel_exporter.db, "mark_sent_many" + ) as mark_sent_many: + otel_exporter._exporters = [("railway", failed), ("langfuse", remaining)] otel_exporter._send_pending() - failed.on_end.assert_called_once() - remaining.on_end.assert_called_once() - mark_sent.assert_not_called() + failed.export.assert_called_once() + remaining.export.assert_called_once() + mark_sent_many.assert_not_called() + self._assert_row_unsent() - conn = sqlite3.connect(self.db_path) - try: - self.assertEqual(conn.execute("SELECT sent FROM waiting").fetchone()[0], 0) - finally: - conn.close() - - def test_processor_construction_failure_shuts_down_already_built_processors(self): - first_processor = MagicMock(name="first_processor") + def test_exporter_construction_failure_shuts_down_already_built_exporters(self): + first_exporter = MagicMock(name="first_exporter") destinations = self._destination_config() with patch.object( otel_exporter, "GrpcOTLPSpanExporter", - return_value=object(), + return_value=first_exporter, ), patch.object( otel_exporter, "HttpOTLPSpanExporter", side_effect=RuntimeError("bad HTTP exporter"), - ), patch.object( - otel_exporter, - "BatchSpanProcessor", - return_value=first_processor, ): with self.assertRaisesRegex(RuntimeError, "bad HTTP exporter"): - otel_exporter._build_processors(json.dumps(destinations)) + otel_exporter._build_exporters(json.dumps(destinations)) - first_processor.shutdown.assert_called_once_with() + first_exporter.shutdown.assert_called_once_with() class OTelExporterReloadTests(unittest.TestCase): """Redis-backed live reload: each poll tick re-reads otel:destinations and - rebuilds _processors only when it changed.""" + rebuilds _exporters only when it changed.""" def setUp(self): self._orig_redis = otel_exporter._redis self._orig_raw = otel_exporter._last_destinations_raw - self._orig_processors = otel_exporter._processors + self._orig_exporters = otel_exporter._exporters self.store = {} class FakeRedis: @@ -267,68 +275,531 @@ def get(_self, key): otel_exporter._redis = FakeRedis() otel_exporter._last_destinations_raw = None - otel_exporter._processors = [] + otel_exporter._exporters = [] def tearDown(self): otel_exporter._redis = self._orig_redis otel_exporter._last_destinations_raw = self._orig_raw - otel_exporter._processors = self._orig_processors + otel_exporter._exporters = self._orig_exporters - def test_reload_builds_processors_from_redis_on_first_read(self): + def test_reload_builds_exporters_from_redis_on_first_read(self): destinations = self._config_for("a", "grpc") self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(destinations) - with patch.object(otel_exporter, "GrpcOTLPSpanExporter", return_value=object()), patch.object( - otel_exporter, "BatchSpanProcessor", return_value=MagicMock(name="p") + with patch.object( + otel_exporter, "GrpcOTLPSpanExporter", return_value=MagicMock(name="e") ): otel_exporter._reload_destinations_if_changed() - self.assertEqual([name for name, _ in otel_exporter._processors], ["a"]) + self.assertEqual([name for name, _ in otel_exporter._exporters], ["a"]) def test_reload_is_a_noop_when_redis_value_is_unchanged(self): destinations = self._config_for("a", "grpc") self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(destinations) - with patch.object(otel_exporter, "GrpcOTLPSpanExporter", return_value=object()), patch.object( - otel_exporter, "BatchSpanProcessor", return_value=MagicMock(name="p") - ) as processor_ctor: + with patch.object( + otel_exporter, "GrpcOTLPSpanExporter", return_value=MagicMock(name="e") + ) as exporter_ctor: otel_exporter._reload_destinations_if_changed() otel_exporter._reload_destinations_if_changed() - processor_ctor.assert_called_once() + exporter_ctor.assert_called_once() - def test_reload_rebuilds_and_shuts_down_old_processors_when_redis_value_changes(self): - old_processor = MagicMock(name="old") - new_processor = MagicMock(name="new") + def test_reload_rebuilds_and_shuts_down_old_exporters_when_redis_value_changes(self): + old_exporter = MagicMock(name="old") + new_exporter = MagicMock(name="new") self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(self._config_for("a", "grpc")) - with patch.object(otel_exporter, "GrpcOTLPSpanExporter", return_value=object()), patch.object( - otel_exporter, "BatchSpanProcessor", return_value=old_processor + with patch.object( + otel_exporter, "GrpcOTLPSpanExporter", return_value=old_exporter ): otel_exporter._reload_destinations_if_changed() self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(self._config_for("b", "http")) - with patch.object(otel_exporter, "HttpOTLPSpanExporter", return_value=object()), patch.object( - otel_exporter, "BatchSpanProcessor", return_value=new_processor + with patch.object( + otel_exporter, "HttpOTLPSpanExporter", return_value=new_exporter ): otel_exporter._reload_destinations_if_changed() - old_processor.shutdown.assert_called_once_with() - self.assertEqual([name for name, _ in otel_exporter._processors], ["b"]) + old_exporter.shutdown.assert_called_once_with() + self.assertEqual([name for name, _ in otel_exporter._exporters], ["b"]) - def test_reload_keeps_previous_processors_when_new_redis_value_is_invalid(self): + def test_reload_keeps_previous_exporters_when_new_redis_value_is_invalid(self): good = MagicMock(name="good") self.store[otel_exporter.DESTINATIONS_KEY] = json.dumps(self._config_for("a", "grpc")) - with patch.object(otel_exporter, "GrpcOTLPSpanExporter", return_value=object()), patch.object( - otel_exporter, "BatchSpanProcessor", return_value=good - ): + with patch.object(otel_exporter, "GrpcOTLPSpanExporter", return_value=good): otel_exporter._reload_destinations_if_changed() self.store[otel_exporter.DESTINATIONS_KEY] = "not json" otel_exporter._reload_destinations_if_changed() good.shutdown.assert_not_called() - self.assertEqual([name for name, _ in otel_exporter._processors], ["a"]) + self.assertEqual([name for name, _ in otel_exporter._exporters], ["a"]) @staticmethod def _config_for(name, protocol): return [{"name": name, "protocol": protocol, "endpoint": "host:1"}] +class OTelExporterQuietFailureTests(unittest.TestCase): + """Failures sqlite and the pricing lookups cannot surface on their own.""" + + def setUp(self): + self.db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + self.db_path = self.db_file.name + self.db_file.close() + db.init_db(self.db_path) + self._orig_exporters = otel_exporter._exporters + otel_exporter._consecutive_empty_polls = 0 + otel_exporter._empty_queue_warned = False + db._cost_failures_logged.clear() + exporter = MagicMock(name="exporter") + exporter.export.return_value = SpanExportResult.SUCCESS + otel_exporter._exporters = [("live", exporter)] + + def tearDown(self): + otel_exporter._exporters = self._orig_exporters + otel_exporter._consecutive_empty_polls = 0 + otel_exporter._empty_queue_warned = False + db._cost_failures_logged.clear() + os.unlink(self.db_path) + + def _poll(self, times): + with patch.object(otel_exporter.db, "DB_PATH", self.db_path): + for _ in range(times): + otel_exporter._send_pending() + + def _insert(self, future_id, sent): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO waiting (future_id, session_id, started_at, finished_at," + " failed, name, sent) VALUES (?, ?, 1.0, 2.0, 0, 'A.b', ?)", + (future_id, "ffeeddccbbaa99887766554433221100", sent), + ) + conn.commit() + finally: + conn.close() + + def test_queue_that_never_receives_rows_warns_once_and_names_the_path(self): + # sqlite creates a missing file rather than failing, so a misdirected + # DB_PATH is indistinguishable from an idle queue without this warning. + with self.assertLogs(otel_exporter.logger, level="WARNING") as captured: + self._poll(otel_exporter.EMPTY_QUEUE_WARNING_POLLS + 4) + + warnings = [m for m in captured.output if "No rows have ever appeared" in m] + self.assertEqual(len(warnings), 1) + self.assertIn(self.db_path, warnings[0]) + + def test_no_warning_before_the_threshold(self): + with self.assertNoLogs(otel_exporter.logger, level="WARNING"): + self._poll(otel_exporter.EMPTY_QUEUE_WARNING_POLLS - 1) + + def test_fully_exported_queue_does_not_warn(self): + # An empty pending set means "everything is delivered" here, not "wrong + # database" -- the row count is what separates the two. + self._insert("0011223344556677", sent=1) + with self.assertNoLogs(otel_exporter.logger, level="WARNING"): + self._poll(otel_exporter.EMPTY_QUEUE_WARNING_POLLS + 4) + + def test_cost_lookup_failure_is_logged_once_per_kind_and_keeps_the_row(self): + rows = [ + { + "future_id": "0011223344556677", + "request_id": "r1", + "agent": "a1", + "created_at": 1.0, + "finished_at": 2.0, + "model": "m", + "input_token_count": 1, + "output_token_count": 1, + "token_count": 2, + } + ] + boom = RuntimeError("no aws_instance_pricing table") + with patch.object(db.pricing, "compute_token_cost", side_effect=boom), patch.object( + db.pricing, "compute_server_cost", side_effect=boom + ), self.assertLogs(db.logger, level="WARNING") as captured: + for index in range(3): + rows[0]["future_id"] = f"001122334455667{index}" + db.write_waiting_rows(rows, None, "proj", db_path=self.db_path) + + self.assertEqual( + len([m for m in captured.output if "Token cost lookup failed" in m]), 1 + ) + self.assertEqual( + len([m for m in captured.output if "Server cost lookup failed" in m]), 1 + ) + + conn = sqlite3.connect(self.db_path) + try: + written, cost = conn.execute( + "SELECT COUNT(*), COALESCE(SUM(total_cost), 0) FROM waiting" + ).fetchone() + finally: + conn.close() + self.assertEqual(written, 3) + self.assertEqual(cost, 0) + + +class OTelExporterDeliveryIntegrityTests(unittest.TestCase): + """Rows that cannot be delivered must not block or be reported as delivered.""" + + def setUp(self): + self.db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + self.db_path = self.db_file.name + self.db_file.close() + db.init_db(self.db_path) + self._orig_exporters = otel_exporter._exporters + self._orig_recorders = dict(otel_exporter._partial_success_recorders) + + def tearDown(self): + otel_exporter._exporters = self._orig_exporters + otel_exporter._partial_success_recorders.clear() + otel_exporter._partial_success_recorders.update(self._orig_recorders) + os.unlink(self.db_path) + + def _insert(self, future_id): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO waiting (future_id, session_id, started_at, finished_at," + " failed, name, sent) VALUES (?, ?, 1.0, 2.0, 0, 'A.b', 0)", + (future_id, "ffeeddccbbaa99887766554433221100"), + ) + conn.commit() + finally: + conn.close() + + def _sent(self, future_id): + conn = sqlite3.connect(self.db_path) + try: + return conn.execute( + "SELECT sent FROM waiting WHERE future_id = ?", (future_id,) + ).fetchone()[0] + finally: + conn.close() + + def test_one_unexportable_row_does_not_block_the_rest_of_the_batch(self): + good = ["0011223344556677", "1122334455667788"] + poison = "00112233445566778899aabbccddeeff" # 128-bit, overflows span_id + for future_id in good: + self._insert(future_id) + self._insert(poison) + + exporter = MagicMock(name="exporter") + exporter.export.return_value = SpanExportResult.SUCCESS + otel_exporter._exporters = [("live", exporter)] + with patch.object(otel_exporter.db, "DB_PATH", self.db_path), patch.object( + otel_exporter.db, "mark_sent_many" + ) as mark_sent_many: + otel_exporter._send_pending() + + self.assertEqual(len(exporter.export.call_args.args[0]), len(good)) + mark_sent_many.assert_called_once_with(good, self.db_path) + + def test_unexportable_row_is_named_in_the_log(self): + poison = "00112233445566778899aabbccddeeff" + self._insert(poison) + exporter = MagicMock(name="exporter") + exporter.export.return_value = SpanExportResult.SUCCESS + otel_exporter._exporters = [("live", exporter)] + + with patch.object(otel_exporter.db, "DB_PATH", self.db_path), self.assertLogs( + otel_exporter.logger, level="ERROR" + ) as captured: + otel_exporter._send_pending() + + self.assertTrue(any(poison in line for line in captured.output)) + exporter.export.assert_not_called() + + def test_partial_success_rejection_leaves_the_batch_unsent(self): + # The SDK returns SUCCESS for a 200 that rejected spans, so delivery is + # only known by reading partial_success off the response. + future_id = "0011223344556677" + self._insert(future_id) + recorder = otel_exporter._PartialSuccessRecorder("live") + + def export_with_partial_rejection(spans): + # The real hook fires during the HTTP call, after _send_pending has + # reset the recorder, so the mock has to populate it the same way. + recorder.rejected_spans = 1 + recorder.error_message = "span rejected" + return SpanExportResult.SUCCESS + + exporter = MagicMock(name="exporter") + exporter.export.side_effect = export_with_partial_rejection + otel_exporter._exporters = [("live", exporter)] + otel_exporter._partial_success_recorders.clear() + otel_exporter._partial_success_recorders["live"] = recorder + + with patch.object(otel_exporter.db, "DB_PATH", self.db_path): + otel_exporter._send_pending() + + self.assertEqual(self._sent(future_id), 0) + + def test_recorder_reads_rejected_spans_off_an_ok_response(self): + response = ExportTraceServiceResponse() + response.partial_success.rejected_spans = 2 + response.partial_success.error_message = "two bad spans" + http_response = MagicMock(ok=True, content=response.SerializeToString()) + + recorder = otel_exporter._PartialSuccessRecorder("live") + recorder(http_response) + + self.assertEqual(recorder.rejected_spans, 2) + self.assertEqual(recorder.error_message, "two bad spans") + + +class OTelExporterConfigValidationTests(unittest.TestCase): + def test_unsupported_protocol_is_rejected_instead_of_defaulting_to_http(self): + for protocol in ("grcp", None, "", 7): + with self.subTest(protocol=protocol): + with self.assertRaisesRegex(ValueError, "protocol must be one of"): + otel_exporter._configured_destinations( + json.dumps([{"name": "a", "protocol": protocol, "endpoint": "h:1"}]) + ) + + def test_protocol_case_is_normalized(self): + for given, expected in (("HTTP", "http"), ("GRPC", "grpc"), ("Http/Protobuf", "http/protobuf")): + with self.subTest(protocol=given): + parsed = otel_exporter._configured_destinations( + json.dumps([{"name": "a", "protocol": given, "endpoint": "h:1"}]) + ) + self.assertEqual(parsed[0]["protocol"], expected) + + def test_supported_protocols_are_accepted(self): + for protocol in otel_exporter.SUPPORTED_PROTOCOLS: + with self.subTest(protocol=protocol): + parsed = otel_exporter._configured_destinations( + json.dumps([{"name": "a", "protocol": protocol, "endpoint": "h:1"}]) + ) + self.assertEqual(parsed[0]["protocol"], protocol) + + +class OTelExporterRowFidelityTests(unittest.TestCase): + def test_missing_identifiers_are_logged_not_silently_dropped(self): + with self.assertLogs(db.logger, level="WARNING") as captured: + db.write_waiting_rows( + [{"future_id": "0011223344556677", "request_id": ""}], + None, + "proj", + db_path=":memory:", + ) + self.assertTrue(any("request_id" in line for line in captured.output)) + + def test_absent_error_name_does_not_fabricate_an_exception_type(self): + span = convert.waiting_row_to_span( + { + "future_id": "0011223344556677", + "session_id": "ffeeddccbbaa99887766554433221100", + "parent_id": None, + "failed": 1, + "error_name": None, + "error_message": "timed out", + "name": "A.b", + "started_at": 1.0, + "finished_at": 2.0, + } + ) + self.assertNotIn("exception.type", span.events[0].attributes) + self.assertEqual(span.events[0].attributes["exception.message"], "timed out") + + def test_empty_output_is_preserved_rather_than_treated_as_absent(self): + self.assertEqual(db._normalize_json_text(""), '""') + self.assertIsNone(db._normalize_json_text(None)) + + +class OTelExporterPlaceholderTests(unittest.TestCase): + """A row that can never be encoded is retired, not retried forever.""" + + def setUp(self): + self.db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + self.db_path = self.db_file.name + self.db_file.close() + db.init_db(self.db_path) + self._orig_exporters = otel_exporter._exporters + otel_exporter._row_export_failures.clear() + self.poison = "00112233445566778899aabbccddeeff" + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO waiting (future_id, session_id, started_at, finished_at," + " failed, name, sent) VALUES (?, ?, 1.0, 2.0, 0, 'A.b', 0)", + (self.poison, "ffeeddccbbaa99887766554433221100"), + ) + conn.commit() + finally: + conn.close() + + def tearDown(self): + otel_exporter._exporters = self._orig_exporters + otel_exporter._row_export_failures.clear() + os.unlink(self.db_path) + + def _poll_once(self, exporter): + otel_exporter._exporters = [("live", exporter)] + with patch.object(otel_exporter.db, "DB_PATH", self.db_path): + otel_exporter._send_pending() + + def test_placeholder_is_sent_only_after_the_attempt_limit(self): + exporter = MagicMock(name="exporter") + exporter.export.return_value = SpanExportResult.SUCCESS + + for _ in range(otel_exporter.MAX_ROW_EXPORT_ATTEMPTS - 1): + self._poll_once(exporter) + exporter.export.assert_not_called() + + self._poll_once(exporter) + sent_spans = exporter.export.call_args.args[0] + self.assertEqual(len(sent_spans), 1) + self.assertEqual(sent_spans[0].name, "canyonos.invalid_span") + + conn = sqlite3.connect(self.db_path) + try: + self.assertEqual(conn.execute("SELECT sent FROM waiting").fetchone()[0], 1) + finally: + conn.close() + + def test_export_failures_do_not_count_toward_the_row_limit(self): + # Otherwise a spell of receiver downtime would replace every queued row + # with a placeholder. + good = "1122334455667788" + conn = sqlite3.connect(self.db_path) + try: + conn.execute("DELETE FROM waiting") + conn.execute( + "INSERT INTO waiting (future_id, session_id, started_at, finished_at," + " failed, name, sent) VALUES (?, ?, 1.0, 2.0, 0, 'A.b', 0)", + (good, "ffeeddccbbaa99887766554433221100"), + ) + conn.commit() + finally: + conn.close() + + down = MagicMock(name="down") + down.export.return_value = SpanExportResult.FAILURE + for _ in range(otel_exporter.MAX_ROW_EXPORT_ATTEMPTS + 3): + self._poll_once(down) + + self.assertNotIn(good, otel_exporter._row_export_failures) + recovered = MagicMock(name="recovered") + recovered.export.return_value = SpanExportResult.SUCCESS + self._poll_once(recovered) + self.assertEqual( + recovered.export.call_args.args[0][0].name, "A.b" + ) + + def test_placeholder_is_not_disguised_as_an_agent_failure(self): + session_id = "ffeeddccbbaa99887766554433221100" + placeholder = convert.invalid_row_placeholder_span( + { + "future_id": self.poison, + "session_id": session_id, + "started_at": 1.0, + "finished_at": 2.0, + }, + "int too big to convert", + ) + + self.assertEqual(placeholder.name, "canyonos.invalid_span") + self.assertIs(placeholder.attributes["canyonos.export.invalid"], True) + self.assertEqual(placeholder.attributes["canyonos.future_id"], self.poison) + self.assertEqual(placeholder.events, ()) + self.assertEqual(placeholder.context.trace_id, int(session_id, 16)) + self.assertTrue(0 < placeholder.context.span_id <= 2**64 - 1) + + +class OTelExporterLifecycleLoggingTests(unittest.TestCase): + """Connectivity and recovery must be visible without waiting for traffic.""" + + def setUp(self): + self._orig_exporters = otel_exporter._exporters + otel_exporter._destination_healthy.clear() + + def tearDown(self): + otel_exporter._exporters = self._orig_exporters + otel_exporter._destination_healthy.clear() + + def test_reachable_destination_is_reported_at_build_time(self): + exporter = MagicMock(name="exporter") + exporter.export.return_value = SpanExportResult.SUCCESS + with self.assertLogs(otel_exporter.logger, level="INFO") as captured: + otel_exporter._probe_destination("live", exporter) + exporter.export.assert_called_once_with([]) + self.assertTrue( + any("answered a connectivity check" in line for line in captured.output) + ) + + def test_unreachable_destination_is_reported_at_build_time(self): + exporter = MagicMock(name="exporter") + exporter.export.side_effect = RuntimeError("connection refused") + with self.assertLogs(otel_exporter.logger, level="WARNING") as captured: + otel_exporter._probe_destination("dead", exporter) + self.assertTrue( + any("did not answer a connectivity check" in line for line in captured.output) + ) + self.assertTrue(any("connection refused" in line for line in captured.output)) + + def test_recovery_is_logged_only_after_a_failure(self): + db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + db_path = db_file.name + db_file.close() + self.addCleanup(os.unlink, db_path) + db.init_db(db_path) + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO waiting (future_id, session_id, started_at, finished_at," + " failed, name, sent) VALUES ('0011223344556677'," + " 'ffeeddccbbaa99887766554433221100', 1.0, 2.0, 0, 'A.b', 0)" + ) + conn.commit() + finally: + conn.close() + + exporter = MagicMock(name="exporter") + exporter.export.return_value = SpanExportResult.FAILURE + otel_exporter._exporters = [("live", exporter)] + with patch.object(otel_exporter.db, "DB_PATH", db_path): + otel_exporter._send_pending() + self.assertIs(otel_exporter._destination_healthy["live"], False) + + exporter.export.return_value = SpanExportResult.SUCCESS + with self.assertLogs(otel_exporter.logger, level="INFO") as captured: + otel_exporter._send_pending() + + self.assertTrue( + any("is accepting spans again" in line for line in captured.output) + ) + + def test_steady_success_does_not_log_recovery(self): + db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + db_path = db_file.name + db_file.close() + self.addCleanup(os.unlink, db_path) + db.init_db(db_path) + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO waiting (future_id, session_id, started_at, finished_at," + " failed, name, sent) VALUES ('0011223344556677'," + " 'ffeeddccbbaa99887766554433221100', 1.0, 2.0, 0, 'A.b', 0)" + ) + conn.commit() + finally: + conn.close() + + exporter = MagicMock(name="exporter") + exporter.export.return_value = SpanExportResult.SUCCESS + otel_exporter._destination_healthy["live"] = True + otel_exporter._exporters = [("live", exporter)] + + with patch.object(otel_exporter.db, "DB_PATH", db_path), self.assertLogs( + otel_exporter.logger, level="INFO" + ) as captured: + otel_exporter._send_pending() + + self.assertTrue(any("Exported 1 span(s)" in line for line in captured.output)) + self.assertFalse( + any("is accepting spans again" in line for line in captured.output) + ) + + if __name__ == "__main__": unittest.main() From 9dce8a5dbbbc6cfa3cd3e66c58817b37530204dc Mon Sep 17 00:00:00 2001 From: Saaketh <74433967+Saaketh0@users.noreply.github.com> Date: Thu, 10 Sep 2026 11:52:37 -0700 Subject: [PATCH 05/10] CAN-333: CLI fixes and README consolidation (#87) - Merge cli/README.md's install/command/usage docs into root README.md, correct stale onboarding (new-app command name, workflow vs dashboard port) and add the global_controller.yaml config walkthrough. - Trim cli/README.md to a short pointer at ARCHITECTURE.md. - Un-embed the accidentally nested examples/repo git repo (stale gitlink in the index, no actual .git left on disk) from a prior state -- kept out of scope here, that lives on docs/fleshing-docs-clean. - Assorted small fixes across cli/canyonos/ (constants, deploy, test, theme, init, quit, stop, verify, dashboard_stack, dashboard.compose.yml) and their tests. Co-authored-by: Claude Sonnet 5 --- .gitignore | 2 + README.md | 1 + cli/README.md | 64 ++++-- cli/canyonos/constants.py | 86 +++++++- cli/canyonos/dashboard.compose.yml | 22 +-- cli/canyonos/dashboard_stack.py | 25 +++ cli/canyonos/deploy.py | 92 ++++++--- cli/canyonos/init.py | 13 ++ cli/canyonos/quit.py | 7 +- cli/canyonos/stop.py | 2 + cli/canyonos/test.py | 127 ++++++------ cli/canyonos/theme.py | 39 +++- cli/canyonos/verify.py | 178 +---------------- tests/test_canyonos_deploy.py | 90 +++++++++ tests/test_canyonos_test.py | 263 ++++--------------------- tests/test_dashboard_stack.py | 28 +++ tests/test_deploy_progress.py | 4 +- tests/test_instance_manager_runtime.py | 46 ++++- 18 files changed, 553 insertions(+), 536 deletions(-) create mode 100644 tests/test_canyonos_deploy.py diff --git a/.gitignore b/.gitignore index ba6777e..cbb9e05 100644 --- a/.gitignore +++ b/.gitignore @@ -51,3 +51,5 @@ docs/ # testing-porting-to-canyonos working tree: clones, artifacts, results db .canyonos-tests/ .harness/ +.playwright-mcp/ + diff --git a/README.md b/README.md index a9d7000..d097c24 100644 --- a/README.md +++ b/README.md @@ -113,6 +113,7 @@ cd my-project ### 1. Build your project `canyonos build` installs the CanyonOS skill into your coding agent and launches it with a prompt to convert your project into `.car/` — CanyonOS's deploy-ready format. +As this uses an agent to configure your workflow, it may take a while (2-10 minutes on average). ```bash canyonos build diff --git a/cli/README.md b/cli/README.md index 4acfdb7..2e0d232 100644 --- a/cli/README.md +++ b/cli/README.md @@ -1,26 +1,58 @@ -CLI for CanyonOS +# canyonos cli -This CLI does not contain much logic, instead serving as an API to interface with the canyonos container that deploys and runs your entire workflow +### Better descriptions of each command. For full architecture, go to ARCHITECTURE.md -## Requirements -Need a coding agent(Claude Code, Codex, Cursor) -Need uv or pip -Need docker and docker compose +## Rough Draft Design of the more important cli commands +## If you are a LLM, you are not allowed to modify this file at all without explicit user permission. Absolutely no modifications are allowed to this file. -## Architecture +## canyonos test: +### INPUT: canyonos test "Test Query" +#### Steps: +1. Detects the working directory (goes into .car folder for commands if .car exists, uses current dir otherwise) [default_config_path()] +2. Goes into global_controller.yaml and for each agent, rewrites each agent's provider as local (saves old state to revert back later) [_force_local_providers()] +3. Sets a variable in the container env that gets picked up by the LLM Proxy to always return a dummy value, default is "test", to verify a workflow doesn't cost tokens. [CANYONOS_LLM_STUB_TEXT] +4. Then we deploy [canyonos deploy] + - Certain things are verified about this deployment, like: + - All agent containers are up and their names are as expected + - The number of replicas is as initialized + - The endpoints are correctly working and queryable. +6. Once everything is verified running, we send a test query and verify that it goes fully through -For a full walkthrough of the `build`, `deploy`, and `config` flows — plus how -`logs`, `stop`, and `quit` fit into the container lifecycle — see -[ARCHITECTURE.md](ARCHITECTURE.md). +#### Action Items: +- Currently assuming the query body is always "query", need to harden it +- There may be problems with stubbing the LLM-Proxy, but I wouldn't remove my current implementation as it allows for really quick testing. +- Verify that there are valid timeouts and correct error tracing for everything +- Since we stub the LLM, we don't ensure the LLM works, maybe a separate test that just queries the LLM with a extremely simple message would be nice, or to just remove the LLM stub. -## Serve +#### Future Improvements: +- Add LLM compatable hooks for an LLM to be able to quickly iterate and verify a build works through using test. Test should eventually be a fully verifier to ensure a workflow is valid -`canyonos serve` starts the local CanyonOS dashboard — it reads no project config, so it takes no -arguments. It writes only `CANYONOS_`-prefixed settings into the current directory's `.env`, -leaving every other line unchanged. +## canyonos build: +### INPUT: canyonos build +#### Steps: +1. Asks the user which coding agent they want to use for this [Codex/Claude] +2. Asks the user if they want to download the skill locally or globally (So the skill can be viewed either only in this directory or across your entire laptop) +3. Opens said coding agent, giving it instructions to build a new .car folder with the code (Nicks skill) + 4. Periodically the coding agent should ask the user config related questions (which provider, entrypoints, OTEL location) + 5. Coding agent should also be running canyonos test to verify workflow works +6. Finishes, doesn't run deploy itself. -If you have a workflow running, and want to make a config change, canyonos config automatically would reload the project with your config. If you change the workflow files itself though and want the changes to take effect, you need to redeploy from scratch, running canyonos build for good measure +#### Action Items: +- Coding agent should be using canyonos test to verify the file working, need to add that to skill file and harden canyonos test first. +- Maybe add more skills for it to deploy itself and monitor deployments so the user literally doesn't have to do anything else. -# Use: canyonos -h +#### Future Improvements: +- Add more agent providers (Cursor, Pi, Windsurf, etc...) +- Add a preconfigured config file that can get converted into global_controller.yaml (So provider can be autofilled as AWS/Azure/etc..) + + +## canyonos deploy: +### INPUT: canyonos deploy [optional: --serve True -verbose True] + +#### Steps: + +#### Action Items: + +#### Future Improvements diff --git a/cli/canyonos/constants.py b/cli/canyonos/constants.py index f1aa9d1..05ac30f 100644 --- a/cli/canyonos/constants.py +++ b/cli/canyonos/constants.py @@ -2,7 +2,9 @@ Config/data layer. Holds shared values and parsing helpers""" +import ast import os +import socket import yaml from ruamel.yaml import YAML @@ -10,10 +12,10 @@ DEFAULT_API_PORT = 8080 DEFAULT_DASHBOARD_PORT = 8081 -# The workflow entrypoint is always exposed as POST /main with a {"query": ...} -# body, regardless of what the workflow function is called in the project. -# This should be fixed later, keeping it like this for now though +# Fallback when the real function name/params can't be determined statically +# (see workflow_entrypoint) -- canyonos_core's own examples all follow this shape. WORKFLOW_ROUTE = "main" +DEFAULT_QUERY_PARAM = "query" def default_config_path(): @@ -36,6 +38,84 @@ def workflow_api_port(config_path): return None +def _source_root(config_path): + """Directory `workflow_file` is relative to -- `.car/app` under the .car layout, else the project root.""" + car_root = os.path.dirname(os.path.dirname(config_path)) or "." + return os.path.join(car_root, "app") if os.path.basename(car_root) == ".car" else car_root + + +def _deploy_call_target(tree): + """The name passed as `deploy(, ...)`'s first argument, or None.""" + for node in ast.walk(tree): + is_deploy_call = ( + isinstance(node, ast.Call) + and isinstance(node.func, (ast.Name, ast.Attribute)) + and (node.func.id if isinstance(node.func, ast.Name) else node.func.attr) == "deploy" + ) + if is_deploy_call and node.args and isinstance(node.args[0], ast.Name): + return node.args[0].id + return None + + +def workflow_entrypoint(config_path): + """(route, [(param_name, example_default_or_None), ...]) read statically from the + workflow's own source -- the function `deploy()` is actually called with, not an + assumed name. Returns None if the file, the deploy() call, or the function can't be found.""" + try: + with open(config_path) as f: + config = yaml.safe_load(f) or {} + except (OSError, yaml.YAMLError): + return None + + workflow_file = next( + (a.get("workflow_file") for a in config.get("agents") or [] if a.get("type") == "workflow"), + None, + ) + if not workflow_file: + return None + + workflow_path = os.path.join(_source_root(config_path), workflow_file) + try: + with open(workflow_path) as f: + tree = ast.parse(f.read(), filename=workflow_path) + except (OSError, SyntaxError): + return None + + fn_name = _deploy_call_target(tree) + if fn_name is None: + return None + + fn_def = next( + (n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == fn_name), + None, + ) + if fn_def is None: + return None + + args = [a.arg for a in fn_def.args.args if a.arg != "self"] + defaults = fn_def.args.defaults + first_defaulted = len(args) - len(defaults) + params = [] + for i, name in enumerate(args): + default = None + if i >= first_defaulted: + try: + default = ast.literal_eval(defaults[i - first_defaulted]) + except (ValueError, TypeError): + default = None + params.append((name, default)) + return fn_name, params + + +def port_in_use(port): + """True if something is listening on this host port already.""" + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.5): + return True + except OSError: + return False + + def dashboard_port(config_path): """Host port the local dashboard prefers to start on, falling back to the default.""" try: diff --git a/cli/canyonos/dashboard.compose.yml b/cli/canyonos/dashboard.compose.yml index e6691ab..38c3c99 100644 --- a/cli/canyonos/dashboard.compose.yml +++ b/cli/canyonos/dashboard.compose.yml @@ -1,21 +1,23 @@ services: - postgres: - image: postgres:17-alpine + # Fast-path bundled DB, hardcoded creds -- fine for local dev, not for anything real. + db: + image: postgres:16-alpine environment: - POSTGRES_DB: canyonos POSTGRES_USER: canyonos POSTGRES_PASSWORD: canyonos - volumes: - - postgres-data:/var/lib/postgresql/data + POSTGRES_DB: canyonos healthcheck: - test: ["CMD-SHELL", "pg_isready -U canyonos -d canyonos"] - interval: 3s + test: ["CMD-SHELL", "pg_isready -U canyonos"] + interval: 2s timeout: 3s retries: 20 + ports: + - "127.0.0.1:5432:5432" + api: image: ${CANYONOS_API_IMAGE} depends_on: - postgres: + db: condition: service_healthy # Published on all interfaces (not just 127.0.0.1) so a GC container can # actually reach this via host.docker.internal -- Docker's host-gateway @@ -26,7 +28,7 @@ services: ports: - "3000:3000" environment: - DATABASE_URL: postgres://canyonos:canyonos@postgres:5432/canyonos + DATABASE_URL: postgresql://canyonos:canyonos@db:5432/canyonos JWT_SECRET: ${CANYONOS_JWT_SECRET} CANYONOS_DISABLE_AUTH: "true" CANYONOS_REDIS_HOST: ${CANYONOS_REDIS_HOST} @@ -41,5 +43,3 @@ services: condition: service_healthy ports: - "127.0.0.1:${CANYONOS_WEB_PORT}:8080" -volumes: - postgres-data: diff --git a/cli/canyonos/dashboard_stack.py b/cli/canyonos/dashboard_stack.py index 38e38a2..1f307a6 100644 --- a/cli/canyonos/dashboard_stack.py +++ b/cli/canyonos/dashboard_stack.py @@ -356,6 +356,31 @@ def _cleanup(stack: DashboardStack, manifest: Path) -> None: return +def _dashboard_compose_command(*args: str) -> bool: + """Run a `docker compose` subcommand against the dashboard stack from the current project. + + False (no-op) if the dashboard was never started from here -- there's no + `.env` for `--env-file` to point at, so there's nothing to stop/tear down. + """ + stack = DashboardStack(state_dir=_state_dir(), project_dir=Path.cwd()) + if not stack.env_path.is_file(): + return False + manifest_resource = importlib.resources.files("canyonos").joinpath("dashboard.compose.yml") + with importlib.resources.as_file(manifest_resource) as manifest: + result = _run([*_compose_argv(stack, manifest), *args]) + return result.returncode == 0 + + +def stop_dashboard() -> bool: + """`docker compose stop` -- halts web/api/db, keeping them for a later `canyonos serve`.""" + return _dashboard_compose_command("stop") + + +def teardown_dashboard() -> bool: + """`docker compose down` -- removes the dashboard's web/api/db containers entirely.""" + return _dashboard_compose_command("down") + + def run_dashboard( phase_reporter: Callable[[str, str], None] | None = None, preferred_port: int = DEFAULT_DASHBOARD_PORT, diff --git a/cli/canyonos/deploy.py b/cli/canyonos/deploy.py index 8cddd81..8e61e27 100644 --- a/cli/canyonos/deploy.py +++ b/cli/canyonos/deploy.py @@ -15,6 +15,7 @@ manual step. """ +import json import queue import re import subprocess @@ -27,9 +28,12 @@ from canyonos import ui from canyonos.constants import ( + DEFAULT_QUERY_PARAM, WORKFLOW_ROUTE, default_config_path, + port_in_use, workflow_api_port, + workflow_entrypoint, workspace_relative, ) from canyonos.gc import GCError, deploy_status, post_deploy, workflow_endpoints @@ -133,30 +137,53 @@ def agents_ready_message(self): return f"{ready} agent(s) ready", True -def run_deploy(config_path=None, serve=True, verbose=False): +def run_deploy(config_path=None, serve=True, verbose=False, quiet=False, extra_env=None, banner=True): + """`quiet` skips the log-tail/dashboard UI and returns the GC state right + after the deploy is triggered -- for a caller (`canyonos test`) that wants + its own readiness check instead of this command's own output. + """ # Left as None when unset: canyonos resolves the artifact layout itself. if config_path is not None: config_path = workspace_relative(config_path) if config_path is None: - ui.fail("Config must be inside the project directory being synced.") - return + raise RuntimeError("Config must be inside the project directory being synced.") - run_init() + run_init(banner=banner, extra_env=extra_env) # Copy the current project into the container before building/deploying. if not run_sync(): - return + raise RuntimeError("Could not sync the project into the container.") state = load_state() # Read for display only -- canyonos resolves the path it actually deploys. api_port = workflow_api_port(config_path or default_config_path()) + # Checked here, after run_init() has already torn down any previous deploy, + # so a still-live prior run doesn't read as an unrelated conflict. + if api_port is not None and port_in_use(api_port): + raise RuntimeError( + f"Port {api_port} is already in use, and the workflow needs it. Free it " + f"or change `api_port` in {config_path or default_config_path()}." + ) + try: post_deploy(state["port"], config_path) - _stream_logs_and_autoserve(state, api_port, serve=serve, verbose=verbose) except GCError as e: - ui.fail(e) + raise RuntimeError(str(e)) from None + + if quiet: + # Still bring the dashboard up so anything reachable only through its + # LLM proxy (e.g. a guardrail calling the OpenAI SDK directly) works + # under `canyonos test` too -- just skip the log-tail/summary UI. + if serve: + _start_dashboard() + return state + + _stream_logs_and_autoserve( + state, api_port, config_path or default_config_path(), serve=serve, verbose=verbose + ) + return state def workflow_targets(gc_port, api_port): @@ -180,7 +207,27 @@ def workflow_targets(gc_port, api_port): return [(None, "127.0.0.1", api_port)] if api_port else [] -def _summary_body(dashboard_url, targets): +def _example_route_and_body(config_path): + """(route, body dict) for the curl example -- read from the workflow function's + own signature when possible, falling back to the historical `main`/`query` shape.""" + entrypoint = workflow_entrypoint(config_path) + if not entrypoint: + return WORKFLOW_ROUTE, {DEFAULT_QUERY_PARAM: "your question here"} + + fn_name, params = entrypoint + if not params: + return fn_name, {DEFAULT_QUERY_PARAM: "your question here"} + return fn_name, {name: default if default is not None else "" for name, default in params} + + +def _curl_example(url, body): + """A copy-pasteable `curl -X POST ...` block, indented to sit under the summary's other rows.""" + json_lines = json.dumps(body, indent=2).splitlines() + indented_body = "\n".join(line if i == 0 else f" {line}" for i, line in enumerate(json_lines)) + return f'curl -X POST {url} \\\n -H "Content-Type: application/json" \\\n -d \'{indented_body}\'' + + +def _summary_body(dashboard_url, targets, config_path): """ The contents that go inside the deploy panel""" body = Text() body.append("Dashboard ", "dim") @@ -189,15 +236,14 @@ def _summary_body(dashboard_url, targets): else: body.append("not running -- start it with `canyonos serve`", WHITE) + route, example = _example_route_and_body(config_path) for name, host, port in targets: base = f"http://{host}:{port}" body.append("\n") if name: body.append(f"\n{name}", f"bold {WHITE}") - body.append("\nPOST ", "dim") - body.append(f"{base}/{WORKFLOW_ROUTE}", f"bold {GREEN}") - body.append("\nbody ", "dim") - body.append('{"query": "your question here"}', WHITE) + body.append("\n") + body.append(_curl_example(f"{base}/{route}", example), WHITE) body.append("\npoll ", "dim") body.append(f"{base}/status/", WHITE) if host not in ("127.0.0.1", "localhost"): @@ -205,7 +251,7 @@ def _summary_body(dashboard_url, targets): return body -def print_deploy_summary(dashboard_url, targets): +def print_deploy_summary(dashboard_url, targets, config_path): """The one screen printed once everything is up: dashboard and workflow endpoints. Under `-v` it is printed again on exit, because the log tail continues @@ -215,7 +261,7 @@ def print_deploy_summary(dashboard_url, targets): ui.blank() ui.panel( Panel( - _summary_body(dashboard_url, targets), + _summary_body(dashboard_url, targets, config_path), title=f"[bold {GREEN}]Deploy is live[/]", title_align="left", border_style=GREEN, @@ -235,12 +281,14 @@ def _start_dashboard(): return None -def _deploy_summary(state, api_port, serve): +def _deploy_summary(state, api_port, config_path, serve): summary = ( _start_dashboard() if serve else None, workflow_targets(state["port"], api_port), + config_path, ) print_deploy_summary(*summary) + ui.hint("Tailing logs now, press Ctrl+C to stop. Run `canyonos stop` to stop the workflow.") return summary @@ -252,7 +300,7 @@ def _interrupted(summary=None): print_deploy_summary(*summary) -def _tail_verbose(stream, state, api_port, serve): +def _tail_verbose(stream, state, api_port, config_path, serve): """Every log line, verbatim -- what `-v` restores. Ctrl+C reprints the summary here but not in quiet mode: only this tail keeps @@ -264,12 +312,12 @@ def _tail_verbose(stream, state, api_port, serve): print(line, end="") # Logged exactly once, right after the workflow finishes coming up. if summary is None and "Global controller started, polling every" in line: - summary = _deploy_summary(state, api_port, serve) + summary = _deploy_summary(state, api_port, config_path, serve) except KeyboardInterrupt: _interrupted(summary) -def _tail_quiet(lines, state, api_port, serve): +def _tail_quiet(lines, state, api_port, config_path, serve): """Only the phase transitions, until the workflow is up or something fails. Nothing is echoed raw: the buildx transcript, canyonos' bare prints and grpc's @@ -303,7 +351,7 @@ def _tail_quiet(lines, state, api_port, serve): break if reached_up_marker: - return _deploy_summary(state, api_port, serve) + return _deploy_summary(state, api_port, config_path, serve) _reveal_failure(lines, recent, state) return None @@ -384,7 +432,7 @@ def _reveal_failure(lines, recent, state): ui.hint("Run `canyonos deploy -v` or `canyonos logs` for the full container log.") -def _stream_logs_and_autoserve(state, api_port, serve=True, verbose=False): +def _stream_logs_and_autoserve(state, api_port, config_path, serve=True, verbose=False): """Tail the GC container's logs, and once they show the workflow is up, start the dashboard (unless disabled via `serve=False`) and print where everything lives. Log tailing continues afterwards. @@ -398,10 +446,10 @@ def _stream_logs_and_autoserve(state, api_port, serve=True, verbose=False): ) try: if verbose: - _tail_verbose(process.stdout, state, api_port, serve) + _tail_verbose(process.stdout, state, api_port, config_path, serve) return lines = _queued_lines(process.stdout) - if _tail_quiet(lines, state, api_port, serve) is not None: + if _tail_quiet(lines, state, api_port, config_path, serve) is not None: # Quiet mode stays attached after the summary so Ctrl+C means the # same thing in both modes -- it just swallows what arrives. while lines.get() is not None: diff --git a/cli/canyonos/init.py b/cli/canyonos/init.py index eac2edf..5b3823d 100644 --- a/cli/canyonos/init.py +++ b/cli/canyonos/init.py @@ -24,6 +24,12 @@ # Image Name, need to switch to CanyonCore Organization Namespace later GC_IMAGE = "saakeths/canyonos:latest" GC_CONTAINER_PORT = 8000 +GC_CONTAINER_NAME = "canyonos-global-controller" + +# Same network canyonos_core's own GlobalController creates for local-provider +# Redis/agent containers -- the GC container needs to be on it too, e.g. to +# resolve :50051 for its own cleanup gRPC calls. +LOCAL_NETWORK = "canyonos-local" # Named docker volume mounted at /workspace inside the container. Files are # copied in via `canyonos sync` (docker cp), not mounted live, so host-side @@ -129,13 +135,20 @@ def _port_reachable(port, attempts=10, delay=0.5): def run_container(image=GC_IMAGE, max_attempts=50, extra_env=None): port = GC_CONTAINER_PORT + # Idempotent: succeeds silently if the network already exists (created by + # this or a prior GC/Redis launch). + subprocess.run(["docker", "network", "create", LOCAL_NETWORK], capture_output=True) for _ in range(max_attempts): cmd = [ "docker", "run", "-d", + "--name", + GC_CONTAINER_NAME, "-p", f"127.0.0.1:{port}:{GC_CONTAINER_PORT}", + "--network", + LOCAL_NETWORK, # Docker-outside-of-Docker: GC shells out to `docker` to launch # Redis/agent containers, so it needs the host's real daemon, # not a nested one. diff --git a/cli/canyonos/quit.py b/cli/canyonos/quit.py index 9af5dcc..f7ee16c 100644 --- a/cli/canyonos/quit.py +++ b/cli/canyonos/quit.py @@ -1,14 +1,16 @@ """ Logic for `canyonos quit`: full teardown. Stops and removes the Global Controller container AND deletes the /workspace named volume, so the project -files copied into it are discarded too. (Use `canyonos stop` to only halt a -running deploy while keeping the container and files around.) +files copied into it are discarded too. Also tears down the local dashboard +stack (web/api/db), if one was started from this project. (Use `canyonos stop` +to only halt a running deploy while keeping the containers and files around.) """ import os import subprocess from canyonos import ui +from canyonos.dashboard_stack import teardown_dashboard from canyonos.gc import GCError, post_clean, require_state from canyonos.init import GC_WORKSPACE_VOLUME, STATE_PATH @@ -49,6 +51,7 @@ def run_quit(): # refuses to remove a volume still in use). check=False so a missing # volume doesn't turn teardown into an error. subprocess.run(["docker", "volume", "rm", GC_WORKSPACE_VOLUME], check=False, capture_output=True) + teardown_dashboard() os.remove(STATE_PATH) if already_gone: diff --git a/cli/canyonos/stop.py b/cli/canyonos/stop.py index 519cf49..204c3b2 100644 --- a/cli/canyonos/stop.py +++ b/cli/canyonos/stop.py @@ -4,6 +4,7 @@ """ from canyonos import ui +from canyonos.dashboard_stack import stop_dashboard from canyonos.gc import GCError, post_clean, require_state @@ -15,6 +16,7 @@ def run_stop(): try: with ui.status("Stopping deploy..."): post_clean(state["port"]) + stop_dashboard() ui.ok("Deploy stopped.") except GCError as e: ui.fail(e) diff --git a/cli/canyonos/test.py b/cli/canyonos/test.py index 895e8e0..47e589f 100644 --- a/cli/canyonos/test.py +++ b/cli/canyonos/test.py @@ -1,8 +1,7 @@ """ Logic for `canyonos test`: check a project end to end on this machine. -Four phases, each ending the run if it fails: -- The `.car/` artifact `canyonos build` produced is verified statically +Three phases, each ending the run if it fails: - The project is deployed locally (every agent's `provider` rewritten to `local` for the duration, the original file restored verbatim afterwards) - The running containers are checked against what the config declared - One prompt is sent to the workflow's `/main` endpoint. @@ -15,7 +14,6 @@ import json import os -import socket import subprocess import time import urllib.error @@ -32,12 +30,11 @@ workflow_api_port, workspace_relative, ) -from canyonos.deploy import workflow_targets -from canyonos.gc import GCError, deploy_status, post_deploy -from canyonos.init import load_state, quit_existing, run_init -from canyonos.sync import run_sync +from canyonos.deploy import run_deploy, workflow_targets +from canyonos.gc import _DEPLOY_CONFLICT, deploy_status +from canyonos.init import load_state, quit_existing from canyonos.theme import GREEN, WHITE -from canyonos.verify import ARTIFACT_DIR, verify_build_artifact, verify_runtime +from canyonos.verify import verify_runtime DEFAULT_QUERY = "hello" # `canyonos test` stubs the in-container LLM proxy by default so a smoke test @@ -68,14 +65,6 @@ def _force_local_providers(config_path): return original -def _port_in_use(port): - try: - with socket.create_connection(("127.0.0.1", port), timeout=0.5): - return True - except OSError: - return False - - def _workflow_ready(host, port): """True once the workflow's REST API answers at all. @@ -150,7 +139,6 @@ def __init__(self, query): # log worth reading; before that it holds nothing about the failure. self.deploy_started = False self.phases = [] - self.validation = None self.runtime = None self.endpoint = None self.result = None @@ -161,7 +149,7 @@ def begin(self, name, number, title): """Open a phase, recorded as failed until `done` says otherwise.""" self.phases.append({"name": name, "ok": False, "detail": None}) ui.blank() - ui.say(f"[{number}/4] {title}") + ui.say(f"[{number}/3] {title}") def done(self, detail=None): self.phases[-1].update(ok=True, detail=detail) @@ -174,49 +162,19 @@ def elapsed(self): return round(time.monotonic() - self.started, 3) -def _verify_build(run, config_path): - run.begin("verify_build", 1, "Verify build artifact") - - # A project ported before the .car layout keeps its config at the top level; - # there is no build artifact to check, so the deploy phases still run. - if not config_path.startswith(f"{ARTIFACT_DIR}{os.sep}"): - ui.warn(f"No `{ARTIFACT_DIR}/` artifact -- deploying {config_path} as it is.") - ui.hint(" -> `canyonos build` produces one, and gives this phase something to check.") - run.done("skipped: no .car/ artifact") - return - - run.validation = verify_build_artifact() - stale = len(run.validation["stale"]) - run.done(f"{run.validation['warnings']} warning(s), {stale} stale source(s)") - - def _deploy_locally(run, config_path, api_port, llm_stub=DEFAULT_LLM_STUB): - run.begin("deploy", 2, "Deploy locally") + run.begin("deploy", 1, "Deploy locally") # When stubbing, hand the flag to the GC container; the local runtime # forwards it into every agent so their LLM calls are replaced with canned # text (see canyonos_core/llm_proxy/stub.py). extra_env = {"CANYONOS_LLM_STUB_TEXT": llm_stub} if llm_stub else None if llm_stub: ui.say(f"LLM stub on: every model call returns {llm_stub!r} (no real LLM). Pass --real-llm to disable.") - run_init(banner=False, extra_env=extra_env) - - if not run_sync(): - raise RuntimeError("Could not sync the project into the container.") - - # Only the gRPC host port is bumped when a port is taken (the local runtime's - # launch retry), so an occupied api_port dies 50 attempts later as "no free - # port found". `canyonos serve` also starts looking for its web port at 8080. - if _port_in_use(api_port): - raise RuntimeError( - f"Port {api_port} is already in use, and the workflow needs it. Free it " - f"(`canyonos quit` stops a previous deploy) or change `api_port` in {config_path}." - ) - state = load_state() - try: - post_deploy(state["port"], config_path) - except GCError as e: - raise RuntimeError(str(e)) from None + # quiet=True: skip `canyonos deploy`'s own log-tail/summary UI, we do our + # own HTTP readiness check below instead. serve=True still brings the + # dashboard's LLM proxy up, quietly, for code that calls it directly. + state = run_deploy(config_path, serve=True, quiet=True, extra_env=extra_env, banner=False) run.deploy_started = True _wait_for_workflow(state["port"], api_port) @@ -225,13 +183,13 @@ def _deploy_locally(run, config_path, api_port, llm_stub=DEFAULT_LLM_STUB): def _verify_runtime(run, config_path, gc_port): - run.begin("verify_runtime", 3, "Verify runtime") + run.begin("verify_runtime", 2, "Verify runtime") run.runtime = verify_runtime(config_path, gc_port) run.done(f"{len(run.runtime['agents'])} agent(s) up") def _query(run, gc_port, api_port): - run.begin("query", 4, "Query the workflow") + run.begin("query", 3, "Query the workflow") targets = workflow_targets(gc_port, api_port) if not targets: raise RuntimeError("The deploy reported no workflow endpoint to query.") @@ -256,16 +214,29 @@ def _query(run, gc_port, api_port): run.done(f"answered in {run.elapsed()}s") +def _refuse_if_deploy_running(): + """Bail out before touching anything if a deploy is already up -- otherwise + `_deploy_locally` would tear it down via `run_init`'s own cleanup only to + fail later for an unrelated reason. + """ + try: + state = load_state() + except FileNotFoundError: + return + if (deploy_status(state["port"]) or {}).get("running", False): + raise RuntimeError(_DEPLOY_CONFLICT) + + def _run_test(run, llm_stub=DEFAULT_LLM_STUB): - """Walk the four phases, restoring the config whatever happens.""" + """Walk the three phases, restoring the config whatever happens.""" + _refuse_if_deploy_running() + config_path = workspace_relative(default_config_path()) if config_path is None: raise RuntimeError("Config must be inside the project directory being synced.") if not os.path.isfile(config_path): raise RuntimeError(f"No config at {config_path}. Run `canyonos build` first.") - _verify_build(run, config_path) - api_port = workflow_api_port(config_path) if api_port is None: raise RuntimeError(f"No agent with `type: workflow` in {config_path}; nothing to test.") @@ -327,6 +298,38 @@ def _print_summary(run): ui.blank() +def _readable_result(result): + """`result` unwrapped to its plain value when it's just one field -- the + common case (e.g. `{"reply": "..."}`) reads far better than raw JSON. + """ + if isinstance(result, dict) and len(result) == 1: + value = next(iter(result.values())) + if isinstance(value, str): + return value + return json.dumps(result, indent=2) + + +def _print_io(run): + """A short, scannable input/output pair -- the main panel's own Result field + is the full raw JSON, which gets unreadable fast for a nested result. + """ + body = Text() + body.append("Input ", "dim") + body.append(run.query, WHITE) + body.append("\nOutput ", "dim") + body.append(_readable_result(run.result), WHITE) + ui.panel( + Panel( + body, + title=f"[bold {GREEN}]Input / Output[/]", + title_align="left", + border_style=GREEN, + padding=(1, 4), + ) + ) + ui.blank() + + def _print_failure_logs(run): if run.log_tail: ui.hint(f"last {LOG_TAIL_LINES} lines of the Global Controller log:") @@ -341,7 +344,6 @@ def _payload(run): "query": run.query, "elapsed_s": run.elapsed(), "phases": run.phases, - "validation": run.validation, "runtime": run.runtime, "result": run.result, "error": run.error, @@ -376,13 +378,18 @@ def run_test(prompt=None, as_json=False, llm_stub=DEFAULT_LLM_STUB): container_live = True except (FileNotFoundError, OSError): pass - else: + elif run.error is None: quit_existing() + # else: failed before this run ever started its own deploy (e.g. bad + # config, or `_refuse_if_deploy_running` above) -- nothing of ours to + # clean up, so leave whatever was already there alone. if as_json: print(json.dumps(_payload(run), indent=2)) else: _print_summary(run) + if run.error is None: + _print_io(run) if container_live: _print_failure_logs(run) diff --git a/cli/canyonos/theme.py b/cli/canyonos/theme.py index e74a069..cd2a44e 100644 --- a/cli/canyonos/theme.py +++ b/cli/canyonos/theme.py @@ -4,17 +4,38 @@ The green->white gradient introduced by the `canyonos init` banner, reused across the CLI so everything shares one look. `GREEN` is the primary brand color; `WHITE` the secondary; `GRADIENT` the full ramp for multi-line output. +Both flip to a dark-on-light variant when the terminal background is light. """ +import os +import re +import select +import sys +import termios +import tty + GREEN = "#2BD17E" -WHITE = "#FFFFFF" + +_GRADIENT_DARK = ["#2BD17E", "#55DA98", "#80E3B2", "#AAEDCB", "#D5F6E5", "#FFFFFF"] +_GRADIENT_LIGHT = ["#2BD17E", "#1F9C61", "#177249", "#0F4E31", "#082A1A", "#000000"] + + +def _is_light_background(): + if not sys.stdin.isatty(): + return False + fd = sys.stdin.fileno() + old = termios.tcgetattr(fd) + try: + tty.setraw(fd) + sys.stdout.write("\x1b]11;?\x07") + sys.stdout.flush() + reply = os.read(fd, 32).decode(errors="ignore") if select.select([fd], [], [], 0.1)[0] else "" + finally: + termios.tcsetattr(fd, termios.TCSADRAIN, old) + m = re.search(r"rgb:([0-9a-f]{2})\S*/([0-9a-f]{2})\S*/([0-9a-f]{2})", reply, re.I) + return bool(m) and (0.299 * int(m[1], 16) + 0.587 * int(m[2], 16) + 0.114 * int(m[3], 16)) > 128 + # Primary -> secondary ramp (used for the init banner, top to bottom). -GRADIENT = [ - "#2BD17E", - "#55DA98", - "#80E3B2", - "#AAEDCB", - "#D5F6E5", - "#FFFFFF", -] +GRADIENT = _GRADIENT_LIGHT if _is_light_background() else _GRADIENT_DARK +WHITE = GRADIENT[-1] diff --git a/cli/canyonos/verify.py b/cli/canyonos/verify.py index 4f9a46a..6d5be36 100644 --- a/cli/canyonos/verify.py +++ b/cli/canyonos/verify.py @@ -1,9 +1,5 @@ """ -The two verification passes behind `canyonos test`. - -`verify_build_artifact` checks the `.car/` tree a `canyonos build` produced, -before any container is started: the layout, the porting skill's own validator, -and whether the sources have moved on since the port was taken. +The verification pass behind `canyonos test`. `verify_runtime` checks a running local deploy against what the config declared -- every image built, every replica up -- because the controller logs a warning @@ -13,190 +9,18 @@ This file will also need lots of iteration based on what is needed, will expect it to change alot """ -import hashlib -import json -import os import subprocess -import sys import yaml from rich.table import Table from canyonos import gc, ui -from canyonos.build import AGENTS, install_skill from canyonos.constants import DEFAULT_API_PORT -from canyonos.init import STATE_DIR from canyonos.theme import GREEN -ARTIFACT_DIR = ".car" -SOURCE_DIR = "app" -CONFIG_REL = "config/global_controller.yaml" -PORTING_STATE_REL = "config/.porting-state.json" - -VALIDATOR_NAME = "validate.py" -SKILL_CACHE_DIR = os.path.join(STATE_DIR, "skill") - -# These two rules decide their verdict by importing `canyonos` and probing it for -# env-file injection and editable-install support. The runtime lives in the -# Global Controller image, not on the host running this CLI, so the probe always -# comes back empty here and the rules report a failure that isn't one. -CAPABILITY_GATED_CHECKS = frozenset({"V030", "V031"}) - RUNTIME_PREFIX = "canyonos-local-" -# ------------------------------------------------------------------ # -# Build artifact # -# ------------------------------------------------------------------ # - - -def _find_validator(project_root): - """Path to the porting skill's validate.py, fetching the skill if needed.""" - for spec in AGENTS.values(): - for skill_dir in spec["skill_dirs"].values(): - if not os.path.isabs(skill_dir): - skill_dir = os.path.join(project_root, skill_dir) - candidate = os.path.join(skill_dir, VALIDATOR_NAME) - if os.path.isfile(candidate): - return candidate - - cached = os.path.join(SKILL_CACHE_DIR, VALIDATOR_NAME) - if os.path.isfile(cached): - return cached - if install_skill(SKILL_CACHE_DIR) and os.path.isfile(cached): - return cached - return None - - -def _run_validator(validator, artifact_dir): - """The validator's parsed --json report, or None if it produced no report.""" - result = subprocess.run( - [sys.executable, validator, artifact_dir, "-c", CONFIG_REL, "--json"], - capture_output=True, - text=True, - ) - try: - return json.loads(result.stdout) - except ValueError: - detail = (result.stderr or result.stdout).strip().splitlines() - ui.warn(f" The porting validator did not run: {detail[-1] if detail else 'no output'}") - return None - - -def _drop_unprobeable(report): - """Remove the rules that can only be judged with `canyonos` importable. - - Their verdict without it is not merely uncertain, it is wrong: V030 reports - that the runtime never reads `env_file` when the container's runtime does. - """ - if report.get("capabilities", {}).get("canyonos_core"): - return 0 - - kept = [] - dropped = 0 - for finding in report.get("findings") or []: - if finding["check"] in CAPABILITY_GATED_CHECKS: - if finding["level"] == "ERROR": - report["errors"] = max(report.get("errors", 0) - 1, 0) - elif finding["level"] == "WARN": - report["warnings"] = max(report.get("warnings", 0) - 1, 0) - dropped += 1 - continue - kept.append(finding) - report["findings"] = kept - return dropped - - -_LEVEL_EMITTER = {"ERROR": ui.fail, "WARN": ui.warn} - - -def _report_findings(findings): - for finding in sorted(findings, key=lambda f: (f["level"] != "ERROR", f["check"])): - where = finding.get("path") or "" - if where and finding.get("line"): - where = f"{where}:{finding['line']}" - parts = [finding["check"], where, finding["summary"]] - line = " ".join(part for part in parts if part) - _LEVEL_EMITTER.get(finding["level"], ui.hint)(f" {line}") - - -def _sha256(path): - digest = hashlib.sha256() - with open(path, "rb") as f: - for block in iter(lambda: f.read(65536), b""): - digest.update(block) - return digest.hexdigest() - - -def _stale_sources(project_root, artifact_dir): - """Recorded sources that changed or vanished since the port was taken.""" - try: - with open(os.path.join(artifact_dir, PORTING_STATE_REL)) as f: - state = json.load(f) - except (OSError, ValueError): - return [] - - stale = [] - for relative, expected in (state.get("source_files") or {}).items(): - # The skill's own files are recorded alongside the project's; a newer - # skill would otherwise read as the application having changed. - if relative.startswith(".claude/"): - continue - path = os.path.join(project_root, relative) - if not os.path.isfile(path) or _sha256(path) != expected: - stale.append(relative) - return sorted(stale) - - -def verify_build_artifact(project_root="."): - """Check the `.car/` tree. Raises RuntimeError if it can't be deployed.""" - artifact_dir = os.path.join(project_root, ARTIFACT_DIR) - config_path = os.path.join(artifact_dir, CONFIG_REL) - - if not os.path.isfile(config_path) or not os.path.isdir( - os.path.join(artifact_dir, SOURCE_DIR) - ): - raise RuntimeError( - f"No `{ARTIFACT_DIR}/` artifact here (expected {CONFIG_REL} beside " - f"{SOURCE_DIR}/). Run `canyonos build` first." - ) - ui.ok(f"{ARTIFACT_DIR}/ layout (config/ + {SOURCE_DIR}/)") - - summary = {"errors": 0, "warnings": 0, "findings": [], "stale": []} - - validator = _find_validator(project_root) - if validator is None: - ui.warn("Could not fetch the porting validator; skipping artifact checks.") - ui.hint(" The deploy below still runs -- `canyonos doctor` checks the fetch path.") - else: - report = _run_validator(validator, os.path.abspath(artifact_dir)) - if report is not None: - skipped = _drop_unprobeable(report) - summary.update( - errors=report.get("errors", 0), - warnings=report.get("warnings", 0), - findings=report.get("findings", []), - ) - counts = f"{summary['errors']} error(s), {summary['warnings']} warning(s)" - (ui.fail if summary["errors"] else ui.ok)(f"porting validator: {counts}") - _report_findings(summary["findings"]) - if skipped: - ui.hint(f" {skipped} rule(s) need the canyonos runtime to judge and were skipped") - - summary["stale"] = _stale_sources(project_root, artifact_dir) - for relative in summary["stale"]: - ui.warn(f" source changed since the port: {relative}") - if summary["stale"]: - ui.hint(" -> re-run `canyonos build` to bring the artifact back in step") - - if summary["errors"]: - raise RuntimeError( - f"The build artifact has {summary['errors']} validation error(s); fix them " - "or re-run `canyonos build`." - ) - return summary - - # ------------------------------------------------------------------ # # Runtime # # ------------------------------------------------------------------ # diff --git a/tests/test_canyonos_deploy.py b/tests/test_canyonos_deploy.py new file mode 100644 index 0000000..ac62c88 --- /dev/null +++ b/tests/test_canyonos_deploy.py @@ -0,0 +1,90 @@ +import pytest + +from canyonos import deploy as deploy_cmd +from canyonos.gc import GCError + +CONFIG_PATH = "config/global_controller.yaml" +STATE = {"container_id": "abc", "port": 8000} + + +@pytest.fixture +def deployable(monkeypatch): + """Every step run_deploy drives succeeds unless overridden.""" + monkeypatch.setattr(deploy_cmd, "workspace_relative", lambda p: p) + monkeypatch.setattr(deploy_cmd, "run_init", lambda banner=True, extra_env=None: None) + monkeypatch.setattr(deploy_cmd, "run_sync", lambda: True) + monkeypatch.setattr(deploy_cmd, "load_state", lambda: dict(STATE)) + monkeypatch.setattr(deploy_cmd, "workflow_api_port", lambda _config: 8080) + monkeypatch.setattr(deploy_cmd, "port_in_use", lambda _port: False) + monkeypatch.setattr(deploy_cmd, "post_deploy", lambda *_a: None) + + +def test_a_config_path_outside_the_project_raises(monkeypatch, deployable): + monkeypatch.setattr(deploy_cmd, "workspace_relative", lambda _p: None) + + with pytest.raises(RuntimeError, match="Config must be inside the project directory"): + deploy_cmd.run_deploy(CONFIG_PATH, quiet=True) + + +def test_a_sync_failure_raises(monkeypatch, deployable): + monkeypatch.setattr(deploy_cmd, "run_sync", lambda: False) + + with pytest.raises(RuntimeError, match="Could not sync the project"): + deploy_cmd.run_deploy(CONFIG_PATH, quiet=True) + + +def test_an_occupied_api_port_raises_before_post_deploy(monkeypatch, deployable): + """Checked after run_init() (which already tore down any previous deploy), so a still-live + prior run doesn't read as an unrelated conflict -- only a genuinely occupied port does.""" + calls = [] + monkeypatch.setattr(deploy_cmd, "port_in_use", lambda _port: True) + monkeypatch.setattr(deploy_cmd, "post_deploy", lambda *a: calls.append(a)) + + with pytest.raises(RuntimeError, match="Port 8080 is already in use"): + deploy_cmd.run_deploy(CONFIG_PATH, quiet=True) + + assert calls == [] + + +def test_a_post_deploy_failure_is_reraised_as_a_runtime_error(monkeypatch, deployable): + def boom(*_a): + raise GCError("Deploy failed: conflict") + + monkeypatch.setattr(deploy_cmd, "post_deploy", boom) + + with pytest.raises(RuntimeError, match="Deploy failed: conflict"): + deploy_cmd.run_deploy(CONFIG_PATH, quiet=True) + + +def test_quiet_returns_state_without_streaming(monkeypatch, deployable): + def unexpected(*_a, **_k): + raise AssertionError("quiet=True should skip the log-tail/dashboard UI") + + monkeypatch.setattr(deploy_cmd, "_stream_logs_and_autoserve", unexpected) + + assert deploy_cmd.run_deploy(CONFIG_PATH, quiet=True) == STATE + + +def test_non_quiet_still_streams_and_returns_state(monkeypatch, deployable): + calls = [] + monkeypatch.setattr( + deploy_cmd, "_stream_logs_and_autoserve", + lambda state, api_port, config_path, serve, verbose: calls.append( + (state, api_port, config_path, serve, verbose) + ), + ) + + assert deploy_cmd.run_deploy(CONFIG_PATH, serve=False, verbose=True) == STATE + assert calls == [(STATE, 8080, CONFIG_PATH, False, True)] + + +def test_extra_env_and_banner_are_forwarded_to_run_init(monkeypatch, deployable): + seen = {} + monkeypatch.setattr( + deploy_cmd, "run_init", + lambda banner=True, extra_env=None: seen.update(banner=banner, extra_env=extra_env), + ) + + deploy_cmd.run_deploy(CONFIG_PATH, quiet=True, extra_env={"CANYONOS_LLM_STUB_TEXT": "test"}, banner=False) + + assert seen == {"banner": False, "extra_env": {"CANYONOS_LLM_STUB_TEXT": "test"}} diff --git a/tests/test_canyonos_test.py b/tests/test_canyonos_test.py index 21e0918..e88aa12 100644 --- a/tests/test_canyonos_test.py +++ b/tests/test_canyonos_test.py @@ -1,4 +1,3 @@ -import hashlib import json import subprocess @@ -41,190 +40,6 @@ def project(monkeypatch, tmp_path): return tmp_path -def report(errors=0, warnings=0, findings=(), canyonos=False): - return { - "capabilities": {"canyonos_core": canyonos}, - "errors": errors, - "warnings": warnings, - "findings": list(findings), - } - - -def finding(check, level="ERROR"): - return {"check": check, "level": level, "path": "config/x.yaml", "line": 1, "summary": "s"} - - -# ------------------------------------------------------------------ # -# Locating the porting validator # -# ------------------------------------------------------------------ # - - -def test_validator_prefers_the_project_skill(monkeypatch, project, tmp_path): - codex = tmp_path / "codex-skill" - codex.mkdir() - (codex / "validate.py").write_text("") - local = project / ".claude" / "skills" / "porting-to-canyonos" - local.mkdir(parents=True) - (local / "validate.py").write_text("") - - monkeypatch.setattr( - verify, - "AGENTS", - { - "claude": {"skill_dirs": {"local": ".claude/skills/porting-to-canyonos"}}, - "codex": {"skill_dirs": {"global": str(codex)}}, - }, - ) - assert verify._find_validator(str(project)) == str(local / "validate.py") - - -def test_validator_falls_back_to_the_codex_skill(monkeypatch, project, tmp_path): - codex = tmp_path / "codex-skill" - codex.mkdir() - (codex / "validate.py").write_text("") - - monkeypatch.setattr( - verify, - "AGENTS", - { - "claude": {"skill_dirs": {"local": ".claude/skills/porting-to-canyonos"}}, - "codex": {"skill_dirs": {"global": str(codex)}}, - }, - ) - assert verify._find_validator(str(project)) == str(codex / "validate.py") - - -def test_validator_is_fetched_when_nothing_is_installed(monkeypatch, project, tmp_path): - cache = tmp_path / "cache" - monkeypatch.setattr(verify, "AGENTS", {}) - monkeypatch.setattr(verify, "SKILL_CACHE_DIR", str(cache)) - - def fake_install(dest): - assert dest == str(cache) - cache.mkdir() - (cache / "validate.py").write_text("") - return True - - monkeypatch.setattr(verify, "install_skill", fake_install) - assert verify._find_validator(str(project)) == str(cache / "validate.py") - - -def test_a_validator_that_cannot_be_fetched_does_not_stop_the_run(monkeypatch, project, tmp_path): - monkeypatch.setattr(verify, "AGENTS", {}) - monkeypatch.setattr(verify, "SKILL_CACHE_DIR", str(tmp_path / "empty-cache")) - monkeypatch.setattr(verify, "install_skill", lambda _dest: False) - - summary = verify.verify_build_artifact(str(project)) - - assert summary == {"errors": 0, "warnings": 0, "findings": [], "stale": []} - - -# ------------------------------------------------------------------ # -# Reading the validator's report # -# ------------------------------------------------------------------ # - - -def test_validator_errors_fail_the_phase(monkeypatch, project): - monkeypatch.setattr(verify, "_find_validator", lambda _root: "/validate.py") - monkeypatch.setattr( - verify, "_run_validator", lambda *_: report(errors=1, findings=[finding("V002")]) - ) - - with pytest.raises(RuntimeError): - verify.verify_build_artifact(str(project)) - - -def test_validator_warnings_pass(monkeypatch, project): - monkeypatch.setattr(verify, "_find_validator", lambda _root: "/validate.py") - monkeypatch.setattr( - verify, - "_run_validator", - lambda *_: report(warnings=1, findings=[finding("V018", "WARN")]), - ) - - summary = verify.verify_build_artifact(str(project)) - - assert (summary["errors"], summary["warnings"]) == (0, 1) - - -def test_rules_needing_canyonos_are_dropped_when_it_is_not_importable(monkeypatch, project): - monkeypatch.setattr(verify, "_find_validator", lambda _root: "/validate.py") - monkeypatch.setattr( - verify, - "_run_validator", - lambda *_: report(errors=1, findings=[finding("V030"), finding("V031", "INFO")]), - ) - - summary = verify.verify_build_artifact(str(project)) - - assert summary["errors"] == 0 - assert summary["findings"] == [] - - -def test_rules_needing_canyonos_are_kept_when_it_is_importable(monkeypatch, project): - monkeypatch.setattr(verify, "_find_validator", lambda _root: "/validate.py") - monkeypatch.setattr( - verify, - "_run_validator", - lambda *_: report(errors=1, findings=[finding("V030")], canyonos=True), - ) - - with pytest.raises(RuntimeError): - verify.verify_build_artifact(str(project)) - - -def test_a_missing_car_directory_is_an_error(monkeypatch, tmp_path): - monkeypatch.chdir(tmp_path) - with pytest.raises(RuntimeError, match="Run `canyonos build` first"): - verify.verify_build_artifact(str(tmp_path)) - - -# ------------------------------------------------------------------ # -# Source drift # -# ------------------------------------------------------------------ # - - -def write_porting_state(project, entries): - (project / ".car" / "config" / ".porting-state.json").write_text( - json.dumps({"version": 1, "source_files": entries}) - ) - - -def sha256(path): - return hashlib.sha256(path.read_bytes()).hexdigest() - - -def test_unchanged_sources_are_not_reported_as_stale(project): - source = project / "echo_agent.py" - source.write_text("x = 1\n") - write_porting_state(project, {"echo_agent.py": sha256(source)}) - - assert verify._stale_sources(str(project), str(project / ".car")) == [] - - -def test_changed_and_deleted_sources_are_reported(project): - source = project / "echo_agent.py" - source.write_text("x = 2\n") - write_porting_state( - project, {"echo_agent.py": "0" * 64, "gone.py": "0" * 64} - ) - - assert verify._stale_sources(str(project), str(project / ".car")) == [ - "echo_agent.py", - "gone.py", - ] - - -def test_the_skills_own_files_are_not_reported_as_drift(project): - write_porting_state(project, {".claude/skills/porting-to-canyonos/SKILL.md": "0" * 64}) - - assert verify._stale_sources(str(project), str(project / ".car")) == [] - - -def test_a_hand_written_artifact_has_no_state_to_compare(project): - assert verify._stale_sources(str(project), str(project / ".car")) == [] - - # ------------------------------------------------------------------ # # Runtime verification # # ------------------------------------------------------------------ # @@ -289,14 +104,11 @@ def test_the_workflow_endpoint_falls_back_to_the_configured_port(project, runtim @pytest.fixture def deployable(monkeypatch, project): - """A project where every step past the build check succeeds unless overridden.""" - calls = {"post_deploy": 0, "quit": 0} + """A project where every step succeeds unless overridden.""" + calls = {"run_deploy": 0, "quit": 0} - monkeypatch.setattr(test_cmd, "verify_build_artifact", lambda *a: {"warnings": 0, "stale": []}) - monkeypatch.setattr(test_cmd, "run_init", lambda banner=True, extra_env=None: None) - monkeypatch.setattr(test_cmd, "run_sync", lambda: True) monkeypatch.setattr(test_cmd, "load_state", lambda: {"container_id": "abc", "port": 8000}) - monkeypatch.setattr(test_cmd, "_port_in_use", lambda _port: False) + monkeypatch.setattr(test_cmd, "deploy_status", lambda *_a: None) monkeypatch.setattr(test_cmd, "_wait_for_workflow", lambda *a: None) monkeypatch.setattr(test_cmd, "verify_runtime", lambda *a: {"agents": []}) monkeypatch.setattr(test_cmd, "workflow_targets", lambda *a: [("Workflow", "127.0.0.1", 8080)]) @@ -304,13 +116,14 @@ def deployable(monkeypatch, project): monkeypatch.setattr(test_cmd, "_await_result", lambda *a: {"status": "done", "result": {"r": 1}}) monkeypatch.setattr(test_cmd, "_log_tail", lambda _cid: "boom") - def post_deploy(*_a, **_k): - calls["post_deploy"] += 1 + def run_deploy(*_a, **_k): + calls["run_deploy"] += 1 + return {"container_id": "abc", "port": 8000} def quit_existing(): calls["quit"] += 1 - monkeypatch.setattr(test_cmd, "post_deploy", post_deploy) + monkeypatch.setattr(test_cmd, "run_deploy", run_deploy) monkeypatch.setattr(test_cmd, "quit_existing", quit_existing) return calls @@ -321,11 +134,11 @@ def test_a_passing_run_tears_everything_down(deployable): def test_llm_is_stubbed_by_default(monkeypatch, deployable): - """`canyonos test` hands the stub flag to the GC container so no real LLM is hit.""" + """`canyonos test` hands the stub flag to `canyonos deploy` so no real LLM is hit.""" seen = {} monkeypatch.setattr( - test_cmd, "run_init", - lambda banner=True, extra_env=None: seen.update(extra_env=extra_env), + test_cmd, "run_deploy", + lambda *_a, **kwargs: seen.update(extra_env=kwargs.get("extra_env")) or {"container_id": "abc", "port": 8000}, ) assert test_cmd.run_test("hi") == 0 assert seen["extra_env"] == {"CANYONOS_LLM_STUB_TEXT": "test"} @@ -334,8 +147,8 @@ def test_llm_is_stubbed_by_default(monkeypatch, deployable): def test_real_llm_flag_disables_the_stub(monkeypatch, deployable): seen = {} monkeypatch.setattr( - test_cmd, "run_init", - lambda banner=True, extra_env=None: seen.update(extra_env=extra_env), + test_cmd, "run_deploy", + lambda *_a, **kwargs: seen.update(extra_env=kwargs.get("extra_env")) or {"container_id": "abc", "port": 8000}, ) assert test_cmd.run_test("hi", llm_stub=None) == 0 assert seen["extra_env"] is None @@ -349,16 +162,6 @@ def test_the_provider_is_restored_after_the_run(project, deployable): assert config.read_text() == CONFIG -def test_an_occupied_api_port_fails_before_the_deploy(monkeypatch, deployable, capsys): - monkeypatch.setattr(test_cmd, "_port_in_use", lambda _port: True) - - assert test_cmd.run_test("hi", as_json=True) == 1 - payload = json.loads(capsys.readouterr().out) - - assert deployable["post_deploy"] == 0 - assert "8080 is already in use" in payload["error"] - - def test_a_failed_deploy_keeps_the_container_and_reads_its_log(monkeypatch, deployable, capsys): def boom(*_a): raise RuntimeError("the deploy did not come up") @@ -372,16 +175,34 @@ def boom(*_a): assert payload["log_tail"] == "boom" -def test_a_failure_before_the_deploy_leaves_nothing_running(monkeypatch, deployable, capsys): - monkeypatch.setattr(test_cmd, "run_sync", lambda: False) +def test_a_failure_before_the_deploy_leaves_existing_state_alone(monkeypatch, deployable, capsys): + """A failure that never gets as far as starting this run's own deploy must + not tear down whatever deploy was already there -- see + test_refuses_to_run_when_a_deploy_is_already_up. + """ + def boom(*_a, **_k): + raise RuntimeError("Could not sync the project into the container.") + + monkeypatch.setattr(test_cmd, "run_deploy", boom) assert test_cmd.run_test("hi", as_json=True) == 1 payload = json.loads(capsys.readouterr().out) - assert deployable["quit"] == 1 + assert deployable["quit"] == 0 assert payload["log_tail"] is None +def test_refuses_to_run_when_a_deploy_is_already_up(monkeypatch, deployable, capsys): + monkeypatch.setattr(test_cmd, "deploy_status", lambda *_a: {"running": True}) + + assert test_cmd.run_test("hi", as_json=True) == 1 + payload = json.loads(capsys.readouterr().out) + + assert deployable["run_deploy"] == 0 + assert deployable["quit"] == 0 + assert payload["error"] == test_cmd._DEPLOY_CONFLICT + + def test_json_mode_prints_one_object_and_nothing_else(deployable, capsys): assert test_cmd.run_test("a prompt", as_json=True) == 0 payload = json.loads(capsys.readouterr().out) @@ -391,7 +212,6 @@ def test_json_mode_prints_one_object_and_nothing_else(deployable, capsys): assert payload["result"] == {"r": 1} assert payload["error"] is None assert [(p["name"], p["ok"]) for p in payload["phases"]] == [ - ("verify_build", True), ("deploy", True), ("verify_runtime", True), ("query", True), @@ -408,25 +228,20 @@ def test_a_workflow_error_is_reported_as_a_failure(monkeypatch, deployable, caps assert json.loads(capsys.readouterr().out)["error"] == "agent blew up" -def test_a_flat_layout_project_skips_the_build_check(monkeypatch, tmp_path, deployable, capsys): +def test_a_flat_layout_project_deploys_fine_with_no_car_directory(monkeypatch, tmp_path, deployable, capsys): legacy = tmp_path / "legacy" / "config" legacy.mkdir(parents=True) (legacy / "global_controller.yaml").write_text(CONFIG) monkeypatch.chdir(tmp_path / "legacy") - def unexpected(*_a): - raise AssertionError("the artifact validator should not run without a .car/") - - monkeypatch.setattr(test_cmd, "verify_build_artifact", unexpected) - assert test_cmd.run_test("hi", as_json=True) == 0 payload = json.loads(capsys.readouterr().out) - assert payload["phases"][0] == { - "name": "verify_build", - "ok": True, - "detail": "skipped: no .car/ artifact", - } + assert [(p["name"], p["ok"]) for p in payload["phases"]] == [ + ("deploy", True), + ("verify_runtime", True), + ("query", True), + ] # ------------------------------------------------------------------ # diff --git a/tests/test_dashboard_stack.py b/tests/test_dashboard_stack.py index 448c1a1..77d2600 100644 --- a/tests/test_dashboard_stack.py +++ b/tests/test_dashboard_stack.py @@ -279,6 +279,34 @@ def urlopen(endpoint, timeout): ] +def test_stop_and_teardown_are_a_noop_without_an_env_file(monkeypatch, project): + calls = [] + install_docker(monkeypatch, calls) + + assert dashboard_stack.stop_dashboard() is False + assert dashboard_stack.teardown_dashboard() is False + assert calls == [] + + +def test_stop_dashboard_runs_compose_stop(monkeypatch, project): + calls = [] + install_docker(monkeypatch, calls) + Path.cwd().joinpath(".env").write_text("CANYONOS_JWT_SECRET=x\n") + + assert dashboard_stack.stop_dashboard() is True + assert calls[-1][-1] == "stop" + assert calls[-1][4:6] == ["--env-file", str(project / ".env")] + + +def test_teardown_dashboard_runs_compose_down(monkeypatch, project): + calls = [] + install_docker(monkeypatch, calls) + Path.cwd().joinpath(".env").write_text("CANYONOS_JWT_SECRET=x\n") + + assert dashboard_stack.teardown_dashboard() is True + assert calls[-1][-1] == "down" + + def test_existing_dashboard_container_skips_port_check(monkeypatch, project): calls = [] diff --git a/tests/test_deploy_progress.py b/tests/test_deploy_progress.py index db0bb5a..647c5c5 100644 --- a/tests/test_deploy_progress.py +++ b/tests/test_deploy_progress.py @@ -211,7 +211,7 @@ def test_the_clis_own_status_requests_are_not_shown_or_buffered(monkeypatch): ] ) ) - summary = deploy_cmd._tail_quiet(lines, {"port": 1}, 8080, serve=False) + summary = deploy_cmd._tail_quiet(lines, {"port": 1}, 8080, "config/global_controller.yaml", serve=False) assert summary == ("url", []) assert shown == ["Build complete", "Workflow ready"] @@ -229,7 +229,7 @@ def test_a_build_that_dies_silently_does_not_hang(monkeypatch, capsys): # The queue never yields None: the stream stays open, as it does in reality. lines.put = lambda *a, **k: None - summary = deploy_cmd._tail_quiet(lines, {"port": 1}, 8080, serve=False) + summary = deploy_cmd._tail_quiet(lines, {"port": 1}, 8080, "config/global_controller.yaml", serve=False) assert summary is None assert "Building 2 Docker image(s)" in capsys.readouterr().out diff --git a/tests/test_instance_manager_runtime.py b/tests/test_instance_manager_runtime.py index 41bfb50..93b2836 100644 --- a/tests/test_instance_manager_runtime.py +++ b/tests/test_instance_manager_runtime.py @@ -219,16 +219,17 @@ def test_local_workflow_and_resource_flags_stay_the_same(self): controller = _fake_controller() manager = InstanceManager(controller, controller.redis) - manager.ensure_instances( - [ - { - "name": "Workflow", - "provider": "local", - "type": "workflow", - "resources": {"cpu": 2, "memory": 1024, "gpu": 1}, - } - ] - ) + with patch.object(local_runtime, "_port_bound", return_value=False): + manager.ensure_instances( + [ + { + "name": "Workflow", + "provider": "local", + "type": "workflow", + "resources": {"cpu": 2, "memory": 1024, "gpu": 1}, + } + ] + ) self.assertEqual( controller._run_cmd.call_args.args, @@ -273,6 +274,31 @@ def test_local_workflow_and_resource_flags_stay_the_same(self): ), ) + def test_workflow_bootstrap_fails_fast_on_an_occupied_api_port(self): + controller = _fake_controller() + manager = InstanceManager(controller, controller.redis) + + with patch.object(local_runtime, "_port_bound", return_value=True): + with self.assertRaises(RuntimeError) as ctx: + manager.ensure_instances( + [{"name": "Workflow", "provider": "local", "type": "workflow"}] + ) + + self.assertIn("api_port 8080", str(ctx.exception)) + # The orphan-container `docker inspect` probe still runs -- only `docker run` is skipped. + run_calls = [c for c in controller._run_cmd.call_args_list if c.args[0][:2] == ["docker", "run"]] + self.assertEqual(run_calls, []) + + def test_plain_agent_bootstrap_ignores_api_port_conflicts(self): + """Only `type: workflow` publishes api_port -- a plain agent has nothing to conflict on.""" + controller = _fake_controller() + manager = InstanceManager(controller, controller.redis) + + with patch.object(local_runtime, "_port_bound", return_value=True): + manager.ensure_instances([{"name": "Alpha", "provider": "local"}]) + + controller._run_cmd.assert_called() + def test_agent_id_is_stable_across_repeated_ensure_instances_calls(self): controller = _fake_controller() manager = InstanceManager(controller, controller.redis) From 1c239584ee0182c0f2a3bdf0973e22f08dc9545a Mon Sep 17 00:00:00 2001 From: Nick Huo Date: Wed, 9 Sep 2026 14:54:12 -0700 Subject: [PATCH 06/10] Load nested agent entrypoints with a dotted module name _load_agent derived the module name for spec_from_file_location by stripping .py from the entrypoint path, leaving directory separators intact (e.g. "agents/aml_agent"). Without dots, Python can't establish __package__, so any relative import inside a nested entrypoint (e.g. `from .prompts import PROMPT`) fails with "attempted relative import with no known parent package" at agent load, surfacing as "No agent loaded" at serve time. Convert path separators to dots so __package__ resolves correctly and sibling relative imports work. Co-Authored-By: Claude Sonnet 5 --- canyonos_core/controller/local_controller.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/canyonos_core/controller/local_controller.py b/canyonos_core/controller/local_controller.py index b4e5adb..3ee9329 100644 --- a/canyonos_core/controller/local_controller.py +++ b/canyonos_core/controller/local_controller.py @@ -194,7 +194,12 @@ def _load_agent(self): ) return None - agent_module_name = self.agent_file.replace(".py", "") + # Use a dotted module name (e.g. "agents.aml_agent", not "agents/aml_agent") + # so importlib derives __package__ correctly and package-relative imports + # inside nested entrypoints (e.g. `from .prompts import PROMPT`) resolve. + agent_module_name = ( + self.agent_file.replace(".py", "").replace(os.sep, ".").replace("/", ".") + ) # We assume the agent file is in the same directory as the local controller (e.g. copied by Docker) # or in the current working directory. From 0fb724e3eec0e0ba1807f6d5f163d1cb907053e8 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Thu, 10 Sep 2026 15:36:08 -0700 Subject: [PATCH 07/10] Fix project_id dashless-hex regression: use dashed UUID _assign_new_project_id() minted uuid.uuid4().hex (32-char, no dashes). Dashboard api's bootstrap_canyonos() requires a dashed UUID (UUID_RE / z.string().uuid()) and silently bails with 'CanyonOS identity is unusable, bootstrapping no project' when it doesn't match, leaving the projects table empty and the dashboard UI blank despite a healthy serve stack and spans flowing end-to-end. Fixed by switching to str(uuid.uuid4()). Updated test_global_controller_project_id.py's UUID_HEX_RE to match the dashed format, consistent with test_global_controller_identity.py and test_global_controller_reload.py, which already used dashed fixtures. This exact fix has been written and lost multiple times across unmerged branches (CAN-316-canyonos-serve, docs/fleshing-docs) without ever landing on main. --- canyonos_core/controller/global_controller.py | 2 +- cli/canyonos/constants.py | 39 +++++++++++++++++++ cli/canyonos/deploy.py | 20 +++++++++- tests/test_global_controller_project_id.py | 4 +- 4 files changed, 61 insertions(+), 4 deletions(-) diff --git a/canyonos_core/controller/global_controller.py b/canyonos_core/controller/global_controller.py index 5d0d0c1..934a30f 100644 --- a/canyonos_core/controller/global_controller.py +++ b/canyonos_core/controller/global_controller.py @@ -220,7 +220,7 @@ def _load_config(config_path): @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 = uuid.uuid4().hex + project_id = str(uuid.uuid4()) with open(config_path, "a") as f: f.write(f'project_id: "{project_id}"\n') return project_id diff --git a/cli/canyonos/constants.py b/cli/canyonos/constants.py index 05ac30f..f7c9993 100644 --- a/cli/canyonos/constants.py +++ b/cli/canyonos/constants.py @@ -5,6 +5,8 @@ import ast import os import socket +import urllib.error +import urllib.request import yaml from ruamel.yaml import YAML @@ -12,6 +14,12 @@ DEFAULT_API_PORT = 8080 DEFAULT_DASHBOARD_PORT = 8081 +_EC2_TOKEN_URL = "http://169.254.169.254/latest/api/token" +_EC2_PUBLIC_IP_URL = "http://169.254.169.254/latest/meta-data/public-ipv4" + +_public_ip_cache = None +_public_ip_checked = False + # Fallback when the real function name/params can't be determined statically # (see workflow_entrypoint) -- canyonos_core's own examples all follow this shape. WORKFLOW_ROUTE = "main" @@ -24,6 +32,37 @@ def default_config_path(): return car if os.path.isfile(car) else os.path.join("config", "global_controller.yaml") +def public_ip(timeout=0.3): + """This machine's public IP via the EC2 IMDSv2 metadata service, or None. + + Only meaningful on an EC2 instance -- a laptop or any other host simply + fails to reach the link-local metadata address and gets None back. Kept to + a short timeout and cached for the life of the process so a non-EC2 host + doesn't pay a network-timeout tax on every call site that wants a display + host (deploy summary, `status`, etc). + """ + global _public_ip_cache, _public_ip_checked + if _public_ip_checked: + return _public_ip_cache + _public_ip_checked = True + try: + token_req = urllib.request.Request( + _EC2_TOKEN_URL, + method="PUT", + headers={"X-aws-ec2-metadata-token-ttl-seconds": "21600"}, + ) + token = urllib.request.urlopen(token_req, timeout=timeout).read().decode() + ip_req = urllib.request.Request( + _EC2_PUBLIC_IP_URL, headers={"X-aws-ec2-metadata-token": token} + ) + _public_ip_cache = ( + urllib.request.urlopen(ip_req, timeout=timeout).read().decode().strip() or None + ) + except (OSError, urllib.error.URLError): + _public_ip_cache = None + return _public_ip_cache + + def workflow_api_port(config_path): """Host port the workflow answers on, or None if there isn't one to read.""" try: diff --git a/cli/canyonos/deploy.py b/cli/canyonos/deploy.py index 8e61e27..1dad947 100644 --- a/cli/canyonos/deploy.py +++ b/cli/canyonos/deploy.py @@ -32,6 +32,7 @@ WORKFLOW_ROUTE, default_config_path, port_in_use, + public_ip, workflow_api_port, workflow_entrypoint, workspace_relative, @@ -186,6 +187,21 @@ def run_deploy(config_path=None, serve=True, verbose=False, quiet=False, extra_e return state +def _display_host(host): + """Substitute this machine's own public IP for a loopback host, when discoverable. + + A workflow placed on *this* machine reports `127.0.0.1`/`localhost` -- correct + for curling from the box itself, but useless from anywhere else (e.g. an EC2 + deploy meant to be queried from a laptop). Falls back to `127.0.0.1` off EC2 + (or if the metadata lookup fails), same as before. A workflow placed on a + *different* machine already reports its own real host and passes through + unchanged. + """ + if host not in ("127.0.0.1", "localhost"): + return host + return public_ip() or "127.0.0.1" + + def workflow_targets(gc_port, api_port): """(name, host, port) for each deployed workflow. @@ -196,7 +212,7 @@ def workflow_targets(gc_port, api_port): targets = [ ( endpoint.get("name"), - "127.0.0.1" if endpoint["host"] in ("127.0.0.1", "localhost") else endpoint["host"], + _display_host(endpoint["host"]), endpoint["port"], ) for endpoint in workflow_endpoints(gc_port) @@ -204,7 +220,7 @@ def workflow_targets(gc_port, api_port): ] if targets: return targets - return [(None, "127.0.0.1", api_port)] if api_port else [] + return [(None, _display_host("127.0.0.1"), api_port)] if api_port else [] def _example_route_and_body(config_path): diff --git a/tests/test_global_controller_project_id.py b/tests/test_global_controller_project_id.py index cf5f43a..88ab9dd 100644 --- a/tests/test_global_controller_project_id.py +++ b/tests/test_global_controller_project_id.py @@ -15,7 +15,9 @@ from canyonos_core.controller.global_controller import GlobalController -UUID_HEX_RE = re.compile(r"^[0-9a-f]{32}$") +UUID_HEX_RE = re.compile( + r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" +) def _write_config(body): From 037d17ca043793762785543022753894ae356f22 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Thu, 10 Sep 2026 15:51:57 -0700 Subject: [PATCH 08/10] Fix OSC 11 background-color query leaking onto SSH terminals _is_light_background() gave the terminal only 100ms to answer the OSC 11 background-color query before restoring cooked+echo tty mode. Fine for a local terminal, but a real SSH round-trip (e.g. to an EC2 box) can exceed that, so the reply arrives after echo is back on and gets displayed as literal text ahead of the next command ("^[]11;rgb:.../ ^[\"). Widened the timeout to 400ms and, more importantly, drain any bytes still pending on the fd (right before restoring termios) so a late or duplicate reply is swallowed instead of leaking into the shell. --- cli/canyonos/theme.py | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/cli/canyonos/theme.py b/cli/canyonos/theme.py index cd2a44e..31fc0c4 100644 --- a/cli/canyonos/theme.py +++ b/cli/canyonos/theme.py @@ -29,8 +29,18 @@ def _is_light_background(): tty.setraw(fd) sys.stdout.write("\x1b]11;?\x07") sys.stdout.flush() - reply = os.read(fd, 32).decode(errors="ignore") if select.select([fd], [], [], 0.1)[0] else "" + # 100ms is fine locally but can be exceeded by a real SSH round-trip; + # 400ms gives a laggy remote session a real chance to answer before + # we give up and assume dark. + reply = os.read(fd, 32).decode(errors="ignore") if select.select([fd], [], [], 0.4)[0] else "" finally: + # A reply that arrives just after our timeout (or a second stray one) + # would otherwise sit in the tty buffer and get echoed as literal text + # ahead of the next command once we restore cooked/echo mode below -- + # drain anything still pending, non-blockingly, before that happens. + while select.select([fd], [], [], 0)[0]: + if not os.read(fd, 1024): + break termios.tcsetattr(fd, termios.TCSADRAIN, old) m = re.search(r"rgb:([0-9a-f]{2})\S*/([0-9a-f]{2})\S*/([0-9a-f]{2})", reply, re.I) return bool(m) and (0.299 * int(m[1], 16) + 0.587 * int(m[2], 16) + 0.114 * int(m[3], 16)) > 128 From c0566617066e077c7377d805e6e1c60821fe56c8 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Thu, 10 Sep 2026 16:03:18 -0700 Subject: [PATCH 09/10] deploy: don't treat the OTel exporter's own ERROR: lines as a fatal deploy failure _ERROR_MARKERS' naive "ERROR:" substring match caught ERROR:opentelemetry.exporter.otlp.proto.http.trace_exporter:Failed to export span batch due to timeout... -- which fires on every cold deploy, since the dashboard (the OTel destination) hasn't been started yet at that point in the sequence and the exporter is just retrying as designed. That false positive broke the tail loop before it ever reached the 'Global controller started' success marker, so _start_dashboard() (canyonos serve, bundled into deploy) never got called even though the deploy had fully succeeded. Added _BENIGN_ERROR_PREFIXES, checked before _ERROR_MARKERS, so this logger's own ERROR: lines are treated as non-fatal noise instead. --- cli/canyonos/deploy.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/cli/canyonos/deploy.py b/cli/canyonos/deploy.py index 1dad947..d26490d 100644 --- a/cli/canyonos/deploy.py +++ b/cli/canyonos/deploy.py @@ -58,6 +58,14 @@ "process did not complete successfully", ) +# Loggers whose own ERROR: lines are expected, self-recovering noise -- not a +# reason to abort the deploy. Checked before _ERROR_MARKERS so they never +# match: the OTel exporter logs at ERROR: when a destination (the dashboard's +# ingest) isn't reachable yet, which is normal on every cold deploy since +# `canyonos serve` hasn't been started at that point -- it retries and +# recovers on its own once the dashboard comes up. +_BENIGN_ERROR_PREFIXES = ("ERROR:opentelemetry.",) + # (substring, spinner message, completed message). A None spinner message keeps # whatever the spinner already shows; a None completed message prints nothing. # Matched by substring against the raw line, so a phase that never runs is simply @@ -94,6 +102,8 @@ def _agent_progress(self): return "Starting agents..." def feed(self, line): + if any(prefix in line for prefix in _BENIGN_ERROR_PREFIXES): + return None, None, False if any(marker in line for marker in _ERROR_MARKERS): return None, None, True From 33ab2f5055d08b1c9273099ecf86a64205d01085 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Thu, 10 Sep 2026 16:32:03 -0700 Subject: [PATCH 10/10] Quick Terminal UI Addition --- cli/canyonos/deploy.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/cli/canyonos/deploy.py b/cli/canyonos/deploy.py index d26490d..dd87b01 100644 --- a/cli/canyonos/deploy.py +++ b/cli/canyonos/deploy.py @@ -247,10 +247,15 @@ def _example_route_and_body(config_path): def _curl_example(url, body): - """A copy-pasteable `curl -X POST ...` block, indented to sit under the summary's other rows.""" - json_lines = json.dumps(body, indent=2).splitlines() - indented_body = "\n".join(line if i == 0 else f" {line}" for i, line in enumerate(json_lines)) - return f'curl -X POST {url} \\\n -H "Content-Type: application/json" \\\n -d \'{indented_body}\'' + """A single-line, directly copy-pasteable `curl -X POST ...` command. + + Deliberately not split across `\\`-continued lines or pretty-printed JSON -- + a multi-line block is easy to mangle depending on what actually receives the + paste (some terminals/chat boxes drop the backslashes or the newlines), and + a single line always works no matter where it lands. + """ + compact_body = json.dumps(body) + return f'curl -X POST {url} -H "Content-Type: application/json" -d \'{compact_body}\'' def _summary_body(dashboard_url, targets, config_path):