Merge pull request #28 from xmindlab-heicode/feat/contract-freeze

契约冻结 v1：Manager/客户端 Swarm Run 查询契约（Refs #2 #14 #15）
2026-06-10 18:36:38 +08:00
parent 4bc98cf177 15fe5d379b
commit 5a3296e7d1
11 changed files with 350 additions and 38 deletions
@@ -0,0 +1,163 @@
+"""Contract-freeze tests (agent_swarm#14 / #15) for the Manager/client query contract.
+
+Verifies the frozen Swarm->Manager callback contract that the desktop "task cockpit"
+(heicode-mananger #28/#45/#46) consumes:
+  * every event envelope carries a per-swarm strictly-increasing, gap-free `sequence` (#15.1);
+  * the 13 frozen client-facing event types exist and round-trip with a sequence (#15.2);
+  * approval.approved/rejected fire from the real approval-decision site (#15.2);
+  * swarm.stopped fires from the real stop site; swarm.completed/failed are defined (#15.2);
+  * artifact.created carries the flat {uri, checksum, task_id, created_at} shape (#15.4);
+  * secret_ref/credential_ref pass through as azkv refs but plaintext creds are redacted (#15.3).
+
+Hermetic: REDIS_FAKE, no model key, no real Manager callback (callback url empty).
+
+Run from agent_swarm_v6 (install deps first — needs fakeredis):
+    pip install -r orchestrator/requirements.txt
+    REDIS_FAKE=1 python scripts/test-contract-freeze.py
+"""
+import asyncio
+import json
+import os
+import sys
+from pathlib import Path
+
+os.environ["REDIS_FAKE"] = "1"
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from orchestrator.redis_client import redis_client
+from orchestrator import swarm_runtime as sr_mod
+from orchestrator.swarm_runtime import swarm_runtime, FROZEN_CLIENT_EVENT_TYPES
+
+failures = []
+
+
+def check(name, cond):
+    print(("PASS" if cond else "FAIL"), "-", name)
+    if not cond:
+        failures.append(name)
+
+
+async def stored_events(swarm_id):
+    raw = await redis_client.lrange(f"{swarm_runtime.EVENT_KEY_PREFIX}{swarm_id}", 0, -1)
+    return [json.loads(r) for r in raw]
+
+
+async def new_run(objective="freeze test"):
+    body = {"mode": "swarm", "orchestration_plan": {"objective": objective},
+            # no callback url -> nothing is POSTed; events are still stored locally
+            "callback": {"url": "", "subscribed_events": []},
+            "metadata": {"manager_deployment_id": "m-freeze"}}
+    run, _ = await swarm_runtime.get_or_create_run(body=body, idempotency_key=None, correlation_id="cf")
+    return run
+
+
+async def test_sequence_monotonic():
+    run = await new_run()
+    for i in range(5):
+        await swarm_runtime.emit_event(run, "task.running", task_id=f"t{i}", payload={"agent_role": "impl"})
+    evs = await stored_events(run.swarm_id)
+    seqs = [e.get("sequence") for e in evs]
+    check("every event carries a sequence", all(isinstance(s, int) for s in seqs))
+    check("sequence starts at 1 and is gap-free/increasing", seqs == list(range(1, len(seqs) + 1)))
+
+    # A second run has its OWN sequence space starting at 1 (per-swarm, not global).
+    run2 = await new_run()
+    await swarm_runtime.emit_event(run2, "task.running", task_id="x", payload={"agent_role": "impl"})
+    evs2 = await stored_events(run2.swarm_id)
+    check("sequence is per-swarm (second run restarts at 1)", evs2[0]["sequence"] == 1)
+
+
+async def test_frozen_event_types_roundtrip():
+    check("frozen set has exactly 13 client event types", len(FROZEN_CLIENT_EVENT_TYPES) == 13)
+    expected = {
+        "task.created", "task.claimed", "task.running", "task.completed", "task.failed",
+        "handoff.created", "approval.requested", "approval.approved", "approval.rejected",
+        "artifact.created", "swarm.completed", "swarm.failed", "swarm.stopped",
+    }
+    check("frozen set matches the agreed 13", set(FROZEN_CLIENT_EVENT_TYPES) == expected)
+
+    run = await new_run()
+    for et in FROZEN_CLIENT_EVENT_TYPES:
+        await swarm_runtime.emit_event(run, et, payload={"status": "x"})
+    evs = await stored_events(run.swarm_id)
+    types = [e["event_type"] for e in evs]
+    # The run also emits deployment.status_changed on creation; assert the frozen 13 are all present.
+    check("all 13 frozen types round-trip with an envelope", expected.issubset(set(types)))
+    check("every frozen-type envelope has a sequence", all(isinstance(e["sequence"], int) for e in evs))
+
+
+async def test_artifact_shape():
+    run = await new_run()
+    await swarm_runtime.emit_event(
+        run, "artifact.created", task_id="t-art",
+        artifact={"artifact_id": "art_1", "uri": "git://repo#main", "checksum": "abc123",
+                  "size_bytes": 42},
+    )
+    ev = (await stored_events(run.swarm_id))[-1]
+    p = ev["payload"]
+    check("artifact payload has uri/checksum", p.get("uri") == "git://repo#main" and p.get("checksum") == "abc123")
+    check("artifact task_id defaulted from event task_id", p.get("task_id") == "t-art")
+    check("artifact created_at defaulted to event time", bool(p.get("created_at")))
+    check("artifact size_bytes preserved when known", p.get("size_bytes") == 42)
+
+    # size unknown -> NOT fabricated (rule #9)
+    run2 = await new_run()
+    await swarm_runtime.emit_event(run2, "artifact.created", task_id="t2",
+                                   artifact={"artifact_id": "art_2", "uri": "runtime://x", "checksum": "d"})
+    p2 = (await stored_events(run2.swarm_id))[-1]["payload"]
+    check("unknown size_bytes is omitted, not faked", "size_bytes" not in p2)
+
+
+async def test_approval_and_stop_sites():
+    # approval.approved from the real decision site
+    run = await new_run()
+    run.approvals["ap1"] = {"approval_id": "ap1"}
+    await swarm_runtime.save_run(run)
+    await swarm_runtime.record_approval_decision(run.swarm_id, "ap1", {"decision": "approved"})
+    types = [e["event_type"] for e in await stored_events(run.swarm_id)]
+    check("approval.approved emitted on approve", "approval.approved" in types)
+
+    run2 = await new_run()
+    run2.approvals["ap2"] = {"approval_id": "ap2"}
+    await swarm_runtime.save_run(run2)
+    await swarm_runtime.record_approval_decision(run2.swarm_id, "ap2", {"decision": "rejected", "reason": "no"})
+    types2 = [e["event_type"] for e in await stored_events(run2.swarm_id)]
+    check("approval.rejected emitted on reject", "approval.rejected" in types2)
+
+    # swarm.stopped from the real stop site
+    run3 = await new_run()
+    await swarm_runtime.stop_run(run3.deployment_id, reason="manager stop")
+    ev3 = await stored_events(run3.swarm_id)
+    check("swarm.stopped emitted on stop_run", "swarm.stopped" in [e["event_type"] for e in ev3])
+    stopped = [e for e in ev3 if e["event_type"] == "swarm.stopped"][0]
+    check("swarm.stopped payload status=stopped", stopped["payload"].get("status") == "stopped")
+
+
+async def test_redaction():
+    run = await new_run()
+    await swarm_runtime.emit_event(run, "approval.requested", payload={
+        "approval_id": "ap",
+        "secret_ref": "azkv://vault/secrets/x",   # azkv reference: passes through (not plaintext)
+        "access_token": "PLAINTEXT-SHOULD-VANISH",
+    })
+    p = (await stored_events(run.swarm_id))[-1]["payload"]
+    check("secret_ref (azkv ref) passes through for HM to strip", p.get("secret_ref") == "azkv://vault/secrets/x")
+    check("plaintext access_token is redacted", p.get("access_token") == "[redacted]")
+
+
+async def main():
+    await redis_client.connect()
+    await test_sequence_monotonic()
+    await test_frozen_event_types_roundtrip()
+    await test_artifact_shape()
+    await test_approval_and_stop_sites()
+    await test_redaction()
+    print()
+    if failures:
+        print(f"{len(failures)} contract-freeze check(s) FAILED: {failures}")
+        sys.exit(1)
+    print("all contract-freeze checks passed")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
@@ -47,6 +47,10 @@ class FakeRedis:
    async def get(self, key):
        return self.values.get(key)

