Agentswarm/scripts/test-swarm-autonomous.py

"""Integration test for decentralized-rework P3: agent-driven decomposition (#7).

An executing agent proposes follow-up tasks from the shared seed/run state; the orchestrator
reviews each (confidence floor / dedup-merge / per-run budget) and enqueues accepted ones as real
PENDING tasks with full lineage. This is the bottom-up decomposition that replaces the Master plan.
Hermetic, no model key.

Run from agent_swarm_v6 (install deps first — needs fakeredis):
    pip install -r orchestrator/requirements.txt -r agent/requirements.txt
    REDIS_FAKE=1 ENABLE_AGENT_TASK_PROPOSALS=1 AGENT_PROPOSAL_BUDGET=3 python scripts/test-swarm-autonomous.py
"""
import asyncio
import os
import sys
from pathlib import Path

os.environ["REDIS_FAKE"] = "1"
os.environ["ENABLE_AGENT_TASK_PROPOSALS"] = "1"
os.environ["AGENT_PROPOSAL_BUDGET"] = "3"
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
from orchestrator.task_queue import task_queue
from orchestrator import main as orch

failures = []


def check(name, cond):
    print(("PASS" if cond else "FAIL"), "-", name)
    if not cond:
        failures.append(name)


async def _noop(self, *a, **k):
    return None


async def propose(seed_id, desc, *, agent="agent-A", conf=0.9, role="general", caps=None):
    return await orch.handle_task_proposal(agent, {
        "origin_task_id": seed_id, "description": desc, "reason": "discovered gap in shared state",
        "confidence": conf, "agent_role": role, "required_capabilities": caps or [],
        "trigger_event": "task.completed",
    })


async def main():
    await redis_client.connect()
    sr_mod.SwarmRuntime._post_callback = _noop

    body = {"mode": "swarm", "requirement": {"objective": "Build a CSV parser"},
            "callback": {"url": "http://127.0.0.1:9/cb", "subscribed_events": []},
            "metadata": {"manager_deployment_id": "m-auto"}}
    run, _ = await swarm_runtime.get_or_create_run(body=body, idempotency_key=None, correlation_id="ca")
    seed = await task_queue.create_task(task_id=f"{run.swarm_id}-seed", description="Build a CSV parser",
                                        agent_role="general", required_capabilities=[], enqueue=False)
    await swarm_runtime.attach_task(run, seed.task_id)

    # 1) two distinct high-confidence proposals → ACCEPT → real tasks with lineage
    r1 = await propose(seed.task_id, "Write unit tests for the CSV parser module", role="testing", caps=["testing"])
    r2 = await propose(seed.task_id, "Write API documentation for the parser", role="documentation")
    check("proposal 1 accepted → task created", r1.get("accepted") and r1.get("task_id"))
    check("proposal 2 accepted → task created", r2.get("accepted") and r2.get("task_id"))

    t1 = await task_queue.get_task(r1["task_id"])
    check("created task is agent_proposed (bottom-up, not Master)", t1 and t1.source == "agent_proposed")
    lineage = (t1.context or {}).get("proposal") or {}
    check("created task carries lineage (proposer + origin + reason)",
          lineage.get("proposed_by_agent_id") == "agent-A"
          and lineage.get("origin_task_id") == seed.task_id
          and lineage.get("proposal_reason"))

    # 2) near-duplicate of proposal 1 → MERGE (not a new task)
    rdup = await propose(seed.task_id, "Write unit tests for the CSV parser module")
    check("duplicate proposal → MERGE", rdup.get("decision") == "merge" and not rdup.get("accepted"))
    check("merge names the target task", rdup.get("merge_target_task_id") == r1["task_id"])

    # 3) third distinct proposal → ACCEPT (budget 3: A,B + this = 3)
    r3 = await propose(seed.task_id, "Add a benchmark harness measuring parser throughput", role="general")
    check("proposal 3 accepted (within budget)", r3.get("accepted"))

    # 4) fourth distinct proposal → budget exhausted → REJECT
    r4 = await propose(seed.task_id, "Containerize the parser service with a Dockerfile")
    check("proposal 4 rejected (budget exhausted)",
          not r4.get("accepted") and r4.get("decision") == "reject" and "budget" in (r4.get("reason") or ""))

    # 5) low-confidence proposal → REJECT
    r5 = await propose(seed.task_id, "Maybe refactor something unspecified", conf=0.2)
    check("low-confidence proposal rejected",
          not r5.get("accepted") and r5.get("decision") == "reject")

    # run grew from 1 seed → 1 seed + 3 accepted = 4 tasks; proposals telemetry recorded
    refreshed = await swarm_runtime.get_run(run.swarm_id)
    check("run task graph grew bottom-up to 4 tasks (seed + 3)", len(refreshed.task_ids) == 4)
    check("proposal lifecycle telemetry recorded on run",
          bool(refreshed.metadata.get("proposals")))
    check("accepted-proposal counter = 3", refreshed.metadata.get("proposal_accepted_count") == 3)

    # Post-cutover: proposals are unconditional (swarm is the runtime; no enable flag).
    os.environ.pop("ENABLE_AGENT_TASK_PROPOSALS", None)
    roff = await propose(seed.task_id, "Add a CLI entrypoint for the parser")
    check("proposals processed without any flag (swarm default)", roff.get("decision") != "disabled")

    print()
    if failures:
        print(f"{len(failures)} autonomous-task (P3) check(s) FAILED: {failures}")
        sys.exit(1)
    print("all swarm autonomous-task (P3) checks passed")


if __name__ == "__main__":
    asyncio.run(main())