Agentswarm/scripts/test-autonomous-tasks.py

r"""Module-level tests for agent-proposed autonomous task generation (issue #7).

Hermetic: exercises orchestrator/autonomous_tasks.py directly. The module is pure (no Redis, no
WebSocket, no FastAPI), so this needs no running orchestrator, no model key, and no Redis — it
simulates an agent submitting a proposal from a shared-state snapshot, the policy
accepting/rejecting/merging it, and the accepted proposal producing a new pending-task spec with
full lineage and status transitions.

Run from agent_swarm_v6 (install deps first — the module itself has no third-party deps, but the
repo's standard env is assumed):
    pip install -r orchestrator/requirements.txt -r agent/requirements.txt
    ..\.venv\Scripts\python.exe scripts/test-autonomous-tasks.py
"""
import os
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from orchestrator.autonomous_tasks import (
    TaskProposal,
    ProposalLineage,
    ProposalPolicy,
    ProposalStatus,
    ProposalDecision,
    ExistingTaskRef,
    PROPOSED_SOURCE,
    EVENT_SUBMITTED,
    EVENT_ACCEPTED,
    EVENT_REJECTED,
    EVENT_MERGED,
    review_proposal,
    ingest_accepted_proposal,
    build_lifecycle_events_for_outcome,
    description_similarity,
    assert_not_master_origin,
)

failures = []


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


def make_proposal(**overrides):
    """An agent proposing a follow-up task it discovered from the shared run state."""
    snapshot = {
        "completed_tasks": ["swarm-1-impl"],
        "open_gaps": ["no unit tests for the new module"],
        "pending_task_ids": [],
    }
    kwargs = dict(
        proposed_by_agent_id="agent-7a3f",
        title="Add unit tests for payment module",
        description="Write unit tests covering the new payment module added by the impl task",
        proposal_reason="Implementation landed with no test coverage; gap found in shared state",
        proposal_confidence=0.82,
        agent_role="testing",
        required_capabilities=["testing"],
        lineage=ProposalLineage(
            origin_task_id="swarm-1-impl",
            trigger_event="task.completed",
            shared_state_snapshot=snapshot,
        ),
    )
    kwargs.update(overrides)
    return TaskProposal(**kwargs)


def test_proposal_invariants():
    p = make_proposal()
    check("source pinned to agent_proposed", p.source == PROPOSED_SOURCE)
    check("initial status PROPOSED", p.status == ProposalStatus.PROPOSED)
    check("proposal_id generated", bool(p.proposal_id) and p.proposal_id.startswith("prop-"))
    check("confidence preserved", abs(p.proposal_confidence - 0.82) < 1e-9)

    # source cannot be overridden away from agent_proposed
    p2 = make_proposal(source="planner")
    check("source override ignored", p2.source == PROPOSED_SOURCE)

    # confidence clamped into [0,1]
    p3 = make_proposal(proposal_confidence=5.0)
    check("confidence clamped high", p3.proposal_confidence == 1.0)
    p4 = make_proposal(proposal_confidence=-1.0)
    check("confidence clamped low", p4.proposal_confidence == 0.0)

    # requires an agent id
    try:
        make_proposal(proposed_by_agent_id="")
        check("empty proposer rejected", False)
    except ValueError:
        check("empty proposer rejected", True)


def test_not_master_origin():
    # A Master/planner must not launder generated tasks through this agent path.
    bad = make_proposal(proposed_by_agent_id="master-agent")
    try:
        assert_not_master_origin(bad)
        check("master proposer rejected", False)
    except ValueError:
        check("master proposer rejected", True)

    bad2 = make_proposal(
        lineage=ProposalLineage(origin_task_id="x", trigger_event="planner.decomposed")
    )
    try:
        assert_not_master_origin(bad2)
        check("planner trigger rejected", False)
    except ValueError:
        check("planner trigger rejected", True)

    check("agent proposer allowed", assert_not_master_origin(make_proposal()) is None)


def test_accept_path():
    p = make_proposal()
    outcome = review_proposal(p, ProposalPolicy(), existing_tasks=[])
    check("accept decision", outcome.decision == ProposalDecision.ACCEPT)
    check("accept status transition", p.status == ProposalStatus.ACCEPTED)
    check("accept reason set", bool(p.decision_reason))

    spec = ingest_accepted_proposal(p, swarm_id="swarm-1")
    check("spec source agent_proposed", spec["source"] == PROPOSED_SOURCE)
    check("spec description carried", spec["description"] == p.description)
    check("spec role carried", spec["agent_role"] == "testing")
    # lineage fully threaded into context for audit
    prov = spec["context"]["proposal"]
    check("lineage proposer", prov["proposed_by_agent_id"] == "agent-7a3f")
    check("lineage reason", "test coverage" in prov["proposal_reason"].lower())
    check("lineage origin task", prov["origin_task_id"] == "swarm-1-impl")
    check("lineage trigger", prov["trigger_event"] == "task.completed")
    check("lineage snapshot preserved", prov["shared_state_snapshot"]["open_gaps"][0].startswith("no unit tests"))
    # origin task becomes parent so the new task threads into the run
    check("spec parent = origin", spec["parent_task_id"] == "swarm-1-impl")

    # cannot ingest a non-accepted proposal
    rej = make_proposal(proposal_confidence=0.1)
    review_proposal(rej, ProposalPolicy(), existing_tasks=[])
    try:
        ingest_accepted_proposal(rej)
        check("ingest blocks non-accepted", False)
    except ValueError:
        check("ingest blocks non-accepted", True)