+    async def incr(self, key):
+        self.values[key] = int(self.values.get(key, 0)) + 1
+        return self.values[key]
+
    async def keys(self, pattern):
        prefix = pattern.rstrip("*")
        return [key for key in self.values if key.startswith(prefix)]
@@ -107,6 +107,8 @@ async def main():
            sources = [t.get("source") for t in tasks]
            payloads = [(e.get("payload") or {}) for e in events]
            termination_seen = any(p.get("termination_reason") for p in payloads)
+            ev_types = [e.get("event_type") for e in events]
+            seqs = [e.get("sequence") for e in events]

            # ---- swarm-flow assertions (seed → agent-decompose → self-select → converge) ----
            check("seed: a single objective seed task was injected (no Master plan)",
@@ -120,6 +122,12 @@ async def main():
            check("converge: run reached completed", status == "completed")
            check("converge: a termination_reason was emitted (convergence report)", termination_seen)
            check("synthesize: a final unified summary is present", bool(wf.get("summary")))
+            # ---- frozen Manager/client contract (agent_swarm#14/#15) ----
+            check("contract: terminal swarm.completed event emitted", "swarm.completed" in ev_types)
+            check("contract: every event carries a per-swarm sequence",
+                  bool(seqs) and all(isinstance(s, int) for s in seqs))
+            check("contract: sequence is strictly increasing and gap-free",
+                  seqs == list(range(1, len(seqs) + 1)))
    finally:
        agent.running = False
        if agent_task: