lernie 0.1.45

//! **Holding the line on one conversation** — the seat's own word for
//! watching work happen, which is more than one gesture.
//!
//! # What was wrong, and both halves were the same mistake
//!
//! `follow` was one `ask`: one held connection, its frames collected, the
//! stream printed at the end, and the exit code read off the last frame. That
//! gave two defects with one cause.
//!
//! - **On a quiescent conversation it blocked 31 seconds, printed one newline
//!   and exited 1** (bl-3dca). Nothing is wrong with a conversation at rest —
//!   it is the ordinary state of one an operator has just looked at — but exit
//!   1 with no output is byte for byte what a failed dial, a refused
//!   certificate and a dead engine look like, and those are what a first-time
//!   user suspects.
//! - **On a live one it returned mid-turn, every time** (bl-f076). The engine
//!   ends the stream at the STEP boundary (yog's REMOTE §5.1), not at the
//!   turn's, so three consecutive reads of one conversation ended after 3, 8
//!   and 6 seconds while it worked on. Watching an agent was
//!   `while true; do lernie follow …; done`, which the operator had to invent
//!   and which loses whatever landed between two reads.
//!
//! # The rule this file implements
//!
//! **Hold the line until the conversation rests, or until the user quits.**
//! The engine's own step boundary is not the end of anything an operator
//! cares about, so a read that ends there is re-asked; what ends the WORD is
//! the conversation coming to rest, and that is a fact the engine already
//! answers — `agent`'s `state`. So the loop is: ask what it is doing, and if
//! it is not working, say so and exit 0; otherwise hold a read and print what
//! lands, then ask again.
//!
//! **The state read comes first, which is what fixes the quiescent case at no
//! cost.** A conversation already at rest never opens a held connection at
//! all, so the half-minute of silence is not shortened — it never happens.
//!
//! **An unknown state ends the follow rather than looping on it.** The reply
//! vocabulary paints an unrecognised token as itself (DESIGN §4.9 rung 3), and
//! the safe reading here is *this seat does not know that this is working* —
//! ending, and printing the word, beats holding a connection open forever on a
//! state nobody here understands.
//!
//! # Mail waiting is not rest (bl-87ab, bl-3ecd)
//!
//! `message` (or `start`) then `follow` is the pair every operator types, and
//! it answered *"nothing more will arrive until it is nudged or messaged"* in
//! zero seconds — about a conversation that took a lease twelve seconds later
//! and then ran for another seventy-eight. The deposit had landed and no
//! driver held the inbox lock yet, so the state read was correct as of that
//! instant and the sentence it produced was false: something more was going to
//! arrive, and nothing further was going to be asked of the operator.
//!
//! The engine has no field for *about to start* and inventing one here would
//! be a guess. What it does have is the deposit itself: **an inbox with mail in
//! it is a turn that has not begun**, so the state read alone was never enough
//! and the second read is what makes the answer true. So a rest with mail
//! waiting is not an ending — the watch holds, exactly as it holds on a live
//! one, and says once that it is waiting so the hold never reads as a hang.
//! Ctrl-C is the way out it always was.
//!
//! **It costs one read, and only on the path that was wrong.** A conversation
//! genuinely at rest has an empty inbox and answers as fast as it ever did;
//! nothing is asked before the state read, and nothing extra is asked while a
//! conversation is working. And a mail count this seat could not READ is
//! treated as no mail: the state read has already said rest, and a probe that
//! failed is not grounds for holding somebody's connection open forever.
//!
//! # `--json` narrates nothing (bl-87ab)
//!
//! `--help` promises *"the frames exactly as they crossed, one envelope per
//! line, which is what a script wants"*, and this word broke it on the one
//! line that mattered: every frame of a watch was an envelope and the line
//! saying the watch was OVER was prose, so a reader calling `json.loads` on
//! each line died on the terminator. The rule is now one rule for both
//! sentences this file writes — under `--json` this seat says nothing of its
//! own, and what ends the watch is the `agent` frame that ENDED it, printed as
//! it crossed. It carries the state in a field, which is what a script wanted
//! from the sentence. The elapsed is this end's own fact and a script has its
//! own clock; the rendered form keeps it.
//!
//! # Why the frames are handed to a sink
//!
//! A held read that printed only at the end would not be a follow. So this is
//! the one place in the crate where the product is written as it arrives, and
//! the writing stays the entry point's: [`follow`] takes a sink, `src/main.rs`
//! hands it a printer, and the suite hands it a `Vec`. The decision is still
//! entirely in the library, where a test reads it back.

use std::path::Path;
use std::time::{Duration, Instant};

use serde_json::Value;

use crate::channel::Reach;
use crate::cli::Verdict;
use crate::render::{Form, said};
use crate::reply::convs::AgentState;
use crate::reply::stream::Stream;
use crate::reply::{Read, Reply, read};

/// **How long a read that brought nothing waits before asking again.**
///
/// The engine paces this loop on its own — a held read stays open while the
/// tail is quiet — so this is insurance against an engine that closes an empty
/// read at once, where the alternative is a busy loop on somebody else's
/// socket. It is not a poll interval: a read that brought a frame asks again
/// immediately.
const SETTLE: Duration = Duration::from_millis(250);

