Agentswarm/orchestrator/autonomous_tasks.py

"""Agent-proposed autonomous task generation (issue #7).

This module lets an *execution unit* (an agent) propose a NEW task it discovered while
working — e.g. it notices the shared run state is missing tests, docs, or a follow-up step —
and lets the orchestrator decide whether to accept, reject, or merge that proposal into the
runtime task graph.

Design boundaries (read before extending):

- This is the AGENT proposing, NOT a Master/planner generating work. The planner fallback
  (``master_agent.plan`` / ``ENABLE_PLANNER_FALLBACK``) is a top-down decomposition done by a
  controller. This path is strictly bottom-up: the proposal carries ``source="agent_proposed"``
  and ``proposed_by_agent_id``, and ``review_proposal``/``ingest_accepted_proposal`` MUST NOT be
  driven by master_agent. ``assert_not_master_origin`` enforces that at runtime.
- PURE module: no Redis, no WebSocket, no FastAPI imports. The functions take plain data in and
  return plain data out, so the orchestrator integrator wires them into the live loop and the
  hermetic test exercises them at the module level without any infrastructure.
- UNCONDITIONAL: agent task proposals are the swarm's ONLY decomposition path — there is no
  enable flag (this repo is the swarm runtime; see docs/swarm/decentralized-rework-plan.md). The
  WS ``task_proposal`` branch + ``handle_task_proposal`` in main.py are always active.
- HONESTY (rule #9): what this module implements — the proposal model, the review policy
  (confidence/budget/dedup), the accepted->task-spec mapping with full lineage, and the
  lifecycle-event builders. Wiring into the live loop is in main.py (see
  ``docs/swarm/autonomous-task-generation.md``).
"""
from __future__ import annotations

import time
import uuid
from dataclasses import dataclass, field, asdict
from enum import Enum
from typing import Any, Dict, List, Optional, Sequence


# --- constants --------------------------------------------------------------------------

# The Task.source value a proposal-derived task carries. Distinct from "manual"/"planner"/
# "runtime_bridge"/"dynamic_handoff" so the Manager/audit can attribute bottom-up tasks.
PROPOSED_SOURCE = "agent_proposed"

# Lifecycle event types. Internal runtime events (NOT registered Heicode Manager events): an
# integrator either maps them onto an existing HM event or records them as telemetry. Kept here
# so producers and the collector share one vocabulary. See doc §事件.
EVENT_SUBMITTED = "task.proposal_submitted"
EVENT_ACCEPTED = "task.proposal_accepted"
EVENT_REJECTED = "task.proposal_rejected"
EVENT_MERGED = "task.proposal_merged"


class ProposalStatus(str, Enum):
    """Lifecycle status of an agent-submitted task proposal."""
    PROPOSED = "proposed"
    ACCEPTED = "accepted"
    REJECTED = "rejected"
    MERGED = "merged"
    EXPIRED = "expired"


class ProposalDecision(str, Enum):
    """Outcome of the review policy for a single proposal."""
    ACCEPT = "accept"
    REJECT = "reject"
    MERGE = "merge"


# --- data models ------------------------------------------------------------------------

@dataclass
class ProposalLineage:
    """Provenance of a proposal: where it came from and what triggered it.

    This is what makes a proposed task auditable end-to-end: the originating task, the event
    that prompted the agent to propose, and the shared-state snapshot the agent reasoned over.
    """
    origin_task_id: Optional[str] = None
    trigger_event: Optional[str] = None
    # A redaction-safe snapshot of the shared run state the agent based the proposal on
    # (e.g. completed task summaries, open gaps, pending task ids). Free-form on purpose; the
    # dedup/policy logic only reads a few well-known keys (see review_proposal).
    shared_state_snapshot: Dict[str, Any] = field(default_factory=dict)