def test_reject_low_confidence():
    p = make_proposal(proposal_confidence=0.3)
    outcome = review_proposal(p, ProposalPolicy(min_confidence=0.6), existing_tasks=[])
    check("reject low confidence", outcome.decision == ProposalDecision.REJECT)
    check("reject status", p.status == ProposalStatus.REJECTED)
    check("reject reason mentions confidence", "confidence" in (p.decision_reason or ""))


def test_reject_budget_exhausted():
    p = make_proposal()
    policy = ProposalPolicy(remaining_proposal_budget=0)
    outcome = review_proposal(p, policy, existing_tasks=[])
    check("reject when budget exhausted", outcome.decision == ProposalDecision.REJECT)
    check("budget reject status", p.status == ProposalStatus.REJECTED)


def test_merge_duplicate():
    existing = [
        ExistingTaskRef(
            task_id="swarm-1-existing-tests",
            description="Write unit tests covering the new payment module added by the impl task",
            agent_role="testing",
            status="pending",
        ),
    ]
    p = make_proposal()
    outcome = review_proposal(p, ProposalPolicy(), existing_tasks=existing)
    check("merge decision on duplicate", outcome.decision == ProposalDecision.MERGE)
    check("merge status", p.status == ProposalStatus.MERGED)
    check("merge target recorded", p.merged_into_task_id == "swarm-1-existing-tests")
    check("merge outcome target", outcome.merge_target_task_id == "swarm-1-existing-tests")

    # a completed task is NOT a dedup target (follow-up rounds allowed)
    done = [ExistingTaskRef(task_id="t", description=p.description, status="completed")]
    p2 = make_proposal()
    outcome2 = review_proposal(p2, ProposalPolicy(), existing_tasks=done)
    check("completed task not deduped", outcome2.decision == ProposalDecision.ACCEPT)


def test_similarity_metric():
    check("identical similarity 1.0", description_similarity("add tests now", "add tests now") == 1.0)
    check("disjoint similarity 0.0", description_similarity("alpha beta", "gamma delta") == 0.0)
    check("empty similarity 0.0", description_similarity("", "anything") == 0.0)


def test_lifecycle_events():
    # accept -> submitted + accepted
    p = make_proposal()
    out = review_proposal(p, ProposalPolicy(), existing_tasks=[])
    events = build_lifecycle_events_for_outcome(p, out)
    check("accept emits 2 events", len(events) == 2)
    check("first event submitted", events[0]["event_type"] == EVENT_SUBMITTED)
    check("second event accepted", events[1]["event_type"] == EVENT_ACCEPTED)
    check("event carries proposer", events[0]["proposed_by_agent_id"] == "agent-7a3f")
    check("event carries source", events[0]["source"] == PROPOSED_SOURCE)

    # reject -> submitted + rejected
    r = make_proposal(proposal_confidence=0.1)
    out_r = review_proposal(r, ProposalPolicy(), existing_tasks=[])
    ev_r = build_lifecycle_events_for_outcome(r, out_r)
    check("reject second event", ev_r[1]["event_type"] == EVENT_REJECTED)

    # merge -> submitted + merged with target
    existing = [ExistingTaskRef(task_id="dupe", description=make_proposal().description, status="pending")]
    m = make_proposal()
    out_m = review_proposal(m, ProposalPolicy(), existing_tasks=existing)
    ev_m = build_lifecycle_events_for_outcome(m, out_m)
    check("merge second event", ev_m[1]["event_type"] == EVENT_MERGED)
    check("merge event has target", ev_m[1]["merge_target_task_id"] == "dupe")


def main():
    test_proposal_invariants()
    test_not_master_origin()
    test_accept_path()
    test_reject_low_confidence()
    test_reject_budget_exhausted()
    test_merge_duplicate()
    test_similarity_metric()
    test_lifecycle_events()

    print()
    if failures:
        print(f"{len(failures)} FAILED: {failures}")
        sys.exit(1)
    print("ALL AUTONOMOUS-TASK TESTS PASSED")


if __name__ == "__main__":
    main()