feat(quickstart): z.ai live message check through the Console + next channel

.github/workflows/harness-quickstart.yml

@@ -4,25 +4,33 @@ on:
   schedule:
     - cron: "17 5 * * *"
   workflow_dispatch:
+    inputs:
+      channel:
+        description: Installer channel to validate
+        type: choice
+        options: [main, next]
+        default: main
 
 permissions:
   contents: read
 
 concurrency:
-  group: harness-quickstart
+  group: harness-quickstart-${{ inputs.channel || 'main' }}
   cancel-in-progress: true
 
 jobs:
   validate:
-    name: harness quickstart / latest registry
+    name: harness quickstart / latest registry (${{ inputs.channel || 'main' }})
     runs-on: ubuntu-latest
-    timeout-minutes: 20
+    timeout-minutes: 25
     steps:
       - uses: actions/checkout@v5
 
       - name: Run quickstart validator
         env:
           HARNESS_QUICKSTART_ARTIFACTS_DIR: ${{ github.workspace }}/target/harness-quickstart
+          III_CHANNEL: ${{ inputs.channel || 'main' }}
+          ZAI_API_KEY: ${{ secrets.ZAI_API_KEY }}
         run: harness/tests/quickstart/run-ci.sh
 
       - name: Show quickstart logs
@@ -77,6 +85,7 @@ jobs:
             target/harness-quickstart/EVIDENCE.md
             target/harness-quickstart/timings.tsv
             target/harness-quickstart/cli-version.txt
+            target/harness-quickstart/console-send.json
             target/harness-quickstart/console-status.json
             target/harness-quickstart/config.yaml
             target/harness-quickstart/iii.lock

harness/tests/e2e/README.md

@@ -44,7 +44,7 @@ scenario through one global `allow: ["*"]` policy:
   accuracy, remediation, and clarity.
 - `reactive_automation`: orchestrates three parallel database writers,
   trigger-spawned aggregate reactors, and a single finalizer. The CI stack uses
-  SQLite, so the scenario proves bounded `database::row-change` discovery and
+  SQLite, so the scenario proves bounded `database::row-changed` discovery and
   the namespaced `state`-trigger fallback. The evaluator queries the resulting
   tables, verifies trigger provenance from session metadata, and checks that all
   run-owned triggers were removed.

harness/tests/e2e/run-ci.sh

@@ -137,7 +137,7 @@ start_process database \
   --url "$iii_url" \
   --config "$database_config"
 wait_for_function database::query
-wait_for_trigger_type database::row-change
+wait_for_trigger_type database::row-changed
 
 start_process state \
   "$repo_root/state/target/release/state" \

harness/tests/e2e/src/scenarios/reactive_automation/prompt.rs

@@ -21,7 +21,7 @@ Namespace every other resource and session with `{run_label}`.
 Required execution:
 
 1. Before spawning writers, list the available trigger types and probe
-   `database::row-change` against `{orders}`. Allow at most 30 seconds for the probe, remove its
+   `database::row-changed` against `{orders}`. Allow at most 30 seconds for the probe, remove its
    test row afterward, and record the result.
 
 2. Before spawning writers, register a namespaced `state` trigger targeting `harness::react`.

harness/tests/e2e/src/scenarios/reactive_automation/queries.rs

@@ -56,7 +56,7 @@ pub(super) async fn collect(
         row_change_probe: calls.iter().position(|call| {
             call.function_id == "engine::register_trigger"
                 && call.arguments.get("trigger_type").and_then(Value::as_str)
-                    == Some("database::row-change")
+                    == Some("database::row-changed")
         }),
         state_reaction: calls.iter().position(|call| {
             call.function_id == "engine::register_trigger"

harness/tests/quickstart/README.md

@@ -1,9 +1,8 @@
 # Harness quickstart validator
 
-This validator exercises the published installation path without provider
-credentials or model calls. It runs in an isolated temporary home and project
-directory, installs the `iii` CLI, starts a clean engine, and follows the
-documented commands:
+This validator exercises the published installation path. It runs in an
+isolated temporary home and project directory, installs the `iii` CLI, starts
+a clean engine, and follows the documented commands:
 
 ```bash
 printf 'workers: []\n' > config.yaml
@@ -15,6 +14,17 @@ The check waits for the engine and the core harness/Console function surface,
 calls `console::status`, and fetches the Console HTTP root. It also verifies
 that `config.yaml` and `iii.lock` were produced by the registry install.
 
+When `ZAI_API_KEY` is set, the validator goes one step further: it adds the
+Z.AI provider (`iii worker add provider-zai`), then sends a real message
+through the Console's `/ws` proxy — the same WebSocket path the browser SPA
+uses — via `console_send.py`, and asserts the turn completes with a non-empty
+assistant reply (default model `glm-5.2`, overridable with
+`HARNESS_QUICKSTART_MODEL`/`HARNESS_QUICKSTART_PROVIDER`). Without the key the
+live check is skipped and recorded as such in `result.json` and `EVIDENCE.md`.
+
+Set `III_CHANNEL=next` to validate the `next` installer channel instead of
+`main`; in CI this is exposed as the `channel` input on manual dispatches.
+
 Run it locally with:
 
 ```bash

harness/tests/quickstart/console_send.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python3
+"""Send a message through the Console's /ws proxy and wait for the reply.
+
+Speaks the engine WebSocket protocol (invokefunction/invocationresult)
+through the Console proxy — the same path the browser SPA uses — so a pass
+proves the Console-to-engine-to-provider chain end to end:
+
+  1. harness::send {message, model, provider}
+  2. poll harness::status until the turn completes (fail on failed/cancelled)
+  3. session::messages -> last non-empty assistant text
+
+Prints {"session_id", "turn_id", "reply"} as JSON on stdout; progress and
+errors go to stderr. Requires the `websockets` package.
+"""
+
+import argparse
+import asyncio
+import json
+import sys
+import time
+import uuid
+
+import websockets
+
+
+def eprint(message: str) -> None:
+    print(f"[console-send] {message}", file=sys.stderr, flush=True)
+
+
+async def invoke(ws, function_id: str, data: dict, timeout: float = 30.0):
+    invocation_id = str(uuid.uuid4())
+    await ws.send(
+        json.dumps(
+            {
+                "type": "invokefunction",
+                "invocation_id": invocation_id,
+                "function_id": function_id,
+                "data": data,
+            }
+        )
+    )
+    deadline = time.monotonic() + timeout
+    while True:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise TimeoutError(f"{function_id}: no invocationresult within {timeout}s")
+        frame = json.loads(await asyncio.wait_for(ws.recv(), remaining))
+        # The proxy is transparent, so unrelated engine frames may interleave.
+        if (
+            frame.get("type") != "invocationresult"
+            or frame.get("invocation_id") != invocation_id
+        ):
+            continue
+        if frame.get("error"):
+            raise RuntimeError(f"{function_id} failed: {json.dumps(frame['error'])}")
+        return frame.get("result")
+
+
+def last_assistant_text(messages: list) -> str:
+    texts = []
+    for item in messages:
+        message = item.get("message") or {}
+        if message.get("role") != "assistant":
+            continue
+        for part in message.get("content") or []:
+            if part.get("type") == "text" and (part.get("text") or "").strip():
+                texts.append(part["text"].strip())
+    return texts[-1] if texts else ""
+
+
+async def transcript(ws, session_id: str) -> list:
+    messages, cursor = [], None
+    while True:
+        page = await invoke(
+            ws,
+            "session::messages",
+            {
+                "session_id": session_id,
+                "limit": 500,
+                "cursor": cursor,
+                "include_custom": True,
+            },
+        )
+        messages.extend(page.get("messages") or [])
+        next_cursor = page.get("next_cursor")
+        if not next_cursor or next_cursor == cursor:
+            break
+        cursor = next_cursor
+    return messages
+
+
+async def run(args) -> None:
+    async with websockets.connect(args.url, max_size=None) as ws:
+        send = await invoke(
+            ws,
+            "harness::send",
+            {
+                "message": args.prompt,
+                "model": args.model,
+                "provider": args.provider,
+                "session": {"title": "Harness quickstart validation"},
+                "options": {
+                    "max_turns": 2,
+                    "max_output_tokens": 1024,
+                    "max_total_tokens": 16384,
+                },
+            },
+        )
+        if not send or not send.get("accepted"):
+            raise RuntimeError(f"harness::send was not accepted: {json.dumps(send)}")
+        session_id = send["session_id"]
+        turn_id = send.get("turn_id")
+        eprint(f"turn {turn_id} accepted on session {session_id}")
+
+        deadline = time.monotonic() + args.timeout
+        last_status = "unknown"
+        while True:
+            status = await invoke(ws, "harness::status", {"session_id": session_id})
+            last_status = status.get("status")
+            if last_status in ("failed", "cancelled"):
+                error = status.get("result_error") or "no error was reported"
+                raise RuntimeError(f"turn ended as {last_status}: {error}")
+            if last_status == "completed" and not status.get("expects_wake"):
+                eprint("turn completed")
+                break
+            if time.monotonic() > deadline:
+                raise TimeoutError(
+                    f"turn did not complete within {args.timeout}s "
+                    f"(last status: {last_status})"
+                )
+            await asyncio.sleep(2)
+
+        reply = last_assistant_text(await transcript(ws, session_id))
+        if not reply:
+            raise RuntimeError("no non-empty assistant reply in the session transcript")
+        eprint(f"assistant replied ({len(reply)} chars)")
+        json.dump(
+            {"session_id": session_id, "turn_id": turn_id, "reply": reply},
+            sys.stdout,
+        )
+        print(flush=True)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--url", required=True, help="Console proxy, e.g. ws://127.0.0.1:3113/ws")
+    parser.add_argument("--model", required=True)
+    parser.add_argument("--provider", required=True)
+    parser.add_argument("--prompt", required=True)
+    parser.add_argument("--timeout", type=int, default=240, help="turn completion timeout in seconds")
+    try:
+        asyncio.run(run(parser.parse_args()))
+    except Exception as error:  # surface a single clean line for the validator
+        eprint(f"FAIL: {error}")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()