onlyne-client 2.0.0

use super::*;

use super::projection::{note_verdict, sync_session};
use super::retire::on_recycled;
use super::settle::{SettleAuthority, on_out};
use super::state::{ControlWord, note_beat};
use super::transport::{serves_ending, serves_task};
use super::turn_end::on_turn_end;

/// Act on one control command that arrived as a delivery.
///
/// `recycle` and `cancel` reach the agent first, so it ends its own turn and
/// sends its terminal report, and the backend close follows whatever the plugin
/// did: a session whose adapter is gone still loses its process, which is the
/// half of §7's recovery ladder the operator drives by hand otherwise. `probe`
/// asks the plugin for a fresh observation and, when that question went out,
/// republishes the projection the reducer already holds, `snapshot` republishes
/// alone, and `focus` asks the backend to bring the live session to the front.
///
/// Answers whether the command named a session this client holds. A `false` is
/// the honest answer for a task this role does not own, and the caller still
/// settles the row: re-offering a command no client can act on spends the
/// delivery forever.
///
/// The same question decides whether anything is *applied*. A command naming a
/// task this role has never held acts on nothing and writes no note: `recycle`
/// and `cancel` put a word in `control_settles` that the settle door and the
/// watchdog both read as this client's own order, and an order accepted for a
/// session the role does not serve is a write to a task somebody else owns. The
/// answer is `Ok(false)` rather than an error because the row still has to
/// settle: a command refused here is refused for good, and acking it as
/// undelivered would have the server offer a command no client can ever act on.
pub async fn on_control(state: &DispatchState, op: &ControlOp) -> Result<bool> {
    let task_id = op.task_id();
    let held = state.holds_task(task_id);
    if !held {
        tracing::warn!(
            op = op.name(),
            task = %task_id,
            "a control command naming no session this role holds was refused; nothing was applied"
        );
        return Ok(false);
    }
    match op {
        ControlOp::Recycle { reason, .. } => {
            // The command asks this session's plugin for its own ending, so the
            // completion that answers it travels the plugin's door carrying the
            // client's authority. The note goes on before the frame leaves: the
            // report and the retirement below race over the row, and the note is
            // the half of that race this client decides. A recycle prescribes the
            // plugin no outcome, so the verdict the note stands for is the one a
            // session that died holding a task leaves: `failed`.
            state.owe_controlled_settle(task_id, ControlWord::Recycle, Instant::now());
            state.recycle_plugin(task_id, reason, None).await;
            on_recycled(state, task_id, crate::backend::CloseReason::Operator)?;
        }
        ControlOp::Cancel { reason, .. } => {
            // The same note for the same reason: a cancel ends the task on the
            // operator's word, and the plugin's `cancelled` report answers it.
            state.owe_controlled_settle(task_id, ControlWord::Cancel, Instant::now());
            state
                .recycle_plugin(task_id, reason, Some(Outcome::Cancelled))
                .await;
            on_recycled(state, task_id, crate::backend::CloseReason::Cancelled)?;
        }
        ControlOp::Probe { .. } => {
            // A probe that found no transport asked nothing, so there is no fresh
            // observation to publish: republishing here would write the projection
            // an answer would have earned for a session nobody serves.
            if state.probe_plugin(task_id).await {
                sync_session(state, task_id).await?;
            }
        }
        ControlOp::Snapshot { .. } => sync_session(state, task_id).await?,
        ControlOp::Focus { .. } => {
            let (backend, session) = {
                let inner = state.inner.lock();
                let session = inner
                    .sessions
                    .values()
                    .find(|slot| slot.task_id.as_deref() == Some(task_id))
                    .map(|slot| slot.session.clone());
                (inner.backend.clone(), session)
            };
            match session {
                Some(session_ref) => {
                    if let Err(error) = backend.focus(&session_ref) {
                        tracing::warn!(error = %error, task = %task_id, "focus refused");
                        if let Err(fault_error) = on_plugin_report(
                            state,
                            // The client files this fault itself, so no plugin
                            // connection is its sender.
                            None,
                            Report::Fault {
                                task_id: Some(task_id.to_string()),
                                session_id: None,
                                generation: None,
                                seq: None,
                                kind: "focus".into(),
                                reason: error.to_string(),
                                desired: None,
                                observed: None,
                            },
                        )
                        .await
                        {
                            tracing::warn!(
                                error = %fault_error,
                                task = %task_id,
                                "focus refused"
                            );
                        }
                    }
                }
                None => {
                    tracing::warn!(task = %task_id, "focus has no live session");
                }
            }
        }
    }
    Ok(true)
}

