diff --git a/.env.example b/.env.example index 8990feb5..ee31ecbf 100644 --- a/.env.example +++ b/.env.example @@ -177,6 +177,7 @@ TRAIN_UPLOAD_LIMIT_MB=4096 SEARCH_ENABLE_WEB_SEARCH=true SEARCH_RESULT_COUNT=5 +SEARCH_TIMEOUT_SECONDS=15 SEARCH_SEARXNG_QUERY_URL="http://gluetun:8080/search?q=&format=json" SEARCH_SEARXNG_SERVER_BASE_URL="http://localhost:8080" SEARCH_SEARXNG_INSTANCE_NAME="Zone Search" @@ -282,8 +283,11 @@ ZONE_MCP_AUTO_MAGENTS=true # ZONE_NOTIFY_DISCORD_WEBHOOK=https://discord.com/api/webhooks/... # ZONE_NOTIFY_EMAIL_TO=alerts@example.com,ops@example.com # ZONE_NOTIFY_TIMEOUT_SECONDS=10 -# SMTP_* - The relay ZONE_NOTIFY_EMAIL_TO, sign-up verification, password -# resets and invitations send through. ALERT_SMTP_* below is Grafana's own. +# SMTP_* - The relay ZONE_NOTIFY_EMAIL_TO, sign-up verification and password +# resets send through. Port 465 connects with implicit TLS; any other port, +# 587 included, upgrades with STARTTLS before signing in. A blank value other +# than SMTP_PASSWORD counts as unset, and an unreadable SMTP_PORT falls back +# to 587. ALERT_SMTP_* below is Grafana's own. # SMTP_HOST=smtp.example.com # SMTP_PORT=587 # SMTP_USER= diff --git a/README.md b/README.md index ad29670e..006859a6 100644 --- a/README.md +++ b/README.md @@ -433,8 +433,6 @@ zone/ │ │ ├── src/video.rs # Training frames pulled out of a submitted clip │ │ ├── src/subject.rs # Subject-aware framing for training crops │ │ └── src/caption.rs # Vision captioning for training sets -│ ├── zone_email/ # Transactional email over SMTP -│ ├── zone_search/ # SearXNG web search client │ ├── zone_vcs/ # Local git operations and GitHub pull requests │ ├── zone_vision/ # Subject detection and subject-aware cropping │ ├── zone_server/ # HTTP/WS server diff --git a/comfyui/custom_nodes/zone_lora/__init__.py b/comfyui/custom_nodes/zone_lora/__init__.py index b501b057..9a7ea380 100644 --- a/comfyui/custom_nodes/zone_lora/__init__.py +++ b/comfyui/custom_nodes/zone_lora/__init__.py @@ -1,4 +1,4 @@ -from comfy_api.latest import ComfyExtension +from comfy_api.latest import ComfyExtension, io from .cleanup_node import ZoneCleanupTrainingRun from .dataset_node import ZoneLoadTrainDataset @@ -11,16 +11,13 @@ install_all() -NODE_CLASS_MAPPINGS = { - 'ZoneIPAdapterFace': ZoneIPAdapterFace, -} - class ZoneLoraExtension(ComfyExtension): - async def get_node_list(self): + async def get_node_list(self) -> list[type[io.ComfyNode]]: return [ ZoneLoadTrainDataset, ZoneCleanupTrainingRun, + ZoneIPAdapterFace, ZoneLoadTrainFolder, ZoneProbeGradient, ZoneProbeLoss, diff --git a/comfyui/custom_nodes/zone_lora/face_node.py b/comfyui/custom_nodes/zone_lora/face_node.py index ac60dd7d..30aa63dc 100644 --- a/comfyui/custom_nodes/zone_lora/face_node.py +++ b/comfyui/custom_nodes/zone_lora/face_node.py @@ -11,6 +11,7 @@ from comfy.ldm.modules.attention import optimized_attention import comfy.model_management as model_management import comfy.utils +from comfy_api.latest import io PLUS_FACE = 'ip-adapter-plus-face_sdxl_vit-h.safetensors' @@ -267,32 +268,30 @@ def patch_sdxl(model, ipa, cond, uncond, weight: float): number += 1 -class ZoneIPAdapterFace: +class ZoneIPAdapterFace(io.ComfyNode): @classmethod - def INPUT_TYPES(cls): - names = folder_paths.get_filename_list('ipadapter') - if not names: - names = [PLUS_FACE] - return { - 'required': { - 'model': ('MODEL',), - 'clip_vision': ('CLIP_VISION',), - 'image': ('IMAGE',), - 'ipadapter_name': (names,), - 'strength': ( - 'FLOAT', - {'default': 0.4, 'min': 0.0, 'max': 2.0, 'step': 0.01}, - ), - } - } - - RETURN_TYPES = ('MODEL',) - FUNCTION = 'apply' - CATEGORY = 'zone' - - def apply(self, model, clip_vision, image, ipadapter_name, strength): + def define_schema(cls) -> io.Schema: + names = folder_paths.get_filename_list('ipadapter') or [PLUS_FACE] + return io.Schema( + node_id='ZoneIPAdapterFace', + display_name='Zone IP-Adapter Face', + category='zone', + inputs=[ + io.Model.Input('model'), + io.ClipVision.Input('clip_vision'), + io.Image.Input('image'), + io.Combo.Input('ipadapter_name', options=names), + io.Float.Input('strength', default=0.4, min=0.0, max=2.0, step=0.01), + ], + outputs=[io.Model.Output()], + ) + + @classmethod + def execute( + cls, model, clip_vision, image, ipadapter_name: str, strength: float + ) -> io.NodeOutput: if strength == 0: - return (model,) + return io.NodeOutput(model) path = folder_paths.get_full_path_or_raise('ipadapter', ipadapter_name) loaded = load_plus_face(path) if 'latents' not in loaded['image_proj']: @@ -314,4 +313,4 @@ def apply(self, model, clip_vision, image, ipadapter_name, strength): uncond = uncond.to(device, dtype=dtype) work = model.clone() patch_sdxl(work, ipa, cond, uncond, float(strength)) - return (work,) + return io.NodeOutput(work) diff --git a/comfyui/tests/test_graph_contract.py b/comfyui/tests/test_graph_contract.py index ac249179..066c0257 100644 --- a/comfyui/tests/test_graph_contract.py +++ b/comfyui/tests/test_graph_contract.py @@ -3,6 +3,7 @@ import asyncio import json import os +import re import subprocess import sys import tempfile @@ -19,6 +20,7 @@ ZONE_NODES = COMFYUI / 'custom_nodes' / 'zone_lora' TRAIN_SERVER = ROOT / 'runner/zone_comfy/src/train.rs' QUALITY_SERVER = ROOT / 'runner/zone_comfy/src/quality.rs' +PEOPLE_SERVER = ROOT / 'runner/zone_comfy/src/people.rs' if not INSTALL.is_dir(): raise RuntimeError('COMFYUI_INSTALL_DIR must point to the pinned ComfyUI checkout') @@ -165,6 +167,53 @@ def rust_graphs() -> dict[str, dict]: } +def resolve(value: object, constants: dict[str, str]) -> object: + if isinstance(value, Expression): + return constants.get(value.text, value) + if isinstance(value, list): + return [resolve(item, constants) for item in value] + if isinstance(value, dict): + return {name: resolve(item, constants) for name, item in value.items()} + return value + + +def people_strings() -> dict[str, str]: + return dict(re.findall(r'const (\w+): &str = "([^"]*)";', PEOPLE_SERVER.read_text())) + + +def people_numbers() -> dict[str, float]: + return { + name: float(value) + for name, value in re.findall(r'const (\w+): f64 = ([0-9.]+);', PEOPLE_SERVER.read_text()) + } + + +def rust_face_graph() -> dict: + source = PEOPLE_SERVER.read_text() + constants = people_strings() + start = source.find('fn inject_ipadapter(') + if start < 0: + raise AssertionError(f'{PEOPLE_SERVER}: inject_ipadapter is missing') + end = source.find('\n}\n', start) + if end <= start: + raise AssertionError(f'{PEOPLE_SERVER}: inject_ipadapter has no closing brace') + graph = { + '4': { + 'class_type': 'CheckpointLoaderSimple', + 'inputs': {'ckpt_name': 'model.safetensors'}, + } + } + for insert in re.finditer(r'object\.insert\(\s*(\w+)\.into\(\),\s*json!\(', source[start:end]): + key = insert.group(1) + if key not in constants: + raise AssertionError(f'{PEOPLE_SERVER}: node key {key} is not a &str constant') + node = RustLiteral(source, start + insert.end(), PEOPLE_SERVER).object() + graph[constants[key]] = resolve(node, constants) + face = graph[constants['IPADAPTER_NODE']] + face['inputs']['model'] = ['4', 0] + return graph + + def flux() -> train_lora.TrainingModel: return train_lora.TrainingModel(architecture='flux', checkpoint='flux.safetensors') @@ -371,6 +420,7 @@ def test_repository_extension_is_part_of_the_loaded_registry(self) -> None: 'ZoneCleanupTrainingRun', 'ZoneIPAdapterFace', 'ZoneLoadTrainDataset', + 'ZoneLoadTrainFolder', 'ZoneProbeGradient', 'ZoneProbeLoss', 'ZoneStageTrainingArtifact', @@ -378,6 +428,51 @@ def test_repository_extension_is_part_of_the_loaded_registry(self) -> None: } self.assertLessEqual(expected, set(self.nodes.NODE_CLASS_MAPPINGS)) + def test_rust_face_graph_matches_the_pinned_registry(self) -> None: + graph = rust_face_graph() + self.assertEqual( + sorted(node['class_type'] for node in graph.values()), + ['CLIPVisionLoader', 'CheckpointLoaderSimple', 'LoadImage', 'ZoneIPAdapterFace'], + ) + self.validate_graph('Rust face', graph) + + def test_rust_face_constants_fit_the_face_node(self) -> None: + node_class = self.nodes.NODE_CLASS_MAPPINGS['ZoneIPAdapterFace'] + inputs, _, _ = schema(node_class) + bounds = inputs['strength'][1] + strength = people_numbers()['IPADAPTER_STRENGTH'] + self.assertGreaterEqual(strength, bounds['min'], f'{PEOPLE_SERVER}: IPADAPTER_STRENGTH') + self.assertLessEqual(strength, bounds['max'], f'{PEOPLE_SERVER}: IPADAPTER_STRENGTH') + self.assertEqual( + people_strings()['IPADAPTER_FILE'], + sys.modules[node_class.__module__].PLUS_FACE, + f'{PEOPLE_SERVER}: IPADAPTER_FILE', + ) + + def test_face_node_at_zero_strength_returns_the_model_untouched(self) -> None: + from execution import _async_map_node_over_list + + node_class = self.nodes.NODE_CLASS_MAPPINGS['ZoneIPAdapterFace'] + model = object() + results = asyncio.run( + _async_map_node_over_list( + 'contract', + 'face', + node_class, + { + 'model': [model], + 'clip_vision': [None], + 'image': [None], + 'ipadapter_name': ['missing.safetensors'], + 'strength': [0.0], + }, + node_class.FUNCTION, + v3_data={}, + ) + ) + self.assertEqual(len(results), 1) + self.assertIs(results[0].result[0], model) + def test_sdxl_people_graphs_match_the_pinned_registry(self) -> None: for name in ( 'sdxl-api.json', diff --git a/comfyui/tests/test_setup_macos.py b/comfyui/tests/test_setup_macos.py index e0908ff3..9a826c21 100644 --- a/comfyui/tests/test_setup_macos.py +++ b/comfyui/tests/test_setup_macos.py @@ -202,6 +202,20 @@ def test_install_trainer_inherits_comfy_site_packages(self) -> None: text=True, ) self.assertNotEqual(mlx.returncode, 0, mlx.stdout + mlx.stderr) + for trainer in (".venv-train", ".venv-train-llm"): + trainer_site = subprocess.check_output( + [ + str(install / trainer / "bin" / "python"), + "-c", + "import sysconfig; print(sysconfig.get_path('purelib'))", + ], + text=True, + ).strip() + with self.subTest(venv=trainer): + self.assertFalse( + Path(trainer_site, "pip").exists(), + "a skipped pip install still bootstrapped pip into the trainer venv", + ) plist = (home / "Library/LaunchAgents/ai.zone.train.plist").read_text( encoding="utf-8" ) diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 87c14fc2..0e729f2e 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -68,6 +68,7 @@ services: - VPN_OPENVPN_USER=${VPN_OPENVPN_USER:+1} - "SEARCH_SEARXNG_QUERY_URL=${SEARCH_SEARXNG_QUERY_URL:-http://gluetun:8080/search?q=&format=json}" - SEARCH_RESULT_COUNT=${SEARCH_RESULT_COUNT:-5} + - SEARCH_TIMEOUT_SECONDS=${SEARCH_TIMEOUT_SECONDS:-15} - MODEL_SEARCH_PROXY_URL=${MODEL_SEARCH_PROXY_URL:-} - GITHUB_API_URL=${GITHUB_API_URL:-} - TOOL_RUNNER_PROXY_URL=${TOOL_RUNNER_PROXY_URL:-} diff --git a/docker-compose.yml b/docker-compose.yml index 634c5493..80667f25 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -387,6 +387,7 @@ services: - VPN_OPENVPN_USER=${VPN_OPENVPN_USER:+1} - "SEARCH_SEARXNG_QUERY_URL=${SEARCH_SEARXNG_QUERY_URL:-http://gluetun:8080/search?q=&format=json}" - SEARCH_RESULT_COUNT=${SEARCH_RESULT_COUNT:-5} + - SEARCH_TIMEOUT_SECONDS=${SEARCH_TIMEOUT_SECONDS:-15} # Belt-and-suspenders HTTP proxy. `make up-vpn` also attaches this # container to Gluetun's network so non-HTTP traffic uses the tunnel. - MODEL_SEARCH_PROXY_URL=${MODEL_SEARCH_PROXY_URL:-} diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 6cb89b0a..6ba6d6c1 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -1623,7 +1623,11 @@ details, and native macOS / bundled NVIDIA instructions. ### `SEARCH_RESULT_COUNT` - **Default**: `5` -- **Description**: Number of search results supplied to chat +- **Description**: Number of search results supplied to chat (held to 1–20). + +### `SEARCH_TIMEOUT_SECONDS` +- **Default**: `15` +- **Description**: Timeout for one SearXNG request, in whole seconds (held to 1–60). ### `SEARCH_SEARXNG_QUERY_URL` - **Default**: `"http://gluetun:8080/search?q=&format=json"` @@ -1841,8 +1845,8 @@ Every notice — an auto-project merge, pause or completion, a regression alert, - **Description**: How long one channel is given per notice. ### `SMTP_HOST`, `SMTP_PORT`, `SMTP_USER`, `SMTP_PASSWORD`, `SMTP_FROM`, `SMTP_FROM_NAME` -- **Default**: *empty* (`SMTP_PORT` 587, `SMTP_FROM_NAME` Zone) -- **Description**: The relay notices, sign-up verification, password resets and invitations send through. `ALERT_SMTP_*` is Grafana's relay and is separate. +- **Default**: *empty* (`SMTP_PORT` 587, `SMTP_FROM_NAME` Zone; a blank value other than `SMTP_PASSWORD` counts as unset and an unreadable `SMTP_PORT` falls back to 587; sign-up and reset mail falls back to `noreply@zone.app` when `SMTP_FROM` is unset, email notices need it set) +- **Description**: The relay notices, sign-up verification and password resets send through. Port 465 connects with implicit TLS; any other port, 587 included, must upgrade with STARTTLS before signing in. `ALERT_SMTP_*` is Grafana's relay and is separate. --- diff --git a/docs/DEPLOY.md b/docs/DEPLOY.md index 2f20d235..bd15b287 100644 --- a/docs/DEPLOY.md +++ b/docs/DEPLOY.md @@ -58,7 +58,7 @@ Verified on `main` (`d3fb7cf`). Reuse is the default; each row says whether it's | MQTT broker | `~/Local/mqtt` (Swoole) | OTA substrate for `ota_fleet` | | Appwrite | Sites, Functions, Storage, Messaging | Delivery targets + backend provisioning (v1.5) | | Charts, cluster | `helm/zone-apps` (`_helpers.tpl` conventions, security contexts, HPA/PDB/NetworkPolicy), `k8s/kind-config.yaml` (`zone-dev`, ports 30000-30002 pre-mapped, CNPG, HAProxy default IngressClass) | New `helm/zone-app` copies conventions; kind is the first k8s target | -| Notifications | None in the server. `ALERT_DISCORD_WEBHOOK_URL` is consumed only by Grafana (`grafana/entrypoint.sh`); `zone_email::EmailService` has three templated senders; reminders deliver as assistant messages into a chat (`db/reminders.rs:61-90`) | New `notify` module | +| Notifications | None in the server. `ALERT_DISCORD_WEBHOOK_URL` is consumed only by Grafana (`grafana/entrypoint.sh`); `services::mail::AccountMail` sends two templated account messages (abnegate-notify `Mailer`); reminders deliver as assistant messages into a chat (`db/reminders.rs:61-90`) | New `notify` module | **Gaps that block the feature (phase 0):** @@ -639,7 +639,7 @@ Stated tradeoff: while blocked, a task holds one of `MAX_CONCURRENT_TASKS = 5` p `Notifier` trait with three channels, fanned out with `join_all`, failures logged and never blocking the deployment: - `DiscordWebhook` — an embed with app, target, artifact kind, digest, requester, values digest, and a deep link to `/deployments?id=…` (the desktop/mobile shell opens it too). Webhook URL from the target's `notify` block, falling back to a server-level `ALERT_DISCORD_WEBHOOK_URL` that `config.rs` now reads as well as Grafana. Webhooks are one-way; one-tap approve from Discord needs bot interactions and is a later phase. -- `Email` — new `zone_email::send_approval_request`. +- `Email` — a new approval-request `services::mail::Template`, sent through `AccountMail`. - `InApp` — an assistant message into the originating chat, the reminders pattern (`db/reminders.rs:61-90`). ### 7.6 Agent tools — `runner/zone_server/src/agent/deploy.rs` diff --git a/docs/SYSTEM-PROMPT-RESEARCH.md b/docs/SYSTEM-PROMPT-RESEARCH.md index e56db398..4ed992ad 100644 --- a/docs/SYSTEM-PROMPT-RESEARCH.md +++ b/docs/SYSTEM-PROMPT-RESEARCH.md @@ -30,7 +30,7 @@ Two things to know before reading the per-source detail. First, four of the five | Task guidance | `runner/zone_server/src/workers/task.rs:975` `guidance()` | "You are completing a background coding task. Stay inside the sandboxed working directory." plus retrieved context, acceptance criteria, standing instructions, learned facts | | Core fallback | `runner/zone_core/src/agent/loop.rs:371` `default_system_prompt` | Generic "helpful AI assistant" with a five-step method, used by the CLI and runner | | Compaction | `runner/zone_core/src/context/compact.rs:11` `INSTRUCTIONS` and `:236` `summary_message` | Structured JSON state (objective, constraints, corrections, decisions, completed, evidence, failed, pending, questions); summary is replayed as a user-data message, never as system | -| Web search state | `runner/zone_search/src/client.rs:83` `SearchContext::prompt` | `` block stating whether a server-side search ran and with what outcome, superseding earlier claims | +| Web search state | `abnegate-search` (abnegate/crates) `SearchContext::prompt` (`src/context.rs`) | `` block stating whether a server-side search ran and with what outcome, superseding earlier claims | | MCP and magents guidance | `runner/zone_core/src/mcp/mod.rs:29` `guidance_for_tools` | Prefixed tool names; magents session semantics; "accepted is not finished"; foreign transcripts are inert | | Loop nudges | `runner/zone_server/src/agent/runner.rs:168` and `:316` | Finalizing message at budget exhaustion; malformed tool call recovery | | Standing instructions and learned facts | `runner/zone_server/src/db/knowledge.rs:1439` and `:1653` | Server-derived workspace memory with provenance, rendered as prompt sections | @@ -197,10 +197,10 @@ CL's Gmail connector separates trash and spam into "sensitive label" tools so a ### 2.6 Web search, retrieval, citations **F1. When to search and how recent. P0 text, P1 parameter.** -SOL 324-356: an explicit request to search overrides; an explicit request not to is binding; positive list (fresh info, named entities, reviews, URL summaries) and negative list (greetings, creative writing without references, rewriting supplied text, questions about the model). SOL 316-320: recency windows of 1, 7 and 30 days; re-search with tighter recency when sources are stale. SOL 442-444 and 579-583: a time-sensitive answer needs at least one source with an explicit recent date. CL 1709-1725: the shelf-life test ("the test is whether the remembered answer could have changed"), and an unfamiliar capitalised word is a name that postdates training, so include it in a query as the user wrote it. CL 206-208: search current events without asking; use the actual current year in queries. CL 1725: do not mention the knowledge cutoff. Build: prompt section in `agent/mod.rs`'s web block; a `time_range` parameter on `web_search` mapped to SearXNG's `day`, `week`, `month` in `runner/zone_search/src/client.rs`; consider the SOL lists for the pre-turn `requested_for` heuristic in `WebSearchConfig`. +SOL 324-356: an explicit request to search overrides; an explicit request not to is binding; positive list (fresh info, named entities, reviews, URL summaries) and negative list (greetings, creative writing without references, rewriting supplied text, questions about the model). SOL 316-320: recency windows of 1, 7 and 30 days; re-search with tighter recency when sources are stale. SOL 442-444 and 579-583: a time-sensitive answer needs at least one source with an explicit recent date. CL 1709-1725: the shelf-life test ("the test is whether the remembered answer could have changed"), and an unfamiliar capitalised word is a name that postdates training, so include it in a query as the user wrote it. CL 206-208: search current events without asking; use the actual current year in queries. CL 1725: do not mention the knowledge cutoff. Build: prompt section in `agent/mod.rs`'s web block; a `time_range` parameter on `web_search` mapped to SearXNG's `day`, `week`, `month` in `abnegate-search` (abnegate/crates, `src/query.rs`); consider the SOL lists for the pre-turn `requested_for` heuristic in `WebSearchConfig`. **F1 follow-up. The published date never reaches the model. P1.** -`time_range` shipped in PR 3 and narrows the SearXNG request, but the date on the way back is discarded: `SearxngResult` in `runner/zone_search/src/client.rs` deserializes only `title`, `url` and `content`, `SearchHit` carries only those three, and `format_search_context` renders title, URL and snippet. SearXNG returns `publishedDate` for most engines, so this is thrown away rather than unavailable. Two consequences, both of which undercut what PR 3 just built. The model cannot tell whether a recency-narrowed search actually worked, because a `time_range: day` result set looks exactly like an unnarrowed one, so it cannot judge whether to narrow further or give up. And it cannot follow two rules the shipped prompt already gives it: "Time-sensitive answers need a source explicitly dated recently" has no date to read, and "Drop a source you are unsure of rather than hedge it" becomes a choice between dropping everything and dropping nothing. Carry `publishedDate` through `SearxngResult` into `SearchHit` and render it in `format_search_context`. +`time_range` shipped in PR 3 and narrows the SearXNG request, but the date on the way back is discarded: `Entry` in `abnegate-search` (abnegate/crates, `src/entry.rs`) deserializes only `title`, `url` and `content`, `SearchHit` carries only those three, and `format_search_context` renders title, URL and snippet. SearXNG returns `publishedDate` for most engines, so this is thrown away rather than unavailable. Two consequences, both of which undercut what PR 3 just built. The model cannot tell whether a recency-narrowed search actually worked, because a `time_range: day` result set looks exactly like an unnarrowed one, so it cannot judge whether to narrow further or give up. And it cannot follow two rules the shipped prompt already gives it: "Time-sensitive answers need a source explicitly dated recently" has no date to read, and "Drop a source you are unsure of rather than hedge it" becomes a choice between dropping everything and dropping nothing. Carry `publishedDate` through `Entry` into `SearchHit` and render it in `format_search_context`. **F2. Typed citation IDs. P1.** GK 758-765: cite only through IDs the tools returned (`[web:id]`, `[post:id]`, `[collection:id]`), placed after the final punctuation of the sentence or cell; structured-data tools are exempt. SOL 422-450: typed reference IDs shared by citations and rich UI; never write raw markdown links; one citation block per paragraph. CL 7728: claims in your own words, citation tags are attribution not permission to quote. Build: tag every `web_search`, `search_knowledge`, `read_document` and `search_chat_history` hit with a stable ID in the tool output, instruct the model to cite `[web:3]`, and resolve IDs server-side into `Citation` records in `citations.rs`, rejecting unknown IDs. Server-side resolution only closes the path it owns: the console renders assistant text through react-markdown, so a raw markdown link or a bare autolink in the reply reaches the user without ever passing the citation check. Carry SOL 449 into both halves. Prompt: never write a markdown link or a bare URL, cite by ID. Renderer: `MessageContent` rejects or de-links any anchor whose target does not resolve to a `Citation` on that message. Without the renderer half this item makes invented URLs look validated rather than making them impossible. `query_prometheus`, `get_build_status` and the other structured tools stay citation-exempt (GK 762). @@ -251,7 +251,7 @@ SOL 707-734 is a complete contract: the schedule is an iCal `VEVENT` with `RRULE CC 1688-1820 (Monitor: each stdout line is an event; "silence is not success" at 1738-1748 with a wrong-and-right grep example; a `gh pr checks` poller at 1717-1726; 200 ms batching; auto-stop on flood). CC 888 (a background command re-invokes the loop on exit; foreground sleep is blocked). CC 2107-2162 (self-paced wakeups sized to what is being waited for; `noop` ticks collapsed in the UI). Zone's `tail_task_log`, `get_task_run` and `get_build_status` force the model to poll and spend rounds. Build: `wait_for(target, timeout)` covering task-run completion, GitHub check completion and background job exit, implemented server-side, suspending the loop without consuming an iteration. **I3. Notification cost rule. P0 when a notify tool is exposed.** -CC 1875-1906: err toward not sending; one line under 200 characters; lead with what the user would act on; skip when the user is present. `zone_notify` exists; apply when it becomes a tool. +CC 1875-1906: err toward not sending; one line under 200 characters; lead with what the user would act on; skip when the user is present. `workers::notify` (abnegate-notify) exists; apply when it becomes a tool. ### 2.10 Git and task specifics @@ -326,7 +326,7 @@ GK 29 allows disclosure on explicit request. Recommend the same: an operator-own | C8 | Testing proportionality for tasks | P0 | task guidance | CX 125-126 | | D1 | Final message rules and anti-tic bans | P0 | prompt builder | CC 86-99, CL 114-126, SOL 84, CX 41-43, 81, 100 | | D4 | Narration cadence | P0 | prompt builder | CC 32, CL 130 | -| F1 | When to search, recency rules | P0 text, P1 parameter | web section; `time_range` in `zone_search` | SOL 316-356, 442-444, 579-583, CL 206-208, 1709-1725 | +| F1 | When to search, recency rules | P0 text, P1 parameter | web section; `time_range` in `abnegate-search` | SOL 316-356, 442-444, 579-583, CL 206-208, 1709-1725 | | F4 | Trusting results | P0 | web section | CL 1963, 1781, SOL 561 | | F5 | Document retrieval discipline | P0 | documents section | SOL 1480-1487, 1538, 1559-1565, 1653-1655 | | G2 | Past-chat provenance | P0 | evidence section | CL 1342-1372, 1359 | diff --git a/manager/frontend/e2e/chats-context.e2e.ts b/manager/frontend/e2e/chats-context.e2e.ts index 7d6db240..0c1635bf 100644 --- a/manager/frontend/e2e/chats-context.e2e.ts +++ b/manager/frontend/e2e/chats-context.e2e.ts @@ -72,7 +72,7 @@ for (const width of [1280, 390]) { await expect(page.locator('.chat-item')).toHaveCount(1); await page.locator('.chat-item').click(); const meter = page.getByRole('button', { name: /^Context / }); - await expect(meter).toContainText('9%'); + await expect(meter).toContainText('3.1k / 32K'); await page.locator('.message-form textarea').fill('Continue from the previous results.'); await expect.poll(() => draft).toContain('Continue'); await mkdir('/tmp/zone-context-ui-integration-artifacts', { recursive: true }); @@ -188,7 +188,7 @@ test('empty chat previews and context updates survive adversarial frame ordering await page.locator('.chat-item').click(); const meter = page.getByRole('button', { name: /^Context / }); const input = page.locator('.message-form textarea'); - await expect(meter).toContainText('9%'); + await expect(meter).toContainText('3.1k / 32K'); await meter.focus(); await page.keyboard.press('Enter'); await expect(page.getByRole('region', { name: 'Context usage details' })).toBeVisible(); @@ -197,6 +197,8 @@ test('empty chat previews and context updates survive adversarial frame ordering const forwardTab = browserName === 'webkit' ? 'Alt+Tab' : 'Tab'; const backTab = browserName === 'webkit' ? 'Alt+Shift+Tab' : 'Shift+Tab'; await page.keyboard.press(backTab); + await expect(page.getByRole('button', { name: /^Sources:/ })).toBeFocused(); + await page.keyboard.press(backTab); await expect(input).toBeFocused(); await page.keyboard.press(backTab); await expect(page.getByRole('button', { name: 'Attach files' })).toBeFocused(); @@ -213,10 +215,10 @@ test('empty chat previews and context updates survive adversarial frame ordering breakdown: { ...initial.breakdown, conversation: 2200 }, }; await socket.emit({ type: 'context', chat_id: chat.id, message_id: null, usage }); - await expect(meter).toContainText('12%'); + await expect(meter).toContainText('4.1k / 32K'); hold = false; await (held as unknown as Route).fulfill({ json: { context: initial } }); - await expect(meter).toContainText('12%'); + await expect(meter).toContainText('4.1k / 32K'); let generation = 'first'; socket.setOnSend(async () => { await socket.emit({ type: 'message_start', message_id: generation, role: 'assistant' }); @@ -290,7 +292,8 @@ test('empty chat previews and context updates survive adversarial frame ordering usage: { ...initial, status: 'compacting' }, }); await socket.disconnect(); - await expect(meter).toContainText('Unavailable'); + await expect(meter).not.toContainText('Compacting'); + await expect(meter).toContainText('3.1k / 32K'); await expect( page.getByText(/Connection interrupted during generation.*last observation/) ).toBeInViewport(); @@ -321,8 +324,8 @@ test('empty chat previews and context updates survive adversarial frame ordering hold = false; usage = initial; await (held as unknown as Route).fulfill({ json: { context: initial } }); - await expect(meter).toContainText('9%'); + await expect(meter).toContainText('3.1k / 32K'); await expect(meter).not.toContainText('Unavailable'); await page.reload(); - await expect(meter).toContainText('9%'); + await expect(meter).toContainText('3.1k / 32K'); }); diff --git a/manager/frontend/e2e/models.e2e.ts b/manager/frontend/e2e/models.e2e.ts index 7c52428b..3ec1e431 100644 --- a/manager/frontend/e2e/models.e2e.ts +++ b/manager/frontend/e2e/models.e2e.ts @@ -143,44 +143,88 @@ function parseMultipart( return fields; } -function asTrainRequest(request: { +type PostedRequest = { headers: () => Record; postDataJSON: () => unknown; postDataBuffer: () => Buffer | null; -}): TrainRequest { +}; + +type MultipartFields = Map; + +function multipartFields(request: PostedRequest): MultipartFields | null { const contentType = request.headers()['content-type'] ?? ''; - if (!contentType.includes('multipart/form-data')) { - return request.postDataJSON() as TrainRequest; - } + if (!contentType.includes('multipart/form-data')) return null; const buffer = request.postDataBuffer(); - if (!buffer) throw new Error('empty train post'); - const fields = parseMultipart(buffer, contentType); + if (!buffer) throw new Error('empty multipart post'); + return parseMultipart(buffer, contentType); +} + +function multipartImages(fields: MultipartFields): TrainImage[] { const meta = JSON.parse(fields.get('images')?.value.toString('utf8') ?? '[]') as Array<{ filename: string; caption: string; }>; + return meta.map((image, index) => { + const before = fields.get(`before_${index}`)?.value.toString('base64'); + return { + filename: image.filename, + caption: image.caption, + bytes_base64: fields.get(`image_${index}`)?.value.toString('base64') ?? '', + ...(before ? { before_base64: before } : {}), + }; + }); +} + +function postedImages(request: PostedRequest): TrainImage[] { + const fields = multipartFields(request); + if (!fields) return (request.postDataJSON() as { images: TrainImage[] }).images; + return multipartImages(fields); +} + +const stagedUploads = new WeakMap>(); +const UPLOAD_PATH = /^\/api\/models\/train\/uploads\/([^/]+)$/; +const TRAIN_URL = /\/api\/models\/train(\?.*)?$/; + +/** The images the console staged under each training upload id on `page`. */ +function staged(page: Page): Map { + const known = stagedUploads.get(page); + if (known) return known; + const uploads = new Map(); + stagedUploads.set(page, uploads); + page.on('request', (request) => { + const id = UPLOAD_PATH.exec(new URL(request.url()).pathname)?.[1]; + if (request.method() !== 'POST' || !id) return; + uploads.set(id, [...(uploads.get(id) ?? []), ...postedImages(request)]); + }); + return uploads; +} + +function asTrainRequest(request: PostedRequest, uploads: Map): TrainRequest { + const fields = multipartFields(request); + if (!fields) return request.postDataJSON() as TrainRequest; + const upload = fields.get('upload_id')?.value.toString('utf8'); return { name: fields.get('name')?.value.toString('utf8') ?? '', base: fields.get('base')?.value.toString('utf8') ?? '', trigger: fields.get('trigger')?.value.toString('utf8') || undefined, - images: meta.map((image, index) => { - const before = fields.get(`before_${index}`)?.value.toString('base64'); - return { - filename: image.filename, - caption: image.caption, - bytes_base64: fields.get(`image_${index}`)?.value.toString('base64') ?? '', - ...(before ? { before_base64: before } : {}), - }; - }), + images: upload ? (uploads.get(upload) ?? []) : multipartImages(fields), }; } +/** WebKit reports a multipart body without its files' bytes, so only other engines show them. */ +function uploaded(image: Partial, browserName: string): Partial { + if (browserName !== 'webkit') return image; + const { bytes_base64: _bytes, before_base64: _before, ...described } = image; + return described; +} + function trainRequests(page: Page): TrainRequest[] { + const uploads = staged(page); const requests: TrainRequest[] = []; page.on('request', (request) => { if (request.method() !== 'POST') return; if (new URL(request.url()).pathname !== '/api/models/train') return; - requests.push(asTrainRequest(request)); + requests.push(asTrainRequest(request, uploads)); }); return requests; } @@ -189,6 +233,8 @@ function trainRequests(page: Page): TrainRequest[] { async function setupModelsRoutes(page: Page, options?: { browseModels?: typeof mockBrowseModels; installedModels?: InstalledModel[] }): Promise { const installedModels = [...(options?.installedModels ?? mockInstalledModels)]; const browseModels = options?.browseModels ?? mockBrowseModels; + const uploads = staged(page); + let nextUpload = 0; // Use glob pattern that matches any URL containing /api/models await routeApi(page, '**/api/models**', (route) => { @@ -244,16 +290,24 @@ async function setupModelsRoutes(page: Page, options?: { browseModels?: typeof m }); } } else if (method === 'POST' && url.includes('/api/models/train/captions')) { - const body = route.request().postDataJSON() as { images: TrainImage[] }; route.fulfill({ status: 200, contentType: 'application/json', body: JSON.stringify({ - captions: body.images.map((_, index) => `a photo, frame ${index + 1}`), + captions: postedImages(route.request()).map((_, index) => `a photo, frame ${index + 1}`), }), }); + } else if (method === 'POST' && new URL(url).pathname === '/api/models/train/uploads') { + nextUpload += 1; + route.fulfill({ + status: 200, + contentType: 'application/json', + body: JSON.stringify({ id: `upload-${nextUpload}` }), + }); + } else if (method === 'POST' && UPLOAD_PATH.test(new URL(url).pathname)) { + route.fulfill({ status: 200, contentType: 'application/json', body: '{}' }); } else if (method === 'POST' && url.includes('/api/models/train')) { - const body = route.request().postDataJSON() as TrainRequest; + const body = asTrainRequest(route.request(), uploads); const adapter = trainedAdapter(body.name); installedModels.push(adapter); route.fulfill({ @@ -322,7 +376,7 @@ function trainPanel(page: Page): Locator { } async function routeTrainResult(page: Page, result: TrainResult): Promise { - await routeApi(page, '**/api/models/train', (route) => { + await routeApi(page, TRAIN_URL, (route) => { if (route.request().method() === 'GET') { route.fulfill({ status: 204, body: '' }); return; @@ -471,7 +525,7 @@ test.describe('Models Page', () => { ).toBeVisible(); }); - test('submits a LoRA and lists the trained adapter', async ({ page }) => { + test('submits a LoRA and lists the trained adapter', async ({ browserName, page }) => { const requests = trainRequests(page); const panel = trainPanel(page); @@ -496,8 +550,9 @@ test.describe('Models Page', () => { expect(body.base).toBe('flux-schnell'); expect(body.trigger).toBe('zne person'); expect(body.images).toHaveLength(1); - expect(body.images[0].filename).toBe('portrait.png'); - expect(body.images[0].bytes_base64).toBe(PNG_BASE64); + expect(body.images[0]).toMatchObject( + uploaded({ filename: 'portrait.png', bytes_base64: PNG_BASE64 }, browserName) + ); await expect(panel.getByLabel('Name', { exact: true })).toHaveValue(''); await expect(panel.getByLabel('Caption for portrait.png')).toHaveCount(0); @@ -533,6 +588,7 @@ test.describe('Models Page', () => { }); test('rejects incomplete Qwen pairs and submits complete pairs in their visible order', async ({ + browserName, page, }) => { const requests = trainRequests(page); @@ -597,7 +653,7 @@ test.describe('Models Page', () => { await expect.poll(() => requests.length).toBe(1); expect(requests[0].base).toBe('qwen-image-edit'); expect(requests[0].trigger).toBeUndefined(); - expect(requests[0].images).toEqual([ + const pairs = [ { filename: 'target-b.png', caption: 'move the subject outside', @@ -610,7 +666,10 @@ test.describe('Models Page', () => { bytes_base64: Buffer.from('target a').toString('base64'), before_base64: Buffer.from('reference a').toString('base64'), }, - ]); + ]; + expect(requests[0].images.map((image) => uploaded(image, browserName))).toEqual( + pairs.map((image) => uploaded(image, browserName)) + ); }); test('clears Qwen references and instructions when switching bases', async ({ page }) => { @@ -659,8 +718,7 @@ test.describe('Models Page', () => { const requested = deferred(); const release = deferred(); await routeApi(page, '**/api/models/train/captions', async (route) => { - const body = route.request().postDataJSON() as { images: TrainImage[] }; - requested.resolve(body.images); + requested.resolve(postedImages(route.request())); await release.promise; await route.fulfill({ status: 200, @@ -702,12 +760,12 @@ test.describe('Models Page', () => { test('freezes every draft control while a training request is in flight', async ({ page }) => { const requested = deferred(); const release = deferred(); - await routeApi(page, '**/api/models/train', async (route) => { + await routeApi(page, TRAIN_URL, async (route) => { if (route.request().method() === 'GET') { await route.fulfill({ status: 204, body: '' }); return; } - requested.resolve(route.request().postDataJSON() as TrainRequest); + requested.resolve(asTrainRequest(route.request(), staged(page))); await release.promise; await route.fulfill({ status: 200, diff --git a/manager/frontend/e2e/screenshots.e2e.ts b/manager/frontend/e2e/screenshots.e2e.ts index f2eb826a..9c34f58e 100644 --- a/manager/frontend/e2e/screenshots.e2e.ts +++ b/manager/frontend/e2e/screenshots.e2e.ts @@ -403,7 +403,14 @@ describeScreenshots('Screenshots - Populated States', () => { body: JSON.stringify([{ id: 'qwen-image-edit', label: 'Qwen Image Edit', edit: true }]), }); }); - await routeApi(page, '**/api/models/train', async (route) => { + await routeApi(page, /\/api\/models\/train\/uploads(\/[^/?]+)?$/, (route) => { + route.fulfill({ + status: 200, + contentType: 'application/json', + body: JSON.stringify({ id: 'studio-upload' }), + }); + }); + await routeApi(page, /\/api\/models\/train(\?.*)?$/, async (route) => { if (route.request().method() === 'GET') { await route.fulfill({ status: 204, body: '' }); return; diff --git a/runner/Cargo.lock b/runner/Cargo.lock index d51b8848..90e0eafa 100644 --- a/runner/Cargo.lock +++ b/runner/Cargo.lock @@ -65,6 +65,40 @@ dependencies = [ "thiserror 2.0.20", ] +[[package]] +name = "abnegate-notify" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "541104394c2a22febfab31f003d93e073a49684afbbcabbc32eb5a27eba4adc8" +dependencies = [ + "abnegate-secret", + "async-trait", + "chrono", + "futures", + "lettre", + "reqwest 0.13.5", + "serde", + "serde_json", + "thiserror 2.0.20", + "tokio", + "tracing", + "url", +] + +[[package]] +name = "abnegate-search" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c829bfcceaef95473e00bc169119324f3ead60cbc692eaf4383b3d2f34449151" +dependencies = [ + "abnegate-secret", + "reqwest 0.13.5", + "serde", + "serde_json", + "thiserror 2.0.20", + "urlencoding", +] + [[package]] name = "abnegate-secret" version = "0.1.3" @@ -5420,9 +5454,9 @@ dependencies = [ [[package]] name = "process-wrap" -version = "9.1.0" +version = "10.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2e842efad9119158434d193c6682e2ebee4b44d6ad801d7b349623b3f57cdf55" +checksum = "1f21b97672d2dc848e7b25701ab4535618b92f4861c13cc3f7f7bed52ad3c8da" dependencies = [ "futures", "indexmap 2.14.1", @@ -6028,9 +6062,9 @@ dependencies = [ [[package]] name = "rmcp" -version = "3.2.0" +version = "3.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42b6914fac0be956fe704a38239c3f44a9f841d1b06a5713d2f638065593f5b5" +checksum = "b23c62fe489ac1d401ab32688cfacac3737a8978dc3343e5361464c7724fd3cb" dependencies = [ "chrono", "futures", @@ -6052,9 +6086,9 @@ dependencies = [ [[package]] name = "rmcp-macros" -version = "3.2.0" +version = "3.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cdf1c49bd4d52014b94db0877410db273c2008f01628b0252a2e9460ad9b7fda" +checksum = "cd740c45d66ceb87e5579082abc27bd771665e464e9660a17a048c721b2a6025" dependencies = [ "darling 0.24.1", "proc-macro2", @@ -9798,7 +9832,7 @@ dependencies = [ "async-trait", "base64 0.23.1", "chrono", - "dirs 6.0.0", + "dirs 7.0.0", "fastembed", "futures", "glob", @@ -9850,7 +9884,7 @@ dependencies = [ "async-trait", "chrono", "dashmap", - "dirs 6.0.0", + "dirs 7.0.0", "futures", "nix", "once_cell", @@ -9881,16 +9915,6 @@ dependencies = [ "zone_installer", ] -[[package]] -name = "zone_email" -version = "0.1.0" -dependencies = [ - "lettre", - "thiserror 2.0.20", - "tokio", - "tracing", -] - [[package]] name = "zone_installer" version = "0.1.0" @@ -9898,7 +9922,7 @@ dependencies = [ "axum", "bytes", "crc", - "dirs 6.0.0", + "dirs 7.0.0", "futures", "rand 0.10.2", "reqwest 0.13.5", @@ -9917,26 +9941,6 @@ dependencies = [ "tracing-subscriber", ] -[[package]] -name = "zone_notify" -version = "0.1.0" -dependencies = [ - "abnegate-secret", - "async-trait", - "chrono", - "futures", - "lettre", - "reqwest 0.13.5", - "serde", - "serde_json", - "thiserror 2.0.20", - "tokio", - "tracing", - "url", - "wiremock", - "zone_core", -] - [[package]] name = "zone_runner" version = "0.1.0" @@ -9956,26 +9960,14 @@ dependencies = [ "tracing-subscriber", ] -[[package]] -name = "zone_search" -version = "0.1.0" -dependencies = [ - "reqwest 0.13.5", - "serde", - "serde_json", - "thiserror 2.0.20", - "tokio", - "tracing", - "urlencoding", - "wiremock", -] - [[package]] name = "zone_server" version = "0.1.0" dependencies = [ "abnegate-exec", "abnegate-http", + "abnegate-notify", + "abnegate-search", "abnegate-secret", "aes-gcm", "aiusg", @@ -9997,7 +9989,6 @@ dependencies = [ "hyper-util", "iana-time-zone", "jsonwebtoken 11.0.0", - "lettre", "metrics", "metrics-exporter-prometheus", "nix", @@ -10028,9 +10019,6 @@ dependencies = [ "zone_comfy", "zone_context", "zone_core", - "zone_email", - "zone_notify", - "zone_search", "zone_vcs", ] diff --git a/runner/Cargo.toml b/runner/Cargo.toml index 5b4444a8..623006bd 100644 --- a/runner/Cargo.toml +++ b/runner/Cargo.toml @@ -1,7 +1,7 @@ [workspace] resolver = "2" -members = ["zone_runner", "zone_core", "zone_chat", "zone_comfy", "zone_email", "zone_notify", "zone_search", "zone_vcs", "zone_server", "zone_cli", "zone_installer", "zone_context", "zone_vision", "zone_desktop"] -default-members = ["zone_runner", "zone_core", "zone_chat", "zone_comfy", "zone_email", "zone_notify", "zone_search", "zone_vcs", "zone_server", "zone_cli", "zone_installer", "zone_context", "zone_vision"] +members = ["zone_runner", "zone_core", "zone_chat", "zone_comfy", "zone_vcs", "zone_server", "zone_cli", "zone_installer", "zone_context", "zone_vision", "zone_desktop"] +default-members = ["zone_runner", "zone_core", "zone_chat", "zone_comfy", "zone_vcs", "zone_server", "zone_cli", "zone_installer", "zone_context", "zone_vision"] [workspace.package] version = "0.1.0" @@ -31,7 +31,7 @@ chrono = { version = "=0.4.45", features = ["serde"] } uuid = { version = "=1.26.0", features = ["v4", "serde"] } reqwest = { version = "=0.13.5", features = ["json", "stream"] } async-trait = "=0.1.92" -rmcp = { version = "=3.2.0", default-features = false } +rmcp = { version = "=3.4.0", default-features = false } # Shared crates, published from abnegate/crates. Caret ranges, unlike every # other dependency here, so a patch release of a crate reaches zone on update. @@ -39,6 +39,8 @@ abnegate-config = "0.1" abnegate-exec = "0.1" abnegate-http = "0.1" abnegate-index = "0.1" +abnegate-notify = "0.1" +abnegate-search = "0.1.3" abnegate-secret = "0.1.3" # Server dependencies @@ -55,7 +57,7 @@ metrics = "=0.24.6" metrics-exporter-prometheus = "=0.18.3" # CLI dependencies -dirs = "=6.0.0" +dirs = "=7.0.0" toml = "=1.1.5" dialoguer = "=0.12.0" indicatif = "=0.18.6" @@ -68,3 +70,9 @@ features = ["signal", "process"] lto = true codegen-units = 1 strip = true + +[profile.dev] +debug = "line-tables-only" + +[profile.dev.package."*"] +debug = false diff --git a/runner/README.md b/runner/README.md index 6bdbb43b..6da74783 100644 --- a/runner/README.md +++ b/runner/README.md @@ -121,6 +121,16 @@ cargo test cargo run --bin zone-server ``` +### Debug Info + +Dev and test builds carry line tables for workspace crates and no debuginfo +for dependencies, so backtraces keep file and line while the build cache stays +small. Full debuginfo for workspace crates, when a debugger needs variables: + +```bash +CARGO_PROFILE_DEV_DEBUG=full cargo build -p zone_server +``` + ### sqlx Offline Query Cache `runner/.cargo/config.toml` sets `SQLX_OFFLINE=true`, so every `sqlx::query!` diff --git a/runner/zone_comfy/src/train.rs b/runner/zone_comfy/src/train.rs index 748c0968..83719499 100644 --- a/runner/zone_comfy/src/train.rs +++ b/runner/zone_comfy/src/train.rs @@ -20,6 +20,7 @@ const MIN_WEIGHT_BYTES: usize = 10_000; const MANIFEST_VERSION: u32 = 1; const CANCEL_TIMEOUT: Duration = Duration::from_secs(30); const REQUEST_TIMEOUT: Duration = Duration::from_secs(5); +const TIMED_OUT: &str = "training timed out"; #[derive(Debug, Deserialize)] pub struct TrainConfig { @@ -897,25 +898,17 @@ async fn wait_prompt( let deadline = tokio::time::Instant::now() + timeout; loop { if tokio::time::Instant::now() >= deadline { - let cleanup = cancel_and_wait(client, config, prompt, false).await; - return Err(WaitFailure { - error: TrainError::Failed("training timed out".into()), - cleanup, - }); + return Err(abandon(client, config, prompt, timed_out()).await); } - let history = match history(client, config, prompt, deadline).await { - Ok(history) => history, - Err(error) => { - let cleanup = cancel_and_wait(client, config, prompt, false).await; - return Err(WaitFailure { error, cleanup }); - } + let history = match tokio::time::timeout_at(deadline, history(client, config, prompt)).await + { + Ok(Ok(history)) => history, + Ok(Err(error)) => return Err(abandon(client, config, prompt, error).await), + Err(_) => return Err(abandon(client, config, prompt, timed_out()).await), }; let entry = match exact_history(&history, prompt) { Ok(entry) => entry, - Err(error) => { - let cleanup = cancel_and_wait(client, config, prompt, false).await; - return Err(WaitFailure { error, cleanup }); - } + Err(error) => return Err(abandon(client, config, prompt, error).await), }; if let Some(entry) = entry { match train_prompt_complete(entry) { @@ -964,20 +957,30 @@ async fn prompt_progress( parse_progress(&response.bytes().await.ok()?) } +async fn abandon( + client: &reqwest::Client, + config: &Config, + prompt: Uuid, + error: TrainError, +) -> WaitFailure { + let cleanup = cancel_and_wait(client, config, prompt, false).await; + WaitFailure { error, cleanup } +} + +fn timed_out() -> TrainError { + TrainError::Failed(TIMED_OUT.into()) +} + async fn history( client: &reqwest::Client, config: &Config, prompt: Uuid, - deadline: tokio::time::Instant, ) -> Result { - let timeout = deadline - .saturating_duration_since(tokio::time::Instant::now()) - .min(REQUEST_TIMEOUT); authorize( config, client .get(format!("{}/history/{prompt}", config.base_url)) - .timeout(timeout), + .timeout(REQUEST_TIMEOUT), ) .send() .await @@ -1069,7 +1072,8 @@ async fn cancel_and_wait( if tokio::time::Instant::now() >= deadline { return false; } - if let Ok(history) = history(client, config, prompt, deadline).await + if let Ok(Ok(history)) = + tokio::time::timeout_at(deadline, history(client, config, prompt)).await && let Ok(Some(entry)) = exact_history(&history, prompt) && train_prompt_terminal(entry) { @@ -1284,6 +1288,8 @@ fn authorize(config: &Config, request: reqwest::RequestBuilder) -> reqwest::Requ mod tests { use super::*; use crate::recipe::{Recipe, RecipeCatalog}; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use wiremock::matchers::{method, path, query_param}; use wiremock::{Mock, MockServer, ResponseTemplate}; @@ -1635,26 +1641,65 @@ mod tests { ); } + #[derive(Clone)] + struct RunningUntilCancelled { + prompt: Uuid, + delay: Duration, + cancelled: Arc, + } + + impl RunningUntilCancelled { + async fn mount(server: &MockServer, prompt: Uuid, delay: Duration) { + let running = Self { + prompt, + delay, + cancelled: Default::default(), + }; + Mock::given(method("GET")) + .and(path(format!("/history/{prompt}"))) + .respond_with(running.clone()) + .mount(server) + .await; + Mock::given(method("POST")) + .and(path(format!("/api/jobs/{prompt}/cancel"))) + .respond_with(Cancel(running.cancelled)) + .mount(server) + .await; + } + } + + impl wiremock::Respond for RunningUntilCancelled { + fn respond(&self, _request: &wiremock::Request) -> ResponseTemplate { + if self.cancelled.load(Ordering::SeqCst) { + return ResponseTemplate::new(200).set_body_json(json!({ + self.prompt.to_string(): {"status": {"status_str": "error"}} + })); + } + ResponseTemplate::new(200) + .set_body_json(json!({ + self.prompt.to_string(): {"status": {"status_str": "running"}} + })) + .set_delay(self.delay) + } + } + + struct Cancel(Arc); + + impl wiremock::Respond for Cancel { + fn respond(&self, _request: &wiremock::Request) -> ResponseTemplate { + self.0.store(true, Ordering::SeqCst); + ResponseTemplate::new(200).set_body_json(json!({"cancelled": true})) + } + } + #[tokio::test] async fn a_graph_that_never_finishes_times_out() { let server = MockServer::start().await; let prompt = Uuid::new_v4(); - uploads(&server).await; - queues(&server, json!({"prompt_id": prompt, "number": 1})).await; - Mock::given(method("GET")) - .and(path(format!("/history/{prompt}"))) - .respond_with( - ResponseTemplate::new(200).set_body_json( - json!({prompt.to_string(): {"status": {"status_str": "running"}}}), - ), - ) - .mount(&server) - .await; + RunningUntilCancelled::mount(&server, prompt, Duration::ZERO).await; - let work = dataset(); - let client = reqwest::Client::new(); let failure = wait_prompt( - &client, + &reqwest::Client::new(), &config(&server), prompt, Duration::from_millis(120), @@ -1664,11 +1709,35 @@ mod tests { .await .unwrap_err(); assert!( - matches!(&failure.error, TrainError::Failed(message) if message.contains("timed out")), + matches!(&failure.error, TrainError::Failed(message) if message == TIMED_OUT), "{}", failure.error ); - drop(work); + assert!(failure.cleanup, "the timed-out graph was cancelled"); + } + + #[tokio::test] + async fn a_comfyui_too_slow_to_answer_still_times_out_at_the_deadline() { + let server = MockServer::start().await; + let prompt = Uuid::new_v4(); + RunningUntilCancelled::mount(&server, prompt, Duration::from_secs(30)).await; + + let failure = wait_prompt( + &reqwest::Client::new(), + &config(&server), + prompt, + Duration::from_millis(200), + None, + None, + ) + .await + .unwrap_err(); + assert!( + matches!(&failure.error, TrainError::Failed(message) if message == TIMED_OUT), + "{}", + failure.error + ); + assert!(failure.cleanup, "the timed-out graph was cancelled"); } #[tokio::test] @@ -2182,10 +2251,7 @@ mod tests { let artifact = "zone-lora-progress"; Mock::given(method("GET")) .and(path(format!("/history/{prompt}"))) - .respond_with(RunningThenDone( - std::sync::atomic::AtomicUsize::new(0), - prompt, - )) + .respond_with(RunningThenDone(AtomicUsize::new(0), prompt)) .mount(&server) .await; Mock::given(method("GET")) @@ -2213,11 +2279,11 @@ mod tests { assert_eq!(update, TrainProgress::new(12, 400)); } - struct RunningThenDone(std::sync::atomic::AtomicUsize, Uuid); + struct RunningThenDone(AtomicUsize, Uuid); impl wiremock::Respond for RunningThenDone { fn respond(&self, _request: &wiremock::Request) -> ResponseTemplate { - let seen = self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let seen = self.0.fetch_add(1, Ordering::SeqCst); let status = if seen == 0 { json!({"status_str": "running"}) } else { diff --git a/runner/zone_context/Cargo.toml b/runner/zone_context/Cargo.toml index b03091af..728c8917 100644 --- a/runner/zone_context/Cargo.toml +++ b/runner/zone_context/Cargo.toml @@ -85,4 +85,5 @@ zone_core = { path = "../zone_core" } [dev-dependencies] tokio = { workspace = true, features = ["test-util", "macros"] } tempfile = "=3.27.0" +zone_core = { path = "../zone_core", features = ["test-support"] } wiremock = "=0.6.5" diff --git a/runner/zone_context/src/embeddings/providers/local.rs b/runner/zone_context/src/embeddings/providers/local.rs index 835e2a6c..c4a1514b 100644 --- a/runner/zone_context/src/embeddings/providers/local.rs +++ b/runner/zone_context/src/embeddings/providers/local.rs @@ -274,6 +274,7 @@ impl EmbeddingService for LocalEmbeddingProvider { #[cfg(test)] mod tests { use super::*; + use zone_core::variables::Variables; #[test] fn resolves_ollama_names() { @@ -333,20 +334,16 @@ mod tests { #[test] fn cache_dir_env_override() { - // Serialise env mutation with other tests in this module. - let _g = ENV_LOCK.lock().unwrap(); - // SAFETY: test-only, guarded by ENV_LOCK. - unsafe { std::env::set_var(CACHE_DIR_ENV, "/tmp/zone-embed-test") }; + let mut variables = Variables::lock(); + variables.set(CACHE_DIR_ENV, "/tmp/zone-embed-test"); assert_eq!( LocalEmbeddingProvider::default_cache_dir(), PathBuf::from("/tmp/zone-embed-test") ); - unsafe { std::env::remove_var(CACHE_DIR_ENV) }; + variables.remove(CACHE_DIR_ENV); assert_ne!( LocalEmbeddingProvider::default_cache_dir(), PathBuf::from("/tmp/zone-embed-test") ); } - - static ENV_LOCK: Mutex<()> = Mutex::new(()); } diff --git a/runner/zone_core/Cargo.toml b/runner/zone_core/Cargo.toml index a63dc3b3..0a128973 100644 --- a/runner/zone_core/Cargo.toml +++ b/runner/zone_core/Cargo.toml @@ -9,6 +9,8 @@ rust-version.workspace = true [features] sqlx = ["abnegate-secret/sqlx"] +# Test-only environment guards (`variables`, `vpn::Hold`) for other crates' tests. +test-support = [] [dependencies] tokio = { workspace = true } @@ -27,7 +29,7 @@ futures = { workspace = true } nix = { workspace = true, features = ["fs"] } # Session storage -dirs = "=6.0.0" +dirs = { workspace = true } abnegate-exec = { workspace = true } abnegate-secret = { workspace = true } diff --git a/runner/zone_core/src/lib.rs b/runner/zone_core/src/lib.rs index 6cd53303..0ef0d6e4 100644 --- a/runner/zone_core/src/lib.rs +++ b/runner/zone_core/src/lib.rs @@ -18,6 +18,8 @@ pub mod mcp; pub mod session; pub mod tools; pub mod types; +#[cfg(any(test, feature = "test-support"))] +pub mod variables; pub mod vpn; // Re-export commonly used types diff --git a/runner/zone_core/src/mcp/client.rs b/runner/zone_core/src/mcp/client.rs index 12378680..002c71fc 100644 --- a/runner/zone_core/src/mcp/client.rs +++ b/runner/zone_core/src/mcp/client.rs @@ -230,7 +230,12 @@ mod tests { use crate::mcp::McpServerSpec; use crate::tools::{Tier, ToolContext, ToolRegistry}; use rmcp::handler::server::wrapper::Parameters; - use rmcp::{ServerHandler, ServiceExt, schemars, tool, tool_handler, tool_router}; + use rmcp::model::{ + CancelledNotification, CancelledNotificationParam, ClientJsonRpcMessage, ClientRequest, + InitializeResult, JsonObject, ListToolsResult, ServerJsonRpcMessage, + }; + use rmcp::transport::{IntoTransport, Transport}; + use rmcp::{RoleServer, ServerHandler, ServiceExt, schemars, tool, tool_handler, tool_router}; use serde::Deserialize; use std::collections::HashMap; use std::path::PathBuf; @@ -384,6 +389,85 @@ mod tests { server_task.abort(); } + #[tokio::test] + async fn a_call_the_server_cancels_and_never_answers_times_out() { + let (server_stream, client_stream) = tokio::io::duplex(64 * 1024); + + let server_task = tokio::spawn(async move { + let mut server = IntoTransport::::into_transport(server_stream); + while let Some(message) = server.receive().await { + let ClientJsonRpcMessage::Request(request) = message else { + continue; + }; + let reply = match request.request { + ClientRequest::InitializeRequest(_) => ServerJsonRpcMessage::response( + InitializeResult::default().into(), + request.id, + ), + ClientRequest::ListToolsRequest(_) => ServerJsonRpcMessage::response( + ListToolsResult::with_all_items(vec![Tool::new( + "hang", + "Cancels every call and never answers", + JsonObject::new(), + )]) + .into(), + request.id, + ), + ClientRequest::CallToolRequest(_) => ServerJsonRpcMessage::notification( + CancelledNotification::new(CancelledNotificationParam::new( + Some(request.id), + Some("abandoned".to_string()), + )) + .into(), + ), + _ => continue, + }; + server.send(reply).await.expect("server send"); + } + }); + + let client = ().serve(client_stream).await.expect("client serve"); + let remote_tools = client.list_all_tools().await.expect("list tools"); + let hub = McpHub { + sessions: vec![Arc::new(McpSession { + name: "silent".to_string(), + remote_tools, + client: Mutex::new(client), + })], + }; + let mut registry = ToolRegistry::new(); + assert_eq!(registry.register_mcp(&hub), 1); + + let result = tokio::time::timeout( + Duration::from_secs(30), + registry.execute( + "silent_hang", + serde_json::json!({}), + &ToolContext { + command_timeout: 1, + ..ToolContext::default() + }, + ), + ) + .await + .expect("the command timeout must end a call the server cancelled and never answered") + .expect("execute silent_hang"); + + assert!(!result.success, "{result:?}"); + assert!( + result + .error + .as_deref() + .unwrap_or("") + .contains("timed out after 1 seconds"), + "a server's cancellation no longer ends the call, so the command timeout must: {result:?}" + ); + + drop(registry); + drop(hub); + server_task.abort(); + } + /// A remote method can write files, spend money or message a stranger, and /// nothing Zone can trust says which one it is. It is therefore gated like /// the calls that cannot be taken back, and the card carries the call diff --git a/runner/zone_core/src/tools/command.rs b/runner/zone_core/src/tools/command.rs index 7033e74f..44107437 100644 --- a/runner/zone_core/src/tools/command.rs +++ b/runner/zone_core/src/tools/command.rs @@ -727,7 +727,8 @@ mod tests { #[tokio::test] async fn shelling_tools_give_the_child_only_the_context_environment() { const MARKER: &str = "ZONE_COMMAND_ENVIRONMENT_MARKER"; - unsafe { std::env::set_var(MARKER, "must-not-reach-a-child") }; + let mut variables = crate::variables::Variables::lock(); + variables.set(MARKER, "must-not-reach-a-child"); let mut context = create_test_context(); context.env = HashMap::from([( @@ -743,7 +744,7 @@ mod tests { .execute(json!({"command": "env"}), &context) .await .unwrap(); - unsafe { std::env::remove_var(MARKER) }; + drop(variables); for result in [command, shell] { assert!(result.success, "{result:?}"); diff --git a/runner/zone_core/src/tools/file.rs b/runner/zone_core/src/tools/file.rs index acaadd35..20f5a83e 100644 --- a/runner/zone_core/src/tools/file.rs +++ b/runner/zone_core/src/tools/file.rs @@ -1178,6 +1178,34 @@ mod tests { assert!(!output.contains("Line 4")); } + /// Tool output is sanitized, so the model reads a file without its + /// invisible formatting characters: an emoji's variation selector and a + /// Persian word's zero-width non-joiner are gone from what it sees, and a + /// patch or rewrite built from that text does not carry them. + #[tokio::test] + async fn read_file_returns_text_without_its_invisible_formatting_characters() { + let dir = tempdir().unwrap(); + let content = + "\u{2764}\u{FE0F} \u{645}\u{6CC}\u{200C}\u{62E}\u{648}\u{627}\u{647}\u{645}\n"; + fs::write(dir.path().join("note.txt"), content).unwrap(); + let context = create_test_context(dir.path()); + + let result = ReadFileTool + .execute(serde_json::json!({"path": "note.txt"}), &context) + .await + .unwrap(); + + assert_eq!( + result.output.as_deref(), + Some("\u{2764} \u{645}\u{6CC}\u{62E}\u{648}\u{627}\u{647}\u{645}\n") + ); + assert_eq!( + fs::read_to_string(dir.path().join("note.txt")).unwrap(), + content, + "reading leaves the file itself untouched" + ); + } + #[test] fn page_text_caps_and_continues_by_character() { let content = "α".repeat(10); diff --git a/runner/zone_core/src/tools/mod.rs b/runner/zone_core/src/tools/mod.rs index 38b8542e..f2fc43e0 100644 --- a/runner/zone_core/src/tools/mod.rs +++ b/runner/zone_core/src/tools/mod.rs @@ -10,18 +10,16 @@ mod identity; pub mod job; mod reason; pub mod routing; -mod sanitize; pub mod tail; mod tier; pub use command::*; pub use file::*; pub use reason::{REASON_DESCRIPTION, REASON_PARAM, reason_property}; -pub use sanitize::sanitize; pub use tier::{CONFIRMED_FROM, Tier}; +use abnegate_secret::sanitize_owned; use async_trait::async_trait; -use sanitize::sanitize_owned; use serde::{Deserialize, Serialize}; use serde_json::Value; use std::collections::{HashMap, HashSet}; @@ -179,7 +177,7 @@ pub struct ToolResult { } impl ToolResult { - /// Wrap successful tool output, [`sanitize`]d on the way in. + /// Wrap successful tool output, [`sanitize`](abnegate_secret::sanitize)d on the way in. pub fn success(output: impl Into) -> Self { Self { success: true, @@ -189,7 +187,7 @@ impl ToolResult { } } - /// Wrap a tool failure, [`sanitize`]d on the way in. + /// Wrap a tool failure, [`sanitize`](abnegate_secret::sanitize)d on the way in. pub fn error(error: impl Into) -> Self { Self { success: false, @@ -751,6 +749,7 @@ pub(crate) mod test_support { #[cfg(test)] mod tests { use super::*; + use abnegate_secret::REDACTED; /// A preview a reader approves has to say what will run. `sh -c` runs one /// command per line, so two lines joined by a space showed them a single @@ -805,11 +804,9 @@ mod tests { #[test] fn success_redacts_a_credential_in_the_output() { let result = ToolResult::success("printenv\nGITHUB_TOKEN=ghp_0123456789abcdefghij\n"); - assert_eq!( - result.output.as_deref(), - Some("printenv\nGITHUB_TOKEN=[REDACTED]\n") - ); - assert_eq!(result.to_message(), "printenv\nGITHUB_TOKEN=[REDACTED]\n"); + let redacted = format!("printenv\nGITHUB_TOKEN={REDACTED}\n"); + assert_eq!(result.output.as_deref(), Some(redacted.as_str())); + assert_eq!(result.to_message(), redacted); } #[test] @@ -819,6 +816,45 @@ mod tests { assert_eq!(result.to_message(), "Error: command not found\n"); } + #[test] + fn success_redacts_a_credential_split_by_an_invisible_character() { + let result = ToolResult::success("fatal: ghp_0123\u{200B}456789abcdefghij rejected"); + assert_eq!(result.output, Some(format!("fatal: {REDACTED} rejected"))); + } + + #[test] + fn success_strips_bidirectional_overrides() { + let result = ToolResult::success("access = \u{202E}nimda\u{202C} check"); + assert_eq!(result.output.as_deref(), Some("access = nimda check")); + } + + #[test] + fn error_strips_a_whole_two_byte_escape() { + let result = ToolResult::error("a\u{1b}7b\u{1b}cc"); + assert_eq!(result.error.as_deref(), Some("abc")); + } + + #[test] + fn error_strips_an_escape_with_intermediates() { + let result = ToolResult::error("a\u{1b}(Bb"); + assert_eq!(result.error.as_deref(), Some("ab")); + } + + #[test] + fn error_strips_a_start_of_string_with_its_payload() { + let result = ToolResult::error("a\u{1b}Xhidden\u{1b}\\b"); + assert_eq!(result.error.as_deref(), Some("ab")); + } + + #[test] + fn success_drops_emoji_joiners_and_variation_selectors() { + let result = ToolResult::success("\u{2764}\u{FE0F} \u{1F468}\u{200D}\u{1F469}"); + assert_eq!( + result.output.as_deref(), + Some("\u{2764} \u{1F468}\u{1F469}") + ); + } + #[test] fn to_message_keeps_the_head_and_tail_of_huge_success_output() { let body = format!("HEAD_MARKER{}TAIL_MARKER", "x".repeat(20_000)); diff --git a/runner/zone_core/src/tools/sanitize.rs b/runner/zone_core/src/tools/sanitize.rs deleted file mode 100644 index 46c3918b..00000000 --- a/runner/zone_core/src/tools/sanitize.rs +++ /dev/null @@ -1,248 +0,0 @@ -//! Hygiene for text a tool hands back. -//! -//! Tool output carries fetched pages, command output and build logs, so it is -//! attacker-influenced text on its way into a terminal, a browser and a stored -//! message. Terminal control sequences are stripped first so a credential -//! cannot hide behind one, then credentials are redacted. - -use std::borrow::Cow; - -use abnegate_secret::redact; - -const ESCAPE: u8 = 0x1B; -const BELL: u8 = 0x07; -const BACKSLASH: u8 = b'\\'; -const TAB: u8 = b'\t'; -const LINE_FEED: u8 = b'\n'; -const CARRIAGE_RETURN: u8 = b'\r'; -const DELETE: u8 = 0x7F; -const CONTROL_LEAD: u8 = 0xC2; -const CONTROL_SEQUENCE_INTRODUCER: u8 = 0x9B; - -/// Strip terminal control sequences from `text` and redact any credential. -/// -/// Text with nothing to remove is returned untouched and unallocated. -pub fn sanitize(text: &str) -> Cow<'_, str> { - let stripped = strip_control_sequences(text); - let redacted = match redact(&stripped) { - Cow::Borrowed(_) => None, - Cow::Owned(redacted) => Some(redacted), - }; - match redacted { - Some(redacted) => Cow::Owned(redacted), - None => stripped, - } -} - -pub(crate) fn sanitize_owned(text: String) -> String { - let sanitized = match sanitize(&text) { - Cow::Borrowed(_) => None, - Cow::Owned(sanitized) => Some(sanitized), - }; - sanitized.unwrap_or(text) -} - -fn strip_control_sequences(text: &str) -> Cow<'_, str> { - if !text.bytes().any(needs_inspection) { - return Cow::Borrowed(text); - } - - let bytes = text.as_bytes(); - let mut output = String::with_capacity(text.len()); - let mut index = 0; - - while index < bytes.len() { - match bytes[index] { - ESCAPE => index = escape_sequence(bytes, index), - CARRIAGE_RETURN => { - output.push('\n'); - index += if bytes.get(index + 1) == Some(&LINE_FEED) { - 2 - } else { - 1 - }; - } - TAB | LINE_FEED => { - output.push(char::from(bytes[index])); - index += 1; - } - 0x00..=0x1F | DELETE => index += 1, - CONTROL_LEAD => match bytes.get(index + 1) { - Some(&CONTROL_SEQUENCE_INTRODUCER) => { - index = control_sequence(bytes, index + 2).unwrap_or(index + 2); - } - Some(0x80..=0x9F) => index += 2, - _ => { - output.push_str(&text[index..index + 2]); - index += 2; - } - }, - _ => { - let start = index; - while index < bytes.len() && !needs_inspection(bytes[index]) { - index += 1; - } - output.push_str(&text[start..index]); - } - } - } - - Cow::Owned(output) -} - -fn needs_inspection(byte: u8) -> bool { - matches!(byte, 0x00..=0x08 | 0x0B..=0x1F | DELETE | CONTROL_LEAD) -} - -/// The end of the sequence introduced by the escape at `index`. -/// -/// An unterminated sequence loses only its introducer, leaving the payload as -/// ordinary text, which is what a terminal would show. -fn escape_sequence(bytes: &[u8], index: usize) -> usize { - match bytes.get(index + 1) { - Some(b']') => operating_system_command(bytes, index + 2).unwrap_or(index + 2), - Some(b'P' | b'^' | b'_') => device_control(bytes, index + 2).unwrap_or(index + 2), - Some(b'[') => control_sequence(bytes, index + 2).unwrap_or(index + 2), - Some(0x40..=0x5F) => index + 2, - _ => index + 1, - } -} - -fn operating_system_command(bytes: &[u8], from: usize) -> Option { - let mut index = from; - while index < bytes.len() { - if bytes[index] == BELL { - return Some(index + 1); - } - if bytes[index] == ESCAPE && bytes.get(index + 1) == Some(&BACKSLASH) { - return Some(index + 2); - } - index += 1; - } - None -} - -fn device_control(bytes: &[u8], from: usize) -> Option { - let mut index = from; - while index < bytes.len() { - if bytes[index] == ESCAPE && bytes.get(index + 1) == Some(&BACKSLASH) { - return Some(index + 2); - } - index += 1; - } - None -} - -fn control_sequence(bytes: &[u8], from: usize) -> Option { - let mut index = from; - while matches!(bytes.get(index), Some(0x30..=0x3F)) { - index += 1; - } - while matches!(bytes.get(index), Some(0x20..=0x2F)) { - index += 1; - } - match bytes.get(index) { - Some(0x40..=0x7E) => Some(index + 1), - _ => None, - } -} - -#[cfg(test)] -mod tests { - use super::*; - use abnegate_secret::REDACTED; - - #[test] - fn leaves_plain_text_alone() { - let text = "Compiling zone_core v0.1.0\n Finished in 4.21s\n"; - assert!(matches!(sanitize(text), Cow::Borrowed(_))); - assert_eq!(sanitize(text), text); - } - - #[test] - fn strips_a_colour_sequence() { - assert_eq!( - sanitize("\u{1b}[31merror\u{1b}[0m: failed"), - "error: failed" - ); - } - - #[test] - fn strips_a_cursor_sequence_with_intermediates() { - assert_eq!( - sanitize("before\u{1b}[?25l\u{1b}[1;2 qafter"), - "beforeafter" - ); - } - - #[test] - fn strips_a_control_sequence_introduced_by_c1() { - assert_eq!(sanitize("before\u{9b}31mafter"), "beforeafter"); - } - - #[test] - fn strips_an_operating_system_command() { - assert_eq!(sanitize("\u{1b}]0;take over the title\u{7}ok"), "ok"); - assert_eq!(sanitize("\u{1b}]8;;https://evil.test\u{1b}\\ok"), "ok"); - } - - #[test] - fn strips_device_control_privacy_and_application_strings() { - assert_eq!(sanitize("a\u{1b}Pq#0;2;0;0;0\u{1b}\\b"), "ab"); - assert_eq!(sanitize("a\u{1b}^private\u{1b}\\b"), "ab"); - assert_eq!(sanitize("a\u{1b}_application\u{1b}\\b"), "ab"); - } - - #[test] - fn strips_a_lone_escape() { - assert_eq!(sanitize("a\u{1b}Mb"), "ab"); - assert_eq!(sanitize("a\u{1b}7b"), "a7b"); - } - - #[test] - fn drops_the_introducer_of_an_unterminated_sequence() { - assert_eq!(sanitize("\u{1b}]0;never terminated"), "0;never terminated"); - assert_eq!(sanitize("\u{1b}[31"), "31"); - } - - #[test] - fn normalises_line_endings() { - assert_eq!(sanitize("one\r\ntwo\rthree\nfour"), "one\ntwo\nthree\nfour"); - } - - #[test] - fn drops_c0_and_c1_controls_but_keeps_tab_and_newline() { - assert_eq!( - sanitize("a\u{0}b\u{8}c\u{b}d\u{c}e\u{1f}f\u{7f}g\u{85}h\ti\nj"), - "abcdefgh\ti\nj" - ); - } - - #[test] - fn keeps_multibyte_text_that_is_not_a_control() { - let text = "résumé £20 中文 🔐"; - assert_eq!(sanitize(text), text); - } - - #[test] - fn redacts_a_credential_hidden_behind_a_control_sequence() { - assert_eq!( - sanitize("token=ghp_0123\u{1b}[0m456789abcdefghij"), - format!("token={REDACTED}") - ); - } - - #[test] - fn redacts_and_strips_together() { - assert_eq!( - sanitize("\u{1b}[31mfatal\u{1b}[0m: sk-0123456789abcdefghij rejected\r\n"), - format!("fatal: {REDACTED} rejected\n") - ); - } - - #[test] - fn sanitize_owned_keeps_the_original_buffer_when_clean() { - let text = String::from("nothing to clean here"); - assert_eq!(sanitize_owned(text.clone()), text); - } -} diff --git a/runner/zone_core/src/variables.rs b/runner/zone_core/src/variables.rs new file mode 100644 index 00000000..0018a306 --- /dev/null +++ b/runner/zone_core/src/variables.rs @@ -0,0 +1,135 @@ +//! Test-only changes to the process environment, made under one process-wide +//! lock and undone before it is released. Compiled only for tests and the +//! `test-support` feature, never into a server or CLI build. +//! +//! `std::env::set_var` and `remove_var` are unsound while another thread reads +//! the environment, and tests run on threads. Every test in the workspace that +//! changes a variable goes through this module (a [`Variables`], a +//! [`crate::vpn::Hold`], which holds one, or [`Variables::install`]), so: +//! +//! - no two changes interleave, whichever module or crate the tests live in; +//! - a test holding a guard reads the values it set, because no other test can +//! change a variable while it holds the lock; +//! - reads through `std::env` on other threads are already serialised against +//! these changes by the standard library's own lock. +//! +//! It cannot cover a read that bypasses `std::env`, such as libc's `getenv` +//! inside `getaddrinfo` or `localtime` on a thread the test did not start. + +use std::env; +use std::ffi::{OsStr, OsString}; +use std::sync::{Mutex, MutexGuard, PoisonError}; + +static LOCK: Mutex<()> = Mutex::new(()); + +/// The process environment, held for this guard's lifetime. Every variable it +/// changes is restored when it drops, while it still holds the lock, so a test +/// that panics leaves nothing behind. +pub struct Variables { + saved: Vec<(&'static str, Option)>, + _lock: MutexGuard<'static, ()>, +} + +impl Variables { + /// Waits for the lock; nothing is changed yet. + pub fn lock() -> Self { + Self { + saved: Vec::new(), + _lock: LOCK.lock().unwrap_or_else(PoisonError::into_inner), + } + } + + /// Waits for the lock and unsets `names`. + pub fn isolated(names: &[&'static str]) -> Self { + let mut variables = Self::lock(); + variables.clear(names); + variables + } + + /// Unsets `names` until the guard drops. + pub fn clear(&mut self, names: &[&'static str]) { + for name in names { + self.remove(name); + } + } + + /// Sets `name` until the guard drops. + pub fn set(&mut self, name: &'static str, value: impl AsRef) { + self.save(name); + // SAFETY: this guard holds LOCK, and every test change to the + // environment takes it (see the module documentation for the limits). + unsafe { env::set_var(name, value) }; + } + + /// Unsets `name` until the guard drops. + pub fn remove(&mut self, name: &'static str) { + self.save(name); + // SAFETY: this guard holds LOCK, and every test change to the + // environment takes it (see the module documentation for the limits). + unsafe { env::remove_var(name) }; + } + + /// Sets `name` for the rest of the process, under the lock, without + /// restoring it: for a value a whole test binary shares, set once before + /// the tests that read it. + pub fn install(name: &'static str, value: impl AsRef) { + let _variables = Self::lock(); + // SAFETY: `_variables` holds LOCK, and every test change to the + // environment takes it (see the module documentation for the limits). + unsafe { env::set_var(name, value) }; + } + + fn save(&mut self, name: &'static str) { + if self.saved.iter().all(|(saved, _)| *saved != name) { + self.saved.push((name, env::var_os(name))); + } + } +} + +impl Drop for Variables { + fn drop(&mut self) { + for (name, value) in self.saved.drain(..).rev() { + // SAFETY: the lock is released only after this restore, when the + // guard's fields drop. + unsafe { + match value { + Some(value) => env::set_var(name, value), + None => env::remove_var(name), + } + } + } + } +} + +#[cfg(test)] +pub(crate) fn held() -> bool { + LOCK.try_lock().is_err() +} + +#[cfg(test)] +mod tests { + use super::*; + + const CHANGED: &str = "ZONE_VARIABLES_TEST_CHANGED"; + const ADDED: &str = "ZONE_VARIABLES_TEST_ADDED"; + + #[test] + fn every_change_is_undone_to_the_value_before_the_first() { + let mut variables = Variables::isolated(&[ADDED]); + // SAFETY: `variables` holds LOCK. + unsafe { env::set_var(CHANGED, "before") }; + variables.set(CHANGED, "first"); + variables.set(CHANGED, "second"); + variables.remove(CHANGED); + variables.set(ADDED, "added"); + assert!(held()); + drop(variables); + + let variables = Variables::lock(); + assert_eq!(env::var(CHANGED).as_deref(), Ok("before")); + assert_eq!(env::var_os(ADDED), None); + // SAFETY: `variables` holds LOCK. + unsafe { env::remove_var(CHANGED) }; + drop(variables); + } +} diff --git a/runner/zone_core/src/vpn.rs b/runner/zone_core/src/vpn.rs index 3a4640d0..9108f87a 100644 --- a/runner/zone_core/src/vpn.rs +++ b/runner/zone_core/src/vpn.rs @@ -7,8 +7,9 @@ //! `ZONE_VPN_REQUIRED=0` allows the public internet even with credentials. use std::env; -use std::ffi::OsString; -use std::sync::{Mutex, MutexGuard, PoisonError}; + +#[cfg(any(test, feature = "test-support"))] +use crate::variables::Variables; /// Why a public fetch refused to leave. pub const OFFLINE: &str = "Public web access is offline until the VPN is enabled."; @@ -31,7 +32,6 @@ const VPN: &str = "ZONE_VPN"; const REQUIRED: &str = "ZONE_VPN_REQUIRED"; const WIREGUARD: &str = "VPN_WIREGUARD_PRIVATE_KEY"; const OPENVPN: &str = "VPN_OPENVPN_USER"; -const NAMES: [&str; 4] = [VPN, REQUIRED, WIREGUARD, OPENVPN]; fn truthy(name: &str) -> bool { match env::var(name) { @@ -82,18 +82,12 @@ pub fn allows_public() -> bool { enabled() || !required() } -static ENVIRONMENT: Mutex<()> = Mutex::new(()); - -fn lock() -> MutexGuard<'static, ()> { - ENVIRONMENT.lock().unwrap_or_else(PoisonError::into_inner) -} - -/// Holds VPN env vars in a known state for the guard's lifetime. -pub struct Hold { - _lock: MutexGuard<'static, ()>, - previous: [Option; 4], -} +/// Holds VPN env vars in a known state for the guard's lifetime, under the +/// lock every [`Variables`] takes. +#[cfg(any(test, feature = "test-support"))] +pub struct Hold(Variables); +#[cfg(any(test, feature = "test-support"))] impl Hold { /// Tunnel on (`ZONE_VPN=1`). pub fn on() -> Self { @@ -115,51 +109,31 @@ impl Hold { Self::apply(None, None, Some("1"), None) } + /// The environment this hold has locked, for changing more variables + /// under the same lock. + pub fn variables(&mut self) -> &mut Variables { + &mut self.0 + } + fn apply( vpn: Option<&str>, required: Option<&str>, wireguard: Option<&str>, openvpn: Option<&str>, ) -> Self { - let held = lock(); - let previous = [ - env::var_os(VPN), - env::var_os(REQUIRED), - env::var_os(WIREGUARD), - env::var_os(OPENVPN), - ]; - // SAFETY: Hold owns ENVIRONMENT for the mutation and the restore. - unsafe { - for (name, value) in [ - (VPN, vpn), - (REQUIRED, required), - (WIREGUARD, wireguard), - (OPENVPN, openvpn), - ] { - match value { - Some(value) => env::set_var(name, value), - None => env::remove_var(name), - } - } - } - Self { - _lock: held, - previous, - } - } -} - -impl Drop for Hold { - fn drop(&mut self) { - // SAFETY: Hold still owns ENVIRONMENT while restoring. - unsafe { - for (name, previous) in NAMES.iter().zip(self.previous.iter()) { - match previous { - Some(value) => env::set_var(name, value), - None => env::remove_var(name), - } + let mut variables = Variables::lock(); + for (name, value) in [ + (VPN, vpn), + (REQUIRED, required), + (WIREGUARD, wireguard), + (OPENVPN, openvpn), + ] { + match value { + Some(value) => variables.set(name, value), + None => variables.remove(name), } } + Self(variables) } } @@ -169,39 +143,39 @@ mod tests { #[test] fn stays_off_when_unset_or_false() { - let _vpn = Hold::off(); + let mut vpn = Hold::off(); assert!(!enabled()); assert!(!required()); assert!(!configured()); - unsafe { env::set_var(VPN, "0") }; + vpn.variables().set(VPN, "0"); assert!(!enabled()); - unsafe { env::set_var(VPN, "") }; + vpn.variables().set(VPN, ""); assert!(!enabled()); } #[test] fn turns_on_for_truthy_values() { - let _vpn = Hold::on(); + let mut vpn = Hold::on(); assert!(enabled()); for value in ["1", "true", "TRUE", "yes", "on"] { - unsafe { env::set_var(VPN, value) }; + vpn.variables().set(VPN, value); assert!(enabled(), "{value}"); } } #[test] fn public_web_is_allowed_when_the_vpn_is_not_configured() { - let _vpn = Hold::off(); + let mut vpn = Hold::off(); assert!(allows_public()); - unsafe { env::set_var(REQUIRED, "0") }; + vpn.variables().set(REQUIRED, "0"); assert!(allows_public()); } #[test] fn public_web_is_allowed_when_the_tunnel_is_on() { - let _vpn = Hold::required_off(); + let mut vpn = Hold::required_off(); assert!(!allows_public()); - unsafe { env::set_var(VPN, "1") }; + vpn.variables().set(VPN, "1"); assert!(allows_public()); } @@ -215,21 +189,21 @@ mod tests { #[test] fn public_web_stays_offline_when_credentials_are_present_and_the_tunnel_is_down() { - let _vpn = Hold::configured_off(); + let mut vpn = Hold::configured_off(); assert!(configured()); assert!(!enabled()); assert!(required()); assert!(!allows_public()); - unsafe { env::remove_var(WIREGUARD) }; - unsafe { env::set_var(OPENVPN, "user") }; + vpn.variables().remove(WIREGUARD); + vpn.variables().set(OPENVPN, "user"); assert!(configured()); assert!(!allows_public()); } #[test] fn public_web_is_allowed_when_required_is_off_even_with_credentials() { - let _vpn = Hold::configured_off(); - unsafe { env::set_var(REQUIRED, "0") }; + let mut vpn = Hold::configured_off(); + vpn.variables().set(REQUIRED, "0"); assert!(configured()); assert!(!required()); assert!(allows_public()); @@ -248,4 +222,10 @@ mod tests { assert_eq!(refusal(false), Some(OFFLINE)); assert_eq!(refusal(true), Some(CHAT_OFFLINE)); } + + #[test] + fn a_hold_takes_the_lock_every_variables_guard_takes() { + let _vpn = Hold::off(); + assert!(crate::variables::held()); + } } diff --git a/runner/zone_email/Cargo.toml b/runner/zone_email/Cargo.toml deleted file mode 100644 index b80f351e..00000000 --- a/runner/zone_email/Cargo.toml +++ /dev/null @@ -1,14 +0,0 @@ -[package] -name = "zone_email" -description = "Transactional email over SMTP" -version.workspace = true -edition.workspace = true -license.workspace = true -repository.workspace = true -rust-version.workspace = true - -[dependencies] -thiserror = { workspace = true } -tokio = { workspace = true } -tracing = { workspace = true } -lettre = { version = "=0.11.23", features = ["tokio1-native-tls", "builder", "smtp-transport"] } diff --git a/runner/zone_email/src/lib.rs b/runner/zone_email/src/lib.rs deleted file mode 100644 index d21ff21a..00000000 --- a/runner/zone_email/src/lib.rs +++ /dev/null @@ -1,347 +0,0 @@ -//! Transactional email over SMTP. -//! -//! [`EmailService`] sends the verification, password reset, and invitation -//! messages an account flow needs. -//! -//! ```no_run -//! # async fn example() -> Result<(), Box> { -//! use zone_email::EmailService; -//! -//! EmailService::from_env()? -//! .send_verification_email("person@example.com", "Sam", "https://zone.test/verify?t=abc") -//! .await?; -//! # Ok(()) -//! # } -//! ``` - -use lettre::address::AddressError; -use lettre::message::header::ContentType; -use lettre::transport::smtp::authentication::Credentials; -use lettre::{Message, SmtpTransport, Transport}; -use std::env; -use thiserror::Error; - -#[derive(Debug, Error)] -pub enum EmailError { - #[error("Failed to build email: {0}")] - BuildError(#[from] lettre::error::Error), - - #[error("Failed to send email: {0}")] - SendError(#[from] lettre::transport::smtp::Error), - - #[error("Invalid email address: {0}")] - AddressError(#[from] AddressError), - - #[error("Email service not configured")] - NotConfigured, - - #[error("Invalid email configuration: {0}")] - InvalidConfig(String), -} - -pub type EmailResult = Result; - -/// Email service configuration -#[derive(Clone)] -pub struct EmailConfig { - pub smtp_host: String, - pub smtp_port: u16, - pub smtp_user: String, - pub smtp_password: String, - pub from_email: String, - pub from_name: String, -} - -/// Custom Debug implementation that redacts the SMTP password -impl std::fmt::Debug for EmailConfig { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("EmailConfig") - .field("smtp_host", &self.smtp_host) - .field("smtp_port", &self.smtp_port) - .field("smtp_user", &self.smtp_user) - .field("smtp_password", &"[REDACTED]") - .field("from_email", &self.from_email) - .field("from_name", &self.from_name) - .finish() - } -} - -impl EmailConfig { - /// Load configuration from environment variables - pub fn from_env() -> EmailResult { - let smtp_host = env::var("SMTP_HOST").map_err(|_| EmailError::NotConfigured)?; - - let smtp_port = env::var("SMTP_PORT") - .unwrap_or_else(|_| "587".to_string()) - .parse::() - .map_err(|e| EmailError::InvalidConfig(format!("Invalid SMTP_PORT: {}", e)))?; - - let smtp_user = env::var("SMTP_USER").map_err(|_| EmailError::NotConfigured)?; - - let smtp_password = env::var("SMTP_PASSWORD").map_err(|_| EmailError::NotConfigured)?; - - let from_email = env::var("SMTP_FROM").unwrap_or_else(|_| "noreply@zone.app".to_string()); - - let from_name = env::var("SMTP_FROM_NAME").unwrap_or_else(|_| "Zone".to_string()); - - Ok(Self { - smtp_host, - smtp_port, - smtp_user, - smtp_password, - from_email, - from_name, - }) - } -} - -/// Email service for sending transactional emails -pub struct EmailService { - config: EmailConfig, - mailer: SmtpTransport, -} - -impl EmailService { - /// Create a new email service with the given configuration - pub fn new(config: EmailConfig) -> EmailResult { - let creds = Credentials::new(config.smtp_user.clone(), config.smtp_password.clone()); - - let mailer = SmtpTransport::relay(&config.smtp_host) - .map_err(|e| EmailError::InvalidConfig(format!("Invalid SMTP host: {}", e)))? - .port(config.smtp_port) - .credentials(creds) - .build(); - - Ok(Self { config, mailer }) - } - - /// Create email service from environment variables - pub fn from_env() -> EmailResult { - let config = EmailConfig::from_env()?; - Self::new(config) - } - - /// Send an email verification email - pub async fn send_verification_email( - &self, - to_email: &str, - to_name: &str, - verification_url: &str, - ) -> EmailResult<()> { - let subject = "Verify your email address"; - let body = format!( - r#" -Hello {}, - -Thank you for signing up for Zone! - -Please verify your email address by clicking the link below: - -{} - -This link will expire in 24 hours. - -If you did not create an account, you can safely ignore this email. - -Best regards, -The Zone Team - "#, - to_name, verification_url - ); - - self.send_email(to_email, subject, &body).await - } - - /// Send a password reset email - pub async fn send_password_reset_email( - &self, - to_email: &str, - to_name: &str, - reset_url: &str, - ) -> EmailResult<()> { - let subject = "Reset your password"; - let body = format!( - r#" -Hello {}, - -We received a request to reset your password for your Zone account. - -Click the link below to reset your password: - -{} - -This link will expire in 1 hour. - -If you did not request a password reset, you can safely ignore this email. -Your password will not be changed unless you click the link above and create a new password. - -Best regards, -The Zone Team - "#, - to_name, reset_url - ); - - self.send_email(to_email, subject, &body).await - } - - /// Send an invitation email - pub async fn send_invitation_email( - &self, - to_email: &str, - org_name: &str, - inviter_name: &str, - invitation_url: &str, - ) -> EmailResult<()> { - let subject = format!("You've been invited to join {} on Zone", org_name); - let body = format!( - r#" -Hello, - -{} has invited you to join the {} organization on Zone. - -Click the link below to accept the invitation: - -{} - -This link will expire in 7 days. - -If you don't have a Zone account yet, you'll be able to create one after clicking the link. - -Best regards, -The Zone Team - "#, - inviter_name, org_name, invitation_url - ); - - self.send_email(to_email, &subject, &body).await - } - - /// Send a generic email - async fn send_email(&self, to_email: &str, subject: &str, body: &str) -> EmailResult<()> { - let email = Message::builder() - .from(format!("{} <{}>", self.config.from_name, self.config.from_email).parse()?) - .to(to_email.parse()?) - .subject(subject) - .header(ContentType::TEXT_PLAIN) - .body(body.to_string())?; - - // Send email in a blocking task to avoid blocking the async runtime - let mailer = self.mailer.clone(); - tokio::task::spawn_blocking(move || mailer.send(&email)) - .await - .map_err(|e| EmailError::InvalidConfig(format!("Task join error: {}", e)))? - .map_err(EmailError::SendError)?; - - Ok(()) - } -} - -/// Mock email service for testing (doesn't actually send emails) -#[cfg(test)] -pub struct MockEmailService { - pub sent_emails: std::sync::Arc>>, -} - -#[cfg(test)] -#[derive(Debug, Clone)] -pub struct SentEmail { - pub to: String, - pub subject: String, - pub body: String, -} - -#[cfg(test)] -impl Default for MockEmailService { - fn default() -> Self { - Self::new() - } -} - -#[cfg(test)] -impl MockEmailService { - pub fn new() -> Self { - Self { - sent_emails: std::sync::Arc::new(tokio::sync::Mutex::new(Vec::new())), - } - } - - pub async fn send_verification_email( - &self, - to_email: &str, - _to_name: &str, - _verification_url: &str, - ) -> EmailResult<()> { - let mut emails = self.sent_emails.lock().await; - emails.push(SentEmail { - to: to_email.to_string(), - subject: "Verify your email address".to_string(), - body: "Verification email".to_string(), - }); - Ok(()) - } - - pub async fn send_password_reset_email( - &self, - to_email: &str, - _to_name: &str, - _reset_url: &str, - ) -> EmailResult<()> { - let mut emails = self.sent_emails.lock().await; - emails.push(SentEmail { - to: to_email.to_string(), - subject: "Reset your password".to_string(), - body: "Password reset email".to_string(), - }); - Ok(()) - } - - pub async fn get_sent_emails(&self) -> Vec { - self.sent_emails.lock().await.clone() - } - - pub async fn clear(&self) { - self.sent_emails.lock().await.clear(); - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[tokio::test] - async fn test_mock_email_service_tracks_sent_emails() { - let service = MockEmailService::new(); - - service - .send_verification_email("test@example.com", "Test User", "http://example.com/verify") - .await - .expect("Failed to send verification email"); - - service - .send_password_reset_email("test@example.com", "Test User", "http://example.com/reset") - .await - .expect("Failed to send reset email"); - - let emails = service.get_sent_emails().await; - assert_eq!(emails.len(), 2); - assert_eq!(emails[0].to, "test@example.com"); - assert_eq!(emails[0].subject, "Verify your email address"); - assert_eq!(emails[1].subject, "Reset your password"); - } - - #[tokio::test] - async fn test_mock_email_service_can_be_cleared() { - let service = MockEmailService::new(); - - service - .send_verification_email("test@example.com", "Test User", "http://example.com/verify") - .await - .expect("Failed to send email"); - - assert_eq!(service.get_sent_emails().await.len(), 1); - - service.clear().await; - - assert_eq!(service.get_sent_emails().await.len(), 0); - } -} diff --git a/runner/zone_notify/Cargo.toml b/runner/zone_notify/Cargo.toml deleted file mode 100644 index a5bc5dfb..00000000 --- a/runner/zone_notify/Cargo.toml +++ /dev/null @@ -1,34 +0,0 @@ -[package] -name = "zone_notify" -description = "Multi-channel notification fan-out" -version.workspace = true -edition.workspace = true -license.workspace = true -repository.workspace = true -rust-version.workspace = true - -[dependencies] -async-trait = { workspace = true } -chrono = { workspace = true } -futures = { workspace = true } -serde = { workspace = true } -serde_json = { workspace = true } -thiserror = { workspace = true } -tokio = { workspace = true } -tracing = { workspace = true } -url = "=2.5.8" - -reqwest = { workspace = true } -lettre = { version = "=0.11.23", default-features = false, features = [ - "builder", - "smtp-transport", - "hostname", - "tokio1-native-tls", -] } - -abnegate-secret = { workspace = true } -zone_core = { path = "../zone_core" } - -[dev-dependencies] -tokio = { workspace = true, features = ["test-util", "macros"] } -wiremock = "=0.6.5" diff --git a/runner/zone_notify/src/backend/discord.rs b/runner/zone_notify/src/backend/discord.rs deleted file mode 100644 index b3d50e61..00000000 --- a/runner/zone_notify/src/backend/discord.rs +++ /dev/null @@ -1,302 +0,0 @@ -//! Discord, over a channel webhook. - -use async_trait::async_trait; -use chrono::SecondsFormat; -use serde_json::{Value, json}; - -use crate::backend::webhook::Webhook; -use crate::channel::Channel; -use crate::endpoint::Endpoint; -use crate::error::NotifyError; -use crate::fanout::DEFAULT_TIMEOUT; -use crate::notification::Notification; -use crate::notifier::Notifier; -use crate::severity::Severity; -use crate::text::truncate; - -const HOSTS: &[&str] = &[ - "discord.com", - "discordapp.com", - "canary.discord.com", - "ptb.discord.com", -]; -const MAX_TITLE_CHARS: usize = 256; -const MAX_DESCRIPTION_CHARS: usize = 4_096; -const MAX_FIELD_NAME_CHARS: usize = 256; -const MAX_FIELD_VALUE_CHARS: usize = 1_024; -const MAX_FIELDS: usize = 25; -const FOOTER: &str = "Zone"; - -/// Delivers to one Discord channel webhook. -#[derive(Debug)] -pub struct Discord { - webhook: Webhook, - name: Option, -} - -impl Discord { - /// Point at `webhook_url`, which must be a Discord webhook URL. - pub fn new(webhook_url: &str) -> Result { - let endpoint = Endpoint::new(webhook_url, HOSTS)?; - Ok(Self { - webhook: Webhook::new(endpoint, DEFAULT_TIMEOUT)?, - name: None, - }) - } - - /// Label this instance, for a workspace with more than one Discord hook. - #[must_use] - pub fn named(mut self, name: impl Into) -> Self { - self.name = Some(name.into()); - self - } - - #[cfg(test)] - pub(crate) fn at_test_server(base_url: &str) -> Self { - Self { - webhook: Webhook::new( - Endpoint::for_test(&format!("{base_url}/api/webhooks/1/secret")), - DEFAULT_TIMEOUT, - ) - .expect("test client"), - name: None, - } - } - - fn payload(&self, notification: &Notification) -> Value { - let mut embed = json!({ - "title": truncate(notification.title(), MAX_TITLE_CHARS), - "color": colour(notification.kind()), - "footer": { "text": FOOTER }, - "timestamp": notification - .timestamp() - .to_rfc3339_opts(SecondsFormat::Secs, true), - }); - - if !notification.body().is_empty() { - embed["description"] = json!(truncate(notification.body(), MAX_DESCRIPTION_CHARS)); - } - - if let Some(link) = notification.url() { - embed["url"] = json!(link); - } - - if !notification.fields().is_empty() { - let fields: Vec = notification - .fields() - .iter() - .take(MAX_FIELDS) - .map(|field| { - json!({ - "name": truncate(field.name(), MAX_FIELD_NAME_CHARS), - "value": truncate(field.value(), MAX_FIELD_VALUE_CHARS), - "inline": true, - }) - }) - .collect(); - embed["fields"] = json!(fields); - } - - json!({ "embeds": [embed] }) - } -} - -#[async_trait] -impl Notifier for Discord { - fn channel(&self) -> Channel { - Channel::DISCORD - } - - fn name(&self) -> Option<&str> { - self.name.as_deref() - } - - async fn deliver(&self, notification: &Notification) -> Result<(), NotifyError> { - self.webhook.post(&self.payload(notification)).await?; - Ok(()) - } -} - -/// Discord renders an embed's accent bar from a decimal RGB integer. -fn colour(severity: Severity) -> u32 { - match severity { - Severity::Info => 0x3498db, - Severity::Success => 0x2ecc71, - Severity::Warning => 0xf39c12, - Severity::Error => 0xe74c3c, - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::endpoint::EndpointError; - use chrono::{TimeZone, Utc}; - use std::time::Duration; - use wiremock::matchers::{body_json, header, method, path}; - use wiremock::{Mock, MockServer, ResponseTemplate}; - - fn at(server: &MockServer) -> Discord { - Discord::at_test_server(&server.uri()) - } - - fn pinned(title: &str, body: &str) -> Notification { - Notification::new(title, body).at(Utc.with_ymd_and_hms(2026, 9, 8, 12, 0, 0).unwrap()) - } - - #[test] - fn every_discord_host_is_accepted_and_others_are_not() { - for host in HOSTS { - let url = format!("https://{host}/api/webhooks/1/token"); - assert!(Discord::new(&url).is_ok(), "{host} should be allowed"); - } - - let error = Discord::new("https://discord.com.attacker.test/api/webhooks/1/token") - .expect_err("lookalike host must be refused"); - assert!(matches!( - error, - NotifyError::Endpoint(EndpointError::HostNotAllowed { .. }) - )); - } - - #[test] - fn the_webhook_url_never_appears_in_debug() { - let discord = - Discord::new("https://discord.com/api/webhooks/1/xxxxSECRETxxxx").expect("valid"); - let rendered = format!("{discord:?}"); - assert!(!rendered.contains("xxxxSECRETxxxx"), "leaked: {rendered}"); - assert!(rendered.contains("discord.com")); - } - - #[tokio::test] - async fn the_payload_is_a_single_embed() { - let server = MockServer::start().await; - Mock::given(method("POST")) - .and(path("/api/webhooks/1/secret")) - .and(header("content-type", "application/json")) - .and(body_json(json!({ - "embeds": [{ - "title": "Build failed", - "color": 15158332, - "footer": { "text": "Zone" }, - "timestamp": "2026-09-08T12:00:00Z", - "description": "3 tests failed", - "url": "https://zone.test/b/1", - "fields": [ - { "name": "Branch", "value": "main", "inline": true }, - { "name": "Commit", "value": "abc", "inline": true } - ] - }] - }))) - .respond_with(ResponseTemplate::new(204)) - .expect(1) - .mount(&server) - .await; - - let notification = pinned("Build failed", "3 tests failed") - .severity(Severity::Error) - .link("https://zone.test/b/1") - .field("Branch", "main") - .field("Commit", "abc"); - - at(&server).deliver(¬ification).await.expect("delivered"); - } - - #[tokio::test] - async fn optional_parts_are_omitted_rather_than_sent_as_null() { - let server = MockServer::start().await; - Mock::given(method("POST")) - .and(body_json(json!({ - "embeds": [{ - "title": "Heads up", - "color": 3447003, - "footer": { "text": "Zone" }, - "timestamp": "2026-09-08T12:00:00Z", - }] - }))) - .respond_with(ResponseTemplate::new(204)) - .expect(1) - .mount(&server) - .await; - - at(&server) - .deliver(&pinned("Heads up", "")) - .await - .expect("delivered"); - } - - #[test] - fn each_severity_maps_to_its_own_colour() { - assert_eq!(colour(Severity::Info), 3447003); - assert_eq!(colour(Severity::Success), 3066993); - assert_eq!(colour(Severity::Warning), 15965202); - assert_eq!(colour(Severity::Error), 15158332); - } - - #[test] - fn discords_field_ceiling_is_respected() { - let mut notification = pinned("Many", ""); - for index in 0..40 { - notification = notification.field(format!("Key {index}"), "value"); - } - - let discord = Discord::new("https://discord.com/api/webhooks/1/token").expect("valid"); - let payload = discord.payload(¬ification); - assert_eq!( - payload["embeds"][0]["fields"] - .as_array() - .expect("fields") - .len(), - MAX_FIELDS - ); - } - - #[test] - fn over_long_text_is_cut_to_discords_limits() { - let notification = - pinned(&"T".repeat(500), &"B".repeat(9_000)).field("N".repeat(400), "V".repeat(2_000)); - - let discord = Discord::new("https://discord.com/api/webhooks/1/token").expect("valid"); - let embed = discord.payload(¬ification); - let embed = &embed["embeds"][0]; - - assert_eq!( - embed["title"].as_str().unwrap().chars().count(), - MAX_TITLE_CHARS - ); - assert_eq!( - embed["description"].as_str().unwrap().chars().count(), - MAX_DESCRIPTION_CHARS - ); - assert_eq!( - embed["fields"][0]["name"].as_str().unwrap().chars().count(), - MAX_FIELD_NAME_CHARS - ); - assert_eq!( - embed["fields"][0]["value"] - .as_str() - .unwrap() - .chars() - .count(), - MAX_FIELD_VALUE_CHARS - ); - } - - #[tokio::test] - async fn a_rate_limit_is_reported_with_the_wait_discord_asked_for() { - let server = MockServer::start().await; - Mock::given(method("POST")) - .respond_with(ResponseTemplate::new(429).insert_header("retry-after", "2")) - .mount(&server) - .await; - - let error = at(&server) - .deliver(&pinned("Title", "Body")) - .await - .expect_err("rate limited"); - - assert_eq!(error.retry_after(), Some(Duration::from_secs(2))); - assert!(error.is_retryable()); - assert!(!error.to_string().contains("secret")); - } -} diff --git a/runner/zone_notify/src/backend/email.rs b/runner/zone_notify/src/backend/email.rs deleted file mode 100644 index f38d8261..00000000 --- a/runner/zone_notify/src/backend/email.rs +++ /dev/null @@ -1,275 +0,0 @@ -//! Email, over SMTP. - -use std::fmt; - -use async_trait::async_trait; -use lettre::message::Mailbox; -use lettre::message::header::ContentType; -use lettre::transport::smtp::authentication::Credentials; -use lettre::{Address, AsyncSmtpTransport, AsyncTransport, Message, Tokio1Executor}; -use zone_core::tools::sanitize; - -use crate::backend::smtp::SmtpConfig; -use crate::channel::Channel; -use crate::error::NotifyError; -use crate::notification::Notification; -use crate::notifier::Notifier; - -/// Implicit-TLS submissions port, the one port relay() is built for. -const SUBMISSIONS_PORT: u16 = 465; - -/// Delivers to a fixed set of recipients through one SMTP relay. -/// -/// This is a separate transport from `zone_email`, which owns the account -/// flow's three templates and keeps its own synchronous relay. That crate's -/// generic send is private, so a notification of arbitrary content cannot go -/// through it without changing it; see this crate's documentation for what -/// folding the two together would involve. -/// -/// Construct and drop this inside a Tokio runtime. When `lettre`'s `pool` -/// feature is on anywhere in the build, its connection pool spawns a task -/// from its own `Drop`, and a panic in a destructor aborts the process rather -/// than unwinding. -pub struct Email { - transport: AsyncSmtpTransport, - from: Mailbox, - recipients: Vec, - host: String, - name: Option, -} - -impl Email { - /// Connect to the relay in `config` and deliver to `recipients`. - pub fn new(config: SmtpConfig, recipients: &[&str]) -> Result { - if recipients.is_empty() { - return Err(NotifyError::Malformed { - message: "an email channel needs at least one recipient".to_string(), - }); - } - - let from = mailbox(&config.from_address, Some(config.from_name.clone()))?; - let recipients = recipients - .iter() - .map(|recipient| mailbox(recipient, None)) - .collect::, NotifyError>>()?; - - let credentials = - Credentials::new(config.user.clone(), config.password.expose().to_string()); - // relay() sets implicit TLS and port 465. Pointing that at 587, the - // submission port every mainstream relay serves with STARTTLS -- and - // the workspace default -- sends a ClientHello to a plaintext - // listener. starttls_relay refuses to send credentials if the upgrade - // fails, so this stays downgrade-safe. - let builder = if config.port == SUBMISSIONS_PORT { - AsyncSmtpTransport::::relay(&config.host) - } else { - AsyncSmtpTransport::::starttls_relay(&config.host) - }; - let transport = builder - .map_err(|error| NotifyError::Smtp { - host: config.host.clone(), - message: describe(error), - })? - .port(config.port) - .credentials(credentials) - .build(); - - Ok(Self { - transport, - from, - recipients, - host: config.host, - name: None, - }) - } - - /// Label this instance, for a workspace with more than one recipient set. - #[must_use] - pub fn named(mut self, name: impl Into) -> Self { - self.name = Some(name.into()); - self - } - - fn compose(&self, notification: &Notification) -> Result { - let mut builder = Message::builder() - .from(self.from.clone()) - .subject(notification.title()); - - for recipient in &self.recipients { - builder = builder.to(recipient.clone()); - } - - builder - .header(ContentType::TEXT_PLAIN) - .body(notification.to_plain_text()) - .map_err(|error| NotifyError::Malformed { - message: error.to_string(), - }) - } -} - -#[async_trait] -impl Notifier for Email { - fn channel(&self) -> Channel { - Channel::EMAIL - } - - fn name(&self) -> Option<&str> { - self.name.as_deref() - } - - async fn deliver(&self, notification: &Notification) -> Result<(), NotifyError> { - let message = self.compose(notification)?; - self.transport - .send(message) - .await - .map_err(|error| NotifyError::Smtp { - host: self.host.clone(), - message: describe(error), - })?; - Ok(()) - } -} - -/// Written out rather than derived: the transport holds the relay password -/// inside a `lettre::Credentials`, which has no redacting `Debug` of its own. -impl fmt::Debug for Email { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - formatter - .debug_struct("Email") - .field("host", &self.host) - .field("from", &self.from) - .field("recipients", &self.recipients) - .field("name", &self.name) - .finish_non_exhaustive() - } -} - -fn mailbox(address: &str, name: Option) -> Result { - let parsed: Address = address.parse().map_err(|_| NotifyError::Malformed { - message: format!("{address} is not a valid email address"), - })?; - Ok(Mailbox::new(name, parsed)) -} - -/// An SMTP error's message, sanitized. -/// -/// A relay's reply text is written by the far end, so it is treated like any -/// other outside text on its way to a log line. -fn describe(error: lettre::transport::smtp::Error) -> String { - sanitize(&error.to_string()).into_owned() -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::severity::Severity; - use abnegate_secret::SecretValue; - - fn config() -> SmtpConfig { - SmtpConfig { - host: "smtp.zone.test".to_string(), - port: 587, - user: "postmaster".to_string(), - password: SecretValue::new("hunter2-not-a-real-password"), - from_address: "noreply@zone.test".to_string(), - from_name: "Zone".to_string(), - } - } - - fn email() -> Email { - Email::new(config(), &["sam@example.test"]).expect("configured") - } - - fn rendered(notification: &Notification) -> String { - String::from_utf8(email().compose(notification).expect("composed").formatted()) - .expect("utf-8") - } - - #[test] - fn a_channel_with_no_recipients_is_refused() { - let error = Email::new(config(), &[]).expect_err("no recipients"); - assert!(matches!(error, NotifyError::Malformed { .. })); - assert!(error.to_string().contains("at least one recipient")); - } - - #[test] - fn an_unparseable_recipient_is_refused() { - let error = Email::new(config(), &["not an address"]).expect_err("bad recipient"); - assert!(matches!(error, NotifyError::Malformed { .. })); - } - - #[test] - fn an_unparseable_sender_is_refused() { - let mut broken = config(); - broken.from_address = "@@@".to_string(); - assert!(matches!( - Email::new(broken, &["sam@example.test"]).expect_err("bad sender"), - NotifyError::Malformed { .. } - )); - } - - #[tokio::test] - async fn the_subject_is_the_title_and_the_body_carries_everything() { - let message = rendered( - &Notification::new("Build failed", "3 tests failed") - .severity(Severity::Error) - .field("Branch", "main") - .link("https://zone.test/b/1"), - ); - - assert!(message.contains("Subject: Build failed")); - assert!(message.contains("From: Zone ")); - assert!(message.contains("To: sam@example.test")); - assert!(message.contains("Content-Type: text/plain")); - assert!(message.contains("3 tests failed")); - assert!(message.contains("Branch: main")); - assert!(message.contains("https://zone.test/b/1")); - } - - #[tokio::test] - async fn every_recipient_is_addressed() { - let email = - Email::new(config(), &["sam@example.test", "alex@example.test"]).expect("configured"); - let message = String::from_utf8( - email - .compose(&Notification::new("Title", "Body")) - .expect("composed") - .formatted(), - ) - .expect("utf-8"); - - assert!(message.contains("sam@example.test")); - assert!(message.contains("alex@example.test")); - } - - #[tokio::test] - async fn the_relay_password_never_reaches_the_message() { - let message = rendered(&Notification::new("Title", "Body")); - assert!(!message.contains("hunter2"), "leaked into the message"); - } - - #[tokio::test] - async fn a_credential_in_the_body_is_redacted_before_it_is_composed() { - let message = rendered(&Notification::new( - "Deploy log", - "GITHUB_TOKEN=ghp_0123456789abcdefghij", - )); - assert!(!message.contains("ghp_0123456789abcdefghij")); - assert!(message.contains("[REDACTED]")); - } - - #[tokio::test] - async fn the_relay_password_never_appears_in_debug() { - let rendered = format!("{:?}", email()); - assert!(!rendered.contains("hunter2"), "leaked: {rendered}"); - assert!(rendered.contains("smtp.zone.test")); - } - - #[tokio::test] - async fn the_channel_and_name_are_reported() { - assert_eq!(email().channel(), Channel::EMAIL); - assert_eq!(email().name(), None); - assert_eq!(email().named("ops").name(), Some("ops")); - } -} diff --git a/runner/zone_notify/src/backend/mod.rs b/runner/zone_notify/src/backend/mod.rs deleted file mode 100644 index 1a5df0c3..00000000 --- a/runner/zone_notify/src/backend/mod.rs +++ /dev/null @@ -1,15 +0,0 @@ -//! The channels this crate ships with. -//! -//! Each backend is one file implementing [`Notifier`](crate::Notifier). -//! Adding another is the same: a file here, exported below. - -mod discord; -mod email; -mod slack; -mod smtp; -mod webhook; - -pub use discord::Discord; -pub use email::Email; -pub use slack::Slack; -pub use smtp::SmtpConfig; diff --git a/runner/zone_notify/src/backend/slack.rs b/runner/zone_notify/src/backend/slack.rs deleted file mode 100644 index 676b947d..00000000 --- a/runner/zone_notify/src/backend/slack.rs +++ /dev/null @@ -1,338 +0,0 @@ -//! Slack, over an incoming webhook. - -use async_trait::async_trait; -use serde_json::{Value, json}; - -use crate::backend::webhook::Webhook; -use crate::channel::Channel; -use crate::endpoint::Endpoint; -use crate::error::NotifyError; -use crate::fanout::DEFAULT_TIMEOUT; -use crate::notification::Notification; -use crate::notifier::Notifier; -use crate::severity::Severity; -use crate::text::truncate; - -const HOSTS: &[&str] = &["hooks.slack.com"]; -const MAX_HEADER_CHARS: usize = 150; -const MAX_SECTION_CHARS: usize = 3_000; -const MAX_FIELD_CHARS: usize = 2_000; -const MAX_FIELDS_PER_SECTION: usize = 10; - -/// Delivers to one Slack incoming webhook. -#[derive(Debug)] -pub struct Slack { - webhook: Webhook, - name: Option, -} - -impl Slack { - /// Point at `webhook_url`, which must be an `hooks.slack.com` URL. - pub fn new(webhook_url: &str) -> Result { - let endpoint = Endpoint::new(webhook_url, HOSTS)?; - Ok(Self { - webhook: Webhook::new(endpoint, DEFAULT_TIMEOUT)?, - name: None, - }) - } - - /// Label this instance, for a workspace with more than one Slack hook. - #[must_use] - pub fn named(mut self, name: impl Into) -> Self { - self.name = Some(name.into()); - self - } - - #[cfg(test)] - pub(crate) fn at_test_server(base_url: &str) -> Self { - Self { - webhook: Webhook::new( - Endpoint::for_test(&format!("{base_url}/services/T000/B000/secret")), - DEFAULT_TIMEOUT, - ) - .expect("test client"), - name: None, - } - } - - fn payload(&self, notification: &Notification) -> Value { - let mut blocks = vec![json!({ - "type": "header", - "text": { - "type": "plain_text", - "text": truncate( - &format!("{} {}", icon(notification.kind()), escape(notification.title())), - MAX_HEADER_CHARS, - ), - "emoji": true, - } - })]; - - if !notification.body().is_empty() { - blocks.push(json!({ - "type": "section", - "text": { - "type": "mrkdwn", - "text": truncate(&escape(notification.body()), MAX_SECTION_CHARS), - } - })); - } - - for chunk in notification.fields().chunks(MAX_FIELDS_PER_SECTION) { - let fields: Vec = chunk - .iter() - .map(|field| { - json!({ - "type": "mrkdwn", - "text": truncate( - &format!("*{}*\n{}", escape(field.name()), escape(field.value())), - MAX_FIELD_CHARS, - ), - }) - }) - .collect(); - blocks.push(json!({ "type": "section", "fields": fields })); - } - - if let Some(link) = notification.url() { - blocks.push(json!({ - "type": "context", - "elements": [{ "type": "mrkdwn", "text": format!("<{}|Open in Zone>", escape(link)) }], - })); - } - - json!({ - "text": truncate(&escape(¬ification.to_plain_text()), MAX_SECTION_CHARS), - "blocks": blocks, - }) - } -} - -#[async_trait] -impl Notifier for Slack { - fn channel(&self) -> Channel { - Channel::SLACK - } - - fn name(&self) -> Option<&str> { - self.name.as_deref() - } - - async fn deliver(&self, notification: &Notification) -> Result<(), NotifyError> { - self.webhook.post(&self.payload(notification)).await?; - Ok(()) - } -} - -fn icon(severity: Severity) -> &'static str { - match severity { - Severity::Info => ":information_source:", - Severity::Success => ":white_check_mark:", - Severity::Warning => ":warning:", - Severity::Error => ":rotating_light:", - } -} - -/// Slack's three reserved mrkdwn characters, escaped in the order it requires. -/// -/// `&` has to go first or the ampersands introduced by the other two would be -/// escaped a second time. -fn escape(text: &str) -> String { - text.replace('&', "&") - .replace('<', "<") - .replace('>', ">") -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::endpoint::EndpointError; - use wiremock::matchers::{body_json, header, method, path}; - use wiremock::{Mock, MockServer, ResponseTemplate}; - - fn at(server: &MockServer) -> Slack { - Slack::at_test_server(&server.uri()) - } - - #[test] - fn ampersands_are_escaped_before_the_angle_brackets() { - assert_eq!(escape("a & b"), "a & b"); - assert_eq!(escape("