/// **Watch one conversation until it rests.** `say` takes each frame's
/// rendering as it lands; the verdict is the sentence that ends the watch.
pub fn follow(
    data_root: &Path,
    workspace: &str,
    agent: &str,
    form: Form,
    say: &mut dyn FnMut(&str),
) -> Verdict {
    let asking = crate::verbs::agent(workspace.to_owned(), agent.to_owned());
    let holding = crate::verbs::follow(workspace.to_owned(), agent.to_owned());
    let mail = crate::verbs::inbox(workspace.to_owned(), agent.to_owned());
    // **What the closing line's elapsed is measured from** (bl-293d): the
    // whole of what this seat can say a duration about. The engine's own
    // turn started before the watch did and is not this end's to time.
    let began = Instant::now();
    let mut told = false;
    loop {
        let standing = match super::sent(data_root, &asking) {
            Ok(frames) => frames,
            Err(reach) => return Verdict::failed(reach.said()),
        };
        match resting(&standing) {
            // The engine answered something this build cannot read a state
            // out of — a refusal, or a kind it does not paint. That answer is
            // the product and the exit code is what says the watch never
            // started.
            Rest::Unreadable => return Verdict::answered(said(&standing, form), false),
            Rest::At(state) => match waiting(data_root, &mail) {
                // Rest with an empty inbox is the ending: nothing is coming.
                0 => return Verdict::ok(ending(&standing, agent, &state, began.elapsed(), form)),
                // Rest with mail in it is a turn about to begin. Hold, and say
                // so once — a silent hold is the thing this word was fixed
                // for, and saying it every quarter second would be worse than
                // silence.
                deposits => {
                    if !told && matches!(form, Form::Rendered) {
                        told = true;
                        say(&crate::render::waiting_on_mail(agent, &state, deposits));
                    }
                }
            },
            Rest::Working => {}
        }
        match held(data_root, &holding, form, say) {
            Err(reach) => return Verdict::failed(reach.said()),
            Ok(0) => std::thread::sleep(SETTLE),
            Ok(_) => {}
        }
    }
}

/// One held read, printed as it arrives. Answers how many frames landed, which
/// is the one fact the loop above needs from it.
///
/// **The fold's whole lifetime is this call**, which is the whole of how REMOTE
/// §5.5's *"onto an empty fold"* is implemented here: one read is one `held`,
/// so a read boundary needs no flag and no field — it is a local's scope. What
/// is printed is still the frame's own append, because a terminal is already a
/// fold and re-printing the accumulation would be the same answer at quadratic
/// cost; the fold exists for the half that CANNOT be printed twice, which is
/// the tool window's closing entry ([`crate::render::tail`]).
fn held(
    data_root: &Path,
    envelope: &Value,
    form: Form,
    say: &mut dyn FnMut(&str),
) -> Result<usize, Reach> {
    let (channel, carried) = super::route(data_root, envelope)
        .sent
        .map_err(Reach::Unsent)?;
    let mut heard = 0usize;
    let mut fold = Stream::default();
    channel.follow(&carried, &mut |frame| {
        say(&crate::render::tail(&frame, &mut fold, form));
        heard += 1;
        true
    })?;
    Ok(heard)
}

/// What the standing read says the conversation is doing.
enum Rest {
    /// A driver is on it, or it is streaming. Hold the line.
    Working,
    /// It has come to rest, in the engine's own word for how.
    At(String),
    /// The engine answered something no state can be read out of.
    Unreadable,
}

/// Read the conversation's own state off its row.
fn resting(standing: &[Value]) -> Rest {
    let Some(Read::Answer(Reply::Agent(row))) = standing.last().map(read) else {
        return Rest::Unreadable;
    };
    match row.state {
        AgentState::Live | AgentState::InFlight => Rest::Working,
        settled => Rest::At(settled.label()),
    }
}

/// **How much mail is waiting to be taken**, which is what tells a rest that
/// is about to end from one that is not.
///
/// **Unreadable is NO mail, deliberately.** The state read above has already
/// said this conversation is at rest, so every way this probe can fail — a
/// channel that dropped, a refusal, an answer of a kind this build does not
/// paint — leaves the watch with exactly the reading it had before the probe
/// existed. Holding a connection open forever on the strength of a question
/// that was not answered is the one outcome worse than the sentence this
/// exists to fix.
fn waiting(data_root: &Path, mail: &Value) -> usize {
    let Ok(frames) = super::sent(data_root, mail) else {
        return 0;
    };
    match frames.last().map(read) {
        Some(Read::Answer(Reply::Inbox(rows))) => rows.len(),
        _ => 0,
    }
}

/// **The line a watch ends on**, in the form the caller asked for.
///
/// Rendered, it is this seat's own sentence, said where every other line of
/// its product is said ([`crate::render::at_rest`]) — and how long the watch
/// held is the one fact only this end has, so it is measured here and rendered
/// there. As JSON it is the `agent` frame that ended the watch, exactly as it
/// crossed: `--json` is the frame stream and this seat adds nothing to it
/// (bl-87ab).
fn ending(standing: &[Value], agent: &str, state: &str, held: Duration, form: Form) -> String {
    match form {
        Form::Json => said(standing, form),
        Form::Rendered => crate::render::at_rest(
            agent,
            state,
            i64::try_from(held.as_secs()).unwrap_or(i64::MAX),
        ),
    }
}

#[cfg(test)]
mod tests;