events

def parse_lines(lines: Iterable[str]) -> tuple[list[TeamEvent], list[str]]:
    """Parse JSONL ``lines`` into events, collecting a warning per malformed line.

    Deterministic: each line is validated as a :class:`~strands_teams.models.TeamEvent`;
    a line that is not valid JSON or not a valid event is skipped and a warning
    string (naming the 1-based line number) is recorded rather than raised
    (AC-3-5). Returns ``(events, warnings)`` in stable line order.
    """
    events: list[TeamEvent] = []
    warnings: list[str] = []
    for i, line in enumerate(lines, start=1):
        try:
            events.append(TeamEvent.model_validate_json(line))
        except (ValueError, TypeError) as e:
            warnings.append(f"line {i}: skipped malformed event ({type(e).__name__})")
    return events, warnings

def replay(events: Iterable[TeamEvent], *, warnings: Iterable[str] = ()) -> ReplayState:
    """Reconstruct run state from events with no live model call (AC-3-1..AC-3-5).

    Applies each event only after all its causal parents (:func:`_causal_order`,
    AC-3-3), records the transition sequence, tracks the latest non-superseded
    event per artifact binding, reconstructs finding and decision payloads, and
    reflects supersession (an event bearing ``epistemic_status="superseded"`` or an
    event naming another in its ``causal_parents`` with an
    ``epistemic_status`` in ``{accepted, rejected, superseded}`` supersedes that
    parent) while retaining the originals in the log (AC-3-4).

    Deterministic: given the same events (and malformed-line ``warnings``), the
    returned :class:`ReplayState` is identical across calls (AC-3-2).

    Args:
        events: The parsed events to replay (typically ``EventLog.read()``).
        warnings: Malformed-line warnings from parsing, folded into the result so
            :meth:`ReplayState.to_canonical_json` reflects them (AC-3-5).

    Returns:
        A :class:`ReplayState` reconstruction.
    """
    event_list = list(events)
    ordered, order_warnings = _causal_order(event_list)

    applied: list[str] = []
    transitions: list[str] = []
    artifacts: dict[str, str] = {}
    findings: dict[str, dict[str, Any]] = {}
    decisions: dict[str, dict[str, Any]] = {}
    superseded: set[str] = set()

    for event in ordered:
        applied.append(event.event_id)
        transitions.append(event.event_type)

        # Supersession (AC-3-4): a superseded-status event marks itself; an
        # accepted/rejected/superseded event supersedes the parents it references.
        if event.epistemic_status == "superseded":
            superseded.add(event.event_id)
        if event.epistemic_status in {"accepted", "rejected", "superseded"}:
            superseded.update(event.causal_parents)

        if _is_finding_event(event):
            findings[event.event_id] = dict(event.payload)
        if _is_decision_event(event):
            decisions[event.event_id] = dict(event.payload)

    # Artifacts: latest non-superseded event that touches each binding, in apply order.
    for event in ordered:
        if event.event_id in superseded:
            continue
        for binding in event.artifact_bindings:
            artifacts[binding] = event.event_id

    return ReplayState(
        applied=applied,
        transitions=transitions,
        artifacts=artifacts,
        findings=findings,
        decisions=decisions,
        superseded=sorted(superseded),
        warnings=[*warnings, *order_warnings],
    )

def replay_log(log: EventLog) -> ReplayState:
    """Replay an :class:`EventLog` end to end, surfacing malformed-line warnings (AC-3-5)."""
    events, warnings = log.read_with_warnings()
    return replay(events, warnings=warnings)

def should_replan(events: Iterable[TeamEvent], *, budget_threshold: float = 0.0) -> ReplanDecision:
    """Decide whether the lead loop should replan (AC-6-1..AC-6-7).

    Scans the event log for any of the five ASTR triggers — ``requirement.changed``,
    ``critical_finding.created``, ``task.failed``, ``budget.threshold_exceeded``,
    ``artifact.interface_changed`` — matching on either an explicit trigger-typed
    event or the structured payload of a generic dispatch/finding/revision event.
    Cumulative recorded cost crossing ``budget_threshold`` fires the budget trigger
    (AC-6-4). Returns ``replan=False`` with no triggers on a trigger-free log
    (AC-6-7).

    Args:
        events: The events to evaluate (typically ``EventLog.read()``).
        budget_threshold: The configured cumulative-cost bound; ``0`` disables the
            cumulative budget check (an explicit ``budget.threshold_exceeded`` event
            still fires).

    Returns:
        A :class:`ReplanDecision` naming the distinct triggers that fired.
    """
    event_list = list(events)
    fired: dict[str, str] = {}

    for event in event_list:
        hit = _event_fires(event)
        if hit is not None:
            trigger, reason = hit
            fired.setdefault(trigger, reason)

    if _budget_exceeded(event_list, budget_threshold=budget_threshold):
        fired.setdefault(TRIGGER_BUDGET_EXCEEDED, f"cumulative cost crossed threshold {budget_threshold}")

    ordered = [t for t in REPLAN_TRIGGERS if t in fired]
    return ReplanDecision(
        replan=bool(ordered),
        triggers=ordered,
        reasons=[fired[t] for t in ordered],
    )