@dataclass
class TaskProposal:
    """A NEW task an agent proposes based on shared run state.

    ``source`` is fixed to PROPOSED_SOURCE and ``proposed_by_agent_id`` is required: this is the
    agent proposing, never a Master. ``proposal_confidence`` is the agent's own 0..1 estimate that
    the task is worth doing; the orchestrator policy (review_proposal), not the agent, decides.
    """
    proposed_by_agent_id: str
    description: str
    proposal_reason: str
    proposal_confidence: float = 0.0
    title: Optional[str] = None
    agent_role: str = "general"
    required_capabilities: List[str] = field(default_factory=list)
    depends_on: List[str] = field(default_factory=list)
    lineage: ProposalLineage = field(default_factory=ProposalLineage)
    proposal_id: str = field(default_factory=lambda: f"prop-{uuid.uuid4().hex[:12]}")
    source: str = PROPOSED_SOURCE
    status: ProposalStatus = ProposalStatus.PROPOSED
    created_at: float = field(default_factory=time.time)
    # The id of an existing task this proposal was merged into (set on MERGE).
    merged_into_task_id: Optional[str] = None
    # Human/policy-readable note attached by review_proposal (why it was rejected, what it merged
    # into, etc.). Carried into the lifecycle event payload.
    decision_reason: Optional[str] = None

    def __post_init__(self):
        # Hard-pin the source: even if a caller passes something else, a TaskProposal is by
        # definition agent-proposed. This keeps Task.source attribution honest.
        self.source = PROPOSED_SOURCE
        if not self.proposed_by_agent_id:
            raise ValueError("TaskProposal requires proposed_by_agent_id (an agent, not a Master)")
        # Clamp confidence into [0, 1] so a bad client value cannot defeat the threshold.
        try:
            self.proposal_confidence = max(0.0, min(1.0, float(self.proposal_confidence)))
        except (TypeError, ValueError):
            self.proposal_confidence = 0.0

    def to_dict(self) -> Dict[str, Any]:
        """Serialize to a plain dict (enums -> values) for events / transport / storage."""
        data = asdict(self)
        data["status"] = self.status.value
        return data


@dataclass
class ProposalPolicy:
    """Acceptance policy for the review step.

    All thresholds are explicit so the integrator can source them from the run's
    orchestration_plan/budget. Nothing here is read from the model — the agent only *proposes*.
    """
    # Minimum agent confidence to consider accepting at all.
    min_confidence: float = 0.6
    # Confidence at/above which a unique proposal is accepted outright. Between min_confidence
    # and this band, a non-duplicate proposal is still accepted; this field is kept so an
    # integrator can split "auto-accept" from "queue for human review" later.
    auto_accept_confidence: float = 0.6
    # Remaining proposal budget for the run (how many more proposed tasks may be enqueued).
    # 0 => budget exhausted => reject. Integrator decrements this as it enqueues.
    remaining_proposal_budget: int = 3
    # Similarity ratio (0..1) at/above which a proposal is treated as a duplicate of an existing
    # task and MERGED rather than enqueued as new.
    dedup_similarity_threshold: float = 0.8


@dataclass
class ExistingTaskRef:
    """The minimal view of an existing task the dedup check needs.

    A plain dataclass (not the Pydantic Task) so this module stays infra-free; the integrator
    builds these from task_queue.get_all_tasks().
    """
    task_id: str
    description: str
    agent_role: str = "general"
    status: str = "pending"


@dataclass
class ReviewOutcome:
    """Result of review_proposal: the decision plus the (mutated) proposal and a reason."""
    decision: ProposalDecision
    proposal: TaskProposal
    reason: str
    # Set when decision is MERGE: the existing task id the proposal folds into.
    merge_target_task_id: Optional[str] = None


# --- dedup helper -----------------------------------------------------------------------

def _normalize_words(text: str) -> set:
    """Lowercase token set used for cheap, dependency-free description similarity."""
    return {w for w in "".join(c.lower() if c.isalnum() else " " for c in (text or "")).split() if w}


def description_similarity(a: str, b: str) -> float:
    """Jaccard similarity of word sets in two descriptions (0..1).

    Deliberately simple and deterministic (no embeddings, no model call) so the dedup decision is
    reproducible and testable. Matches the spirit of the heuristic match score in decision_engine.
    """
    wa, wb = _normalize_words(a), _normalize_words(b)
    if not wa or not wb:
        return 0.0
    inter = len(wa & wb)
    union = len(wa | wb)
    return inter / union if union else 0.0


