Verify genuine Google CPU hybrid MCP from Windows archive

tools/windows_agent_smoke.py

@@ -15,7 +15,7 @@ def check(condition, message):
         raise RuntimeError(message)
 
 
-def smoke(cli, work):
+def smoke(cli, work, models=None, worker=None):
     cli = cli.resolve(strict=True)
     work.mkdir(parents=True, exist_ok=False)
     corpus = work / "documents"
@@ -34,8 +34,22 @@ def smoke(cli, work):
         [str(cli), "index", "--db", str(db), "--root", str(corpus)],
         env=environment, check=True, capture_output=True, timeout=60,
     )
+    check((models is None) == (worker is None), "Supply both models and worker")
+    semantic_args = []
+    embedding = None
+    if models is not None:
+        models = models.resolve(strict=True)
+        worker = worker.resolve(strict=True)
+        semantic_args = ["--model-dir", str(models), "--embedding-worker", str(worker),
+                         "--cpu-threads", "1"]
+        encoded = subprocess.run(
+            [str(cli), "embed", "--db", str(db), *semantic_args, "--text-only"],
+            env=environment, check=True, capture_output=True, timeout=180,
+            text=True, encoding="utf-8",
+        )
+        embedding = json.loads(encoded.stdout)
     child = subprocess.Popen(
-        [str(cli), "mcp", "--db", str(db)], env=environment,
+        [str(cli), "mcp", "--db", str(db), *semantic_args], env=environment,
         stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
         text=True, encoding="utf-8",
     )
@@ -86,7 +100,8 @@ def smoke(cli, work):
         found = request(3, "tools/call", {"name": "search", "arguments": {
             "query": "termination", "extension": "txt", "limit": 3,
         }})["structuredContent"]
-        check(found["mode"] == "lexical_only", "Incorrect readiness mode")
+        expected_mode = "hybrid" if models is not None else "lexical_only"
+        check(found["mode"] == expected_mode, "Incorrect readiness mode")
         check(found["results"][0]["filename"] == "contract.txt", "Wrong document")
         source = request(4, "tools/call", {"name": "read_source", "arguments": {
             "handle": found["results"][0]["source_handle"],
@@ -103,8 +118,31 @@ def smoke(cli, work):
               "Code filter included ordinary documents")
         with cli.open("rb") as stream:
             checksum = hashlib.file_digest(stream, "sha256").hexdigest()
-        report = {"complete": True, "cli_sha256": checksum, "mode": "lexical_only",
-                  "cpu_forced": True, "inference": False, "tool_names": [t["name"] for t in tools],
+        semantic = None
+        if models is not None:
+            lexical = subprocess.run(
+                [str(cli), "search", "--db", str(db), "ending a vendor relationship",
+                 "--extension", "txt", "--json"],
+                env=environment, check=True, capture_output=True, timeout=15,
+                text=True, encoding="utf-8",
+            )
+            check(json.loads(lexical.stdout) == [], "Meaning query has lexical matches")
+            check(embedding["document_inference_calls"] == 3,
+                  "Fixture documents were not actually encoded")
+            semantic = request(7, "tools/call", {"name": "search", "arguments": {
+                "query": "ending a vendor relationship", "extension": "txt", "limit": 3,
+            }})["structuredContent"]
+            check(semantic["mode"] == "hybrid", "Semantic request lost hybrid mode")
+            check(semantic["results"][0]["filename"] == "contract.txt", "Meaning query failed")
+            check(semantic["results"][0]["match_kind"] == "semantic", "Not a semantic match")
+            cited = request(8, "tools/call", {"name": "read_source", "arguments": {
+                "handle": semantic["results"][0]["source_handle"],
+            }})["structuredContent"]
+            check("sixty days" in cited["text"], "Semantic citation lost source text")
+        report = {"complete": True, "cli_sha256": checksum, "mode": expected_mode,
+                  "cpu_forced": True, "inference": models is not None,
+                  "embedding": embedding, "semantic": semantic,
+                  "tool_names": [t["name"] for t in tools],
                   "search": found, "source": source, "code": code}
         with (work / "report.json").open("x", encoding="utf-8") as stream:
             json.dump(report, stream, indent=2)
@@ -123,7 +161,9 @@ if __name__ == "__main__":
     parser.add_argument("--cli", type=Path, required=True)
     parser.add_argument("--work-dir", type=Path, required=True,
                         help="New owned fixture/evidence directory, retained after the run")
+    parser.add_argument("--model-dir", type=Path)
+    parser.add_argument("--embedding-worker", type=Path)
     args = parser.parse_args()
-    result = smoke(args.cli, args.work_dir.resolve())
+    result = smoke(args.cli, args.work_dir.resolve(), args.model_dir, args.embedding_worker)
     print(json.dumps({"complete": result["complete"], "mode": result["mode"],
                       "cli_sha256": result["cli_sha256"]}))