/// `from` is the connection the frame arrived on, which is what decides whether
/// the report may move the session it names: only the connection serving a
/// session speaks for it, and a connection this client holds read-only is served
/// no state at all (§1 (b)). `None` is a report this client composes for itself —
/// the fault a refused `focus` files — and it is the session's own authority.
///
/// A refusal is not an error the sender reads: the frame is answered exactly as
/// an applied one is, because a plugin treats a failed report as a link that
/// died and sends the same thing again. What a refused state frame does instead
/// of being applied is leave the tuple where the serving connection left it and
/// say so here, in the log the operator reads.
///
/// The door is the same question in all three state-carrying arms — heartbeat,
/// completion, fault — because all three move the session they name: the first
/// its observed dimensions, the other two its verdict and the close of its
/// resource. Only the completion and the fault answer for an extra case, where
/// the client has already retired the session on its own command; see
/// `ending_is_authorised`.
pub async fn on_plugin_report(
    state: &DispatchState,
    from: Option<&AdapterIo>,
    report: Report,
) -> Result<()> {
    let subject = task_id_of(&report).to_string();
    let touched = match report {
        Report::Ready { task_id, .. } => {
            let verdict = {
                let inner = state.inner.lock();
                feed_ready(&inner.bridge, &inner.store, &task_id)?
            };
            note_verdict(&verdict, &task_id).is_some()
        }
        Report::Heartbeat {
            task_id,
            // The plugin's own generation is deliberately unread. It is a
            // constant the plugin never raises, and the beat's version takes the
            // generation the session's tuple holds instead — see the stamp
            // below.
            generation: _,
            seq,
            observed,
            ..
        } => {
            // The beat itself is the liveness fact. The server times
            // heartbeats. A quiet, alive session keeps landing fresh rows.
            let (touched, ended) = {
                let mut inner = state.inner.lock();
                if from.is_some_and(|io| !serves_task(&inner, &task_id, io)) {
                    // A connection the client holds read-only is not this session's
                    // transport, so its observation is not this session's state: the
                    // whole point of the demotion is that the task answers through
                    // the connection serving it now. The beat travels no further than
                    // this refusal — a beat has no recipient to hold it for, unlike
                    // the `send` of §1 (c) — and it is logged rather than dropped in
                    // silence, because a supervisor reading the trail of a session
                    // whose tuple stopped moving wants to see who was talking.
                    // Its liveness half does travel, and that half is what the
                    // agent's life depends on: the stamp the silence arm reads is
                    // about the socket behind it, and both copies of a plugin loaded
                    // into one agent process — the shape a workspace that installs
                    // the plugin twice produces — beat on their own connections.
                    // Letting a refused frame leave the stamp alone starves the
                    // session's clock, and the sweep then retires an agent that is
                    // alive and working. The stamp is read only for a session whose
                    // task is still bound and unsettled, so the demotion keeps its
                    // own retirement.
                    note_beat(&mut inner, &task_id, Instant::now());
                    // The wording is the load-bearing part here. An earlier
                    // sentence said "serving no such session", and it is wrong
                    // about the commonest case that reaches here: a pooled
                    // session that settled its delivery still has the plugin
                    // holding and beating for that task, so every beat lands
                    // here while the session sits in the pool reading perfectly
                    // healthy. Read literally it says the client lost a session
                    // it still has, and an operator chasing that looks for a
                    // leak instead of reading the line for what it is.
                    tracing::warn!(
                        task = %task_id,
                        "a beat arrived on a connection that is not this task's transport: \
                         the liveness stamp is refreshed and no state is applied"
                    );
                    // A beat that speaks for no session of this role witnessed no
                    // ending this client may act on.
                    (false, false)
                } else {
                    // The beat's version is the session's own generation beside the
                    // reporter's sequence, never the plugin's generation field. That
                    // field is a constant a plugin never raises, so taking it
                    // verbatim made every frame of a session whose generation had
                    // moved — which is what a returning agent's rebase does — read
                    // as a stale generation and be dropped. The session's tuple is
                    // the authority on which generation is reporting, and the
                    // sequence stays the reporter's own: it is the only part of the
                    // watermark the plugin is the witness of.
                    let row = inner.store.get_session(&task_id).ok().flatten();
                    let stored = stored_observation(&inner.store, row.as_ref());
                    let generation = stored.version.generation;
                    // The task this beat speaks for, as this client holds it. No
                    // record means nothing was ever opened here for that id, which is
                    // not the same fact as an open task: the composition labels open
                    // work, and there is none to label without a record.
                    let task_state = inner
                        .store
                        .task(&task_id)
                        .ok()
                        .flatten()
                        .map(|record| record.task_state);
                    // A beat at or below the newest version this task's reporter has
                    // been seen at is a replay of a frame already taken. The row no
                    // longer moves for a beat that observed nothing, so its version
                    // cannot answer this on its own: the watermark lives beside the
                    // liveness stamp, and a replay buys only the stamp.
                    if !inner.stall.note_beat_seq(&task_id, generation, seq) {
                        note_beat(&mut inner, &task_id, Instant::now());
                        tracing::debug!(
                            task = %task_id,
                            generation,
                            seq,
                            "a heartbeat at or below the watermark was taken as a replay; it refreshes liveness only"
                        );
                        return Ok(());
                    }
                    let (ended, verdict) = match serde_json::from_value::<Observation>(observed) {
                        Ok(body) => {
                            let composed = compose_observation(&stored, body, task_state);
                            // §3c's turn-end witness: this beat moved the agent out
                            // of a turn and into the wait that follows one, over a
                            // task this client still holds open, with no completion
                            // exit standing anywhere. `idle_waiting` is the
                            // composition's own word for that pair, and the
                            // transition is what makes this beat an ending rather
                            // than one more idle report from a session that already
                            // waits.
                            let ended = stored.agent == AgentPhase::Running
                                && composed.agent == AgentPhase::Idle
                                && task_state == Some(TaskState::Pending)
                                && composed.recovery == RecoveryPhase::IdleWaiting;
                            let verdict = apply_persist(
                                &inner.bridge,
                                &inner.store,
                                &task_id,
                                &LifecycleEvent::Heartbeat {
                                    v: Version::new(generation, seq),
                                    body: composed,
                                },
                            )?;
                            (ended, verdict)
                        }
                        Err(error) => {
                            tracing::warn!(task = %task_id, error = %error, "heartbeat carries no readable observation; liveness only");
                            // The log line's own promise: a frame whose observation no
                            // client can read still proves the agent is alive, so the
                            // stamp goes on and the tuple stays where it stood.
                            note_beat(&mut inner, &task_id, Instant::now());
                            (false, Verdict::Ignored(IgnoredReason::NoOp))
                        }
                    };
                    note_verdict(&verdict, &task_id);
                    let touched = match &verdict {
                        Verdict::Applied(_) => {
                            note_beat(&mut inner, &task_id, Instant::now());
                            inner.stall.note_applied(&task_id, Instant::now());
                            true
                        }
                        Verdict::Ignored(IgnoredReason::NoOp) => {
                            // A beat that moves no dimension is still the agent saying it is
                            // alive, and this is where the ordinary long turn lands: a model
                            // streaming for minutes reports `agent: running` over and over, the
                            // tuple never changes, and every one of those beats is a no-op to the
                            // reducer. Leaving the stamp alone for them starves the clock the
                            // silence arm reads, and the sweep then closes the pane under an agent
                            // that is working — the shape a live role died of at thirty seconds
                            // into a turn. So the stamp goes on, and the frame travels: it is
                            // what refreshes the mirror's own `last_seen`.
                            //
                            // The row itself does not move. The plan's rule for a heartbeat is
                            // that it refreshes `last_seen` and that only changed content is
                            // persisted — and `(generation, seq)` beside `updated_at` is the
                            // watermark projection writes are ordered by, not a liveness
                            // counter. Advancing it here made a beat that observed nothing read
                            // as a fresh write: a reader saw `updated_at` move and could not
                            // tell a session that had changed from one that had merely said it
                            // was still there.
                            note_beat(&mut inner, &task_id, Instant::now());
                            true
                        }
                        Verdict::Ignored(_) | Verdict::Rejected(_) => false,
                    };
                    // A beat the reducer refused moved nothing, so it is no ending
                    // either.
                    (touched, ended && !matches!(verdict, Verdict::Rejected(_)))
                }
            };
            // §3c's rule runs off the lock: it either hands the one nudge to the
            // session's drive, or settles the delivery blocked.
            if ended {
                on_turn_end(state, &task_id, None).await?;
            }
            touched
        }
        Report::Complete {
            task_id,
            outcome,
            head,
            details,
            ..
        } => {
            // A plugin reports its own ending. The one completion this client has
            // asked for by name arrives on the same frame, so the note is what
            // tells the two apart. Everything else that leaves this arm is the
            // plugin's own claim about work it did, and `on_out` reads the
            // session's row for that claim, refused whole when the row says no
            // turn ran.
            if !ending_is_authorised(state, from, &task_id) {
                tracing::warn!(
                    task = %task_id,
                    "a completion from a connection serving no such session was refused"
                );
                return Ok(());
            }
            let asked = if state.take_controlled_settle(&task_id) {
                SettleAuthority::ControlDriven
            } else {
                SettleAuthority::PluginReport
            };
            on_out(state, &task_id, outcome, head, details, asked).await?;
            false
        }
        Report::Fault {
            task_id: Some(task_id),
            kind,
            reason,
            ..
        } => {
            // A recorded fault is a verdict too: it feeds the mismatch counter and
            // the isolate/terminate ladder of the session it names, so the
            // connection that files one is held to the same door as the connection
            // that ends one.
            if !ending_is_authorised(state, from, &task_id) {
                tracing::warn!(
                    task = %task_id,
                    kind = %kind,
                    "a fault from a connection serving no such session was refused"
                );
                return Ok(());
            }
            let inner = state.inner.lock();
            crate::reconcile::record_fault(&inner.store, &task_id, &kind, "plugin", &reason)?;
            false
        }
        Report::Fault { task_id: None, .. } => false,
    };
    if touched {
        sync_session(state, &subject).await?;
    }
    Ok(())
}

/// Whether the sender of a report that ends or faults a session may do it.
///
/// The strict half is the heartbeat's rule unchanged: only the connection
/// serving a session speaks for it, and `None` is the client's own door — the
/// local CLI's `report`, and the fault a refused `focus` files — which is the
/// session's own authority. `serves_ending` carries the two windows where an
/// ending is legitimate and the strict rule cannot see its sender: the connection
/// held read-only for this task, and the completion that answers this client's
/// own `control` command, which reaches this door after the command has already
/// retired the slot it would have been checked against.
fn ending_is_authorised(state: &DispatchState, from: Option<&AdapterIo>, task_id: &str) -> bool {
    let Some(io) = from else {
        return true;
    };
    let inner = state.inner.lock();
    serves_ending(&inner, task_id, io)
}

/// Compose one plugin heartbeat into the tuple the reducer is allowed to read.
///
/// `client` is the stored tuple the caller already read through
/// `stored_observation` (`client.db`'s `sessions` row) — the same read the beat's
/// version is stamped from, so one frame costs one read of the row it names.
///
/// A heartbeat is authority about exactly three dimensions: the `agent` state
/// the plugin's own turn hooks witnessed, the `resource` it knows because its
/// process is live in it, and the `host` — the pane only this process can name
/// from inside. `delivery`, `recovery`, `generation_live`, `isolate_after`,
/// `terminate_after` and `mismatch_count` are not in that set. The completion
/// intent is created, retried and receipted by this client's own settle path
/// (`settle.rs`, fed from `report.complete`; `onlyne-session` exposes no
/// `Intent*` feed to any other caller), the recovery substate is the label
/// that drain and the reconcile loop carry, and the reconcile tuning with the
/// counter beside it is the role's own configuration — so all six come from the
/// stored row this client already holds, and never from the beat.
/// That row is not the reducer's last state: a beat the reducer ignored is still
/// liveness, so `bump_session_version` raises the row's `(generation, seq)`
/// watermark with no reduce behind it, and the columns can stand ahead of the
/// observation written beside them. Nothing here reads the version — the copy is
/// the six dimensions alone. What the body claims for them is discarded, never
/// read: a plugin that says `delivery: none` because it cannot see the drain
/// must not be able to clear an intent the server has not receipted.
///
/// `task_state` is the task record's own verdict, read from the store beside the
/// row (`None` when this client holds no record for the task). It is here
/// because `idle_waiting` is a fact about the work, not about the tuple: the
/// label says a turn ended while an open task had no completion exit.
///
/// The repairs after the copy are what makes the result legal by construction,
/// and they are independent: one reads the drain, the other the label beside it,
/// and a beat can break both pairings at once. The reducer replaces the agent
/// wholesale and `is_legal` is the tuple's own cross-constraint (§2.2), so a
/// client drain carried onto a dimension the plugin just moved can pair two
/// facts no transition would ever have written together. Each arm below names a
/// pairing the reducer's own rules forbid and downgrades the *client* dimension
/// to the state of that drain the plugin's claim proves stale — never to a
/// value invented for the moment. The consequence at this door is that a beat is
/// composed into legality rather than refused by it; `RejectReason::IllegalObservation`
/// is still the reducer's answer to every other feed, the reconcile loop's own
/// heartbeats included.
fn compose_observation(
    client: &Observation,
    mut body: Observation,
    task_state: Option<TaskState>,
) -> Observation {
    body.delivery = client.delivery;
    body.recovery = client.recovery;
    // The reconcile tuning and its counter are the client's as well, and for the
    // same reason: `isolate_after` and `terminate_after` are the role's own
    // thresholds and `mismatch_count` is what this client's loop increments each
    // time a fact disagrees. A plugin witnesses none of them — the shape it sends
    // carries the defaults and a constant `generation_live` — so a beat that
    // carried them would reset a running isolation ladder on every heartbeat.
    body.generation_live = client.generation_live;
    body.isolate_after = client.isolate_after;
    body.terminate_after = client.terminate_after;
    body.mismatch_count = client.mismatch_count;
    // `Exhausted` means retries burned against an open turn exit, and `Ready`
    // says no turn exit is open. A plugin that moves a session with an exhausted
    // intent back to `ready` has begun a new assignment: the old drain was
    // consumed by whatever settled it, or by the fault path that took the work,
    // and only the exit it hung on is gone.
    if body.agent == AgentPhase::Ready && body.delivery == DeliveryPhase::Exhausted {
        body.delivery = DeliveryPhase::NoIntent;
    }
    // `Accepted` is a post-turn fact and `Booting` is the word a plugin uses for
    // a process that is not there any more (see `observationFor`'s `gone`
    // mapping). The receipt the client wrote cannot belong to a turn that
    // stopped existing, so the drain stays where the plugin left it: open,
    // unacknowledged.
    if body.agent == AgentPhase::Booting && body.delivery == DeliveryPhase::Accepted {
        body.delivery = DeliveryPhase::Pending;
    }
    // A recovery substate rides only the agents `is_legal` allows it on, which is
    // the reducer's own coupling for those transitions (`reduce.rs::transition`),
    // applied here because a heartbeat body bypasses it: `Idle` for the two
    // waiting labels, `Idle` or `Running` for `Draining` — a completion in
    // asynchronous send outlives the turn that reported it — and none of them on
    // `Booting`, `Ready` or `Gone`, whose own definitions say no turn is open to
    // be waiting on or draining from. The `Running` arm is not decoration: the
    // beat of a turn that started after a reminder is exactly that pair, and
    // dropping the label is what keeps the reducer from refusing the beat whole —
    // the frame that says the session went back to work would be the one lost.
    let held = match body.recovery {
        RecoveryPhase::IdleWaiting | RecoveryPhase::IdleFault => body.agent == AgentPhase::Idle,
        RecoveryPhase::Draining => matches!(body.agent, AgentPhase::Idle | AgentPhase::Running),
        RecoveryPhase::NoRecovery => true,
    };
    if !held {
        body.recovery = RecoveryPhase::NoRecovery;
    }
    // An idle agent whose task is still open and whose exit has no receipt is
    // the design's `idle_waiting` (§2.2, §4.2): the turn ended without a
    // completion exit, and the plugin's answer is to send the assignment again.
    // No other writer reaches the label on a live session — the reducer's own
    // turn-end rule (`crates/onlyne-session/src/lifecycle/reduce.rs`,
    // `transition`) needs a `TurnEnded` event that no plugin-backed session
    // feeds, and a beat carries the agent state as its only news — so the
    // composition is where that rule has to land or `idle_waiting` never
    // appears at all. A stronger label the client already holds is left alone:
    // `draining` says the exit is in asynchronous send and `idle_fault` says a
    // fact disagreed, while `idle_waiting` means there is no exit yet.
    if body.agent == AgentPhase::Idle
        && task_state == Some(TaskState::Pending)
        && body.delivery != DeliveryPhase::Accepted
        && body.recovery == RecoveryPhase::NoRecovery
    {
        body.recovery = RecoveryPhase::IdleWaiting;
    }
    body
}

/// Task a state-carrying report names, for the projection publish.
fn task_id_of(report: &Report) -> &str {
    match report {
        Report::Ready { task_id, .. } | Report::Heartbeat { task_id, .. } => task_id,
        Report::Complete { task_id, .. } => task_id,
        Report::Fault { .. } => "",
    }
}
#[cfg(test)]
mod tests;