def find_duplicate(
    proposal: TaskProposal,
    existing_tasks: Sequence[ExistingTaskRef],
    threshold: float,
) -> Optional[ExistingTaskRef]:
    """Return the most similar non-terminal existing task above ``threshold``, else None.

    Only live tasks (pending/assigned/in_progress/blocked) are dedup targets — re-proposing work
    similar to a *completed* task is allowed (the agent may legitimately want a follow-up round).
    """
    live = {"pending", "assigned", "in_progress", "blocked"}
    best: Optional[ExistingTaskRef] = None
    best_sim = threshold
    for task in existing_tasks:
        if task.status not in live:
            continue
        sim = description_similarity(proposal.description, task.description)
        if sim >= best_sim:
            best_sim = sim
            best = task
    return best


# --- core policy ------------------------------------------------------------------------

def review_proposal(
    proposal: TaskProposal,
    policy: ProposalPolicy,
    existing_tasks: Optional[Sequence[ExistingTaskRef]] = None,
) -> ReviewOutcome:
    """Decide whether to accept / reject / merge an agent-submitted proposal.

    Pure function. Order of checks:
      1. Confidence floor: below ``policy.min_confidence`` => REJECT.
      2. Dedup: if it closely matches a live existing task => MERGE into that task.
      3. Budget: if the run's remaining proposal budget is exhausted => REJECT.
      4. Otherwise => ACCEPT.

    Mutates and returns the proposal with its new status / decision_reason so the caller can
    persist it and emit the matching lifecycle event. This NEVER calls a model or a Master — the
    agent proposed; the orchestrator policy disposes.
    """
    existing_tasks = existing_tasks or []

    if proposal.proposal_confidence < policy.min_confidence:
        proposal.status = ProposalStatus.REJECTED
        proposal.decision_reason = (
            f"confidence {proposal.proposal_confidence:.2f} < min {policy.min_confidence:.2f}"
        )
        return ReviewOutcome(ProposalDecision.REJECT, proposal, proposal.decision_reason)

    duplicate = find_duplicate(proposal, existing_tasks, policy.dedup_similarity_threshold)
    if duplicate is not None:
        proposal.status = ProposalStatus.MERGED
        proposal.merged_into_task_id = duplicate.task_id
        proposal.decision_reason = f"duplicate of existing task {duplicate.task_id}"
        return ReviewOutcome(
            ProposalDecision.MERGE,
            proposal,
            proposal.decision_reason,
            merge_target_task_id=duplicate.task_id,
        )

    if policy.remaining_proposal_budget <= 0:
        proposal.status = ProposalStatus.REJECTED
        proposal.decision_reason = "proposal budget exhausted for this run"
        return ReviewOutcome(ProposalDecision.REJECT, proposal, proposal.decision_reason)

    proposal.status = ProposalStatus.ACCEPTED
    proposal.decision_reason = (
        f"accepted (confidence {proposal.proposal_confidence:.2f} >= {policy.min_confidence:.2f})"
    )
    return ReviewOutcome(ProposalDecision.ACCEPT, proposal, proposal.decision_reason)


def ingest_accepted_proposal(
    proposal: TaskProposal,
    *,
    swarm_id: Optional[str] = None,
    root_task_id: Optional[str] = None,
) -> Dict[str, Any]:
    """Map an ACCEPTED proposal into a pending-task spec for the integrator to enqueue.

    Returns a kwargs-shaped dict aligned with ``task_queue.create_task`` (description, agent_role,
    required_capabilities, depends_on, parent_task_id, root_task_id, source, context). The
    integrator calls ``task_queue.create_task(**spec_to_create_task_kwargs(spec))`` (or maps it
    through ``create_tasks_for_run``). The new task starts PENDING — create_task already sets that.

    The full lineage is carried into ``context`` so the proposed task is auditable: which agent
    proposed it, why, from which origin task, on which trigger, and the shared-state snapshot.
    Raises if the proposal is not in ACCEPTED status — merged/rejected proposals never become tasks.
    """
    if proposal.status != ProposalStatus.ACCEPTED:
        raise ValueError(
            f"ingest_accepted_proposal requires ACCEPTED status, got {proposal.status.value}"
        )
    assert_not_master_origin(proposal)

    # parent/root: a proposal born from an origin task is a child of it by default; otherwise it
    # is a new root. The integrator may override root_task_id to thread it into an existing run.
    parent = proposal.lineage.origin_task_id
    root = root_task_id or proposal.lineage.origin_task_id or None

    return {
        "task_id": None,  # let create_task mint a uuid; integrator may prefix with swarm_id
        "title": proposal.title or proposal.description[:80],
        "description": proposal.description,
        "agent_role": proposal.agent_role,
        "required_capabilities": list(proposal.required_capabilities),
        "depends_on": list(proposal.depends_on),
        "parent_task_id": parent,
        "root_task_id": root,
        "source": PROPOSED_SOURCE,
        "context": {
            "source": PROPOSED_SOURCE,
            "agent_role": proposal.agent_role,
            "swarm_id": swarm_id,
            "proposal": {
                "proposal_id": proposal.proposal_id,
                "proposed_by_agent_id": proposal.proposed_by_agent_id,
                "proposal_reason": proposal.proposal_reason,
                "proposal_confidence": proposal.proposal_confidence,
                "origin_task_id": proposal.lineage.origin_task_id,
                "trigger_event": proposal.lineage.trigger_event,
                "shared_state_snapshot": dict(proposal.lineage.shared_state_snapshot),
            },
        },
    }


# --- lifecycle events -------------------------------------------------------------------

def build_proposal_event(event_type: str, proposal: TaskProposal, **extra: Any) -> Dict[str, Any]:
    """Build one proposal-lifecycle event payload.

    Returns a flat payload dict (the shape the orchestrator passes to
    ``swarm_runtime.emit_event(run, event_type, payload=...)``). ``event_type`` must be one of
    the EVENT_* constants. The integrator owns wrapping this in the HM event envelope.
    """
    valid = {EVENT_SUBMITTED, EVENT_ACCEPTED, EVENT_REJECTED, EVENT_MERGED}
    if event_type not in valid:
        raise ValueError(f"unknown proposal event_type {event_type!r}; expected one of {valid}")
    payload: Dict[str, Any] = {
        "event_type": event_type,
        "proposal_id": proposal.proposal_id,
        "proposed_by_agent_id": proposal.proposed_by_agent_id,
        "source": PROPOSED_SOURCE,
        "status": proposal.status.value,
        "title": proposal.title or proposal.description[:80],
        "proposal_reason": proposal.proposal_reason,
        "proposal_confidence": proposal.proposal_confidence,
        "origin_task_id": proposal.lineage.origin_task_id,
        "trigger_event": proposal.lineage.trigger_event,
        "decision_reason": proposal.decision_reason,
        "merged_into_task_id": proposal.merged_into_task_id,
    }
    payload.update(extra)
    return payload


def build_lifecycle_events_for_outcome(
    proposal: TaskProposal,
    outcome: ReviewOutcome,
) -> List[Dict[str, Any]]:
    """Build the ordered event list for a reviewed proposal: submitted -> decision event.

    Always emits ``task.proposal_submitted`` first (the agent's act of proposing is itself an
    auditable fact), then the decision event matching the policy outcome.
    """
    events = [build_proposal_event(EVENT_SUBMITTED, proposal)]
    decision_event = {
        ProposalDecision.ACCEPT: EVENT_ACCEPTED,
        ProposalDecision.REJECT: EVENT_REJECTED,
        ProposalDecision.MERGE: EVENT_MERGED,
    }[outcome.decision]
    extra: Dict[str, Any] = {}
    if outcome.merge_target_task_id:
        extra["merge_target_task_id"] = outcome.merge_target_task_id
    events.append(build_proposal_event(decision_event, proposal, **extra))
    return events


# --- guardrail --------------------------------------------------------------------------

def assert_not_master_origin(proposal: TaskProposal) -> None:
    """Fail loudly if a proposal looks like it came from a Master/planner, not an agent.

    Issue #7 is explicitly about the AGENT proposing. A Master generating tasks is the existing
    planner path and must not be laundered through this module (it would mis-attribute source and
    bypass the planner's own contract). We reject any proposer id / trigger that names a master or
    planner. This is a cheap, explicit invariant — not a substitute for auth.
    """
    proposer = (proposal.proposed_by_agent_id or "").lower()
    trigger = (proposal.lineage.trigger_event or "").lower()
    banned = ("master", "planner")
    if any(tok in proposer for tok in banned) or any(tok in trigger for tok in banned):
        raise ValueError(
            "TaskProposal must originate from an executing agent, not a Master/planner "
            f"(proposed_by_agent_id={proposal.proposed_by_agent_id!r}, "
            f"trigger_event={proposal.lineage.trigger_event!r})"
        )