Meter a session by its provider, and let llama.cpp run over ssh
The rate-limit bar answered a question about an account, and picked the
answer by machine. One machine runs echo, the Claude CLI and a local
model side by side, so every echo session on it drew the CLI's five-hour
window: a quota that session cannot spend and could never run down. A
session now names its meter (`usageProvider`, from
`DriverKind::usage_provider`, which `usage::providers_for` reads too so
the two lists cannot disagree), and the phone matches on machine *and*
provider. Nothing meters echo or llama, and nothing at all is drawn --
including while the first fetch is out, since "checking" under a session
that turns out to meter nothing is a row the screen then withdraws.
Echo gets a meter it can be *told* about instead: `/usage 42`,
`/usage 95 20`, `/usage 42 never`, `/usage notloggedin`,
`/usage unreachable`, `/usage failed`, `/usage off`. Those states cost
real quota to arrange, which is why none of them had been looked at.
And llama.cpp runs wherever a setup says, which was the last of phase 5.
`Transport::reserve_port` is the second half of what a transport is --
"run this" plus "reach this port" -- returning the port the server binds
there and the port that reaches it here, and `Launch::reaching` puts the
`-L` tunnel on the connection that already carries the command. Three
things that came out of building it:
- A forwarded launch gets a pty and every other one keeps `-T`. Killing
the ssh client ends a CLI by closing the stdin it reads; llama-server
never reads its stdin, so the same kill left it running on the far
machine with the model loaded -- one orphan per stopped session.
- The model is looked for on the machine that will serve it, at that
machine's own models directory, so `GET /setups/{id}/models` is what
the spawn screen offers rather than the backend's own downloads.
- The readiness poll watches the process, not only the port: a model
that will not load exits in a second and would otherwise have been
reported as "gave up after 300s". The failure carries the log's tail.
Exercised end to end against this VM over ssh to itself: spawn, load,
answer, outlive a backend restart, be adopted, answer again, and stop --
with both the ssh client and the far llama-server gone afterwards. The
local path, the Claude bar and the spawn screen checked on the emulator.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
1 parent
74110b4d72
commit
127b25e60a
20 files changed
+1212
-143
No files matched your search
@@ -28,6 +28,13 @@
|
||||
//! - `/error [text]` -- a failure, which is otherwise awkward to cause.
|
||||
//! - `/peer [text]` -- a message from another agent, which otherwise takes
|
||||
//! two live sessions and one of them deciding to write.
|
||||
//! - `/usage [what]` -- puts up an invented rate-limit answer, or takes
|
||||
//! it away again (`/usage off`). An echo session meters nothing, so it
|
||||
//! draws no usage bar at all until this is set; what it exists for is
|
||||
//! the states the bar can be in, which otherwise cost real quota to
|
||||
//! reach. `/usage 42`, `/usage 95 20`, `/usage 42 never`,
|
||||
//! `/usage notloggedin`, `/usage unreachable`, `/usage failed`. The
|
||||
//! vocabulary is `usage::Fixture`'s, which is where the states live.
|
||||
//! - `/compact` -- a compaction, start to finish. Typed rather than
|
||||
//! pressed, because the real dialects take it as a typed command too and
|
||||
//! the phone no longer has a button for it.
|
||||
@@ -106,6 +113,11 @@ pub struct EchoDriver {
|
||||
/// way AskUserQuestion does, and the turn resumes when the last of
|
||||
/// them is answered rather than the first.
|
||||
pending_questions: Mutex<Vec<PendingQuestion>>,
|
||||
/// The invented rate-limit answer `/usage` sets, shared with the
|
||||
/// usage monitor that serves it. An echo session meters nothing, so
|
||||
/// this is unset until a test asks for something -- see
|
||||
/// [`crate::usage::Fixture`].
|
||||
usage: crate::usage::Fixture,
|
||||
/// A pretend context, so the status row has something that behaves the
|
||||
/// way a real one does: it grows with each turn, drops to what the
|
||||
/// compaction says it recovered, and a clear leaves it unmeasured. The
|
||||
@@ -345,6 +357,29 @@ impl EchoDriver {
|
||||
return;
|
||||
}
|
||||
|
||||
// Answered here rather than in the turn below, because it is not
|
||||
// a turn: nothing is generated, and what is being exercised is
|
||||
// the *other* screens -- the bar under the header, the button
|
||||
// beside it and the dialog it opens, all of which read the usage
|
||||
// route rather than this transcript.
|
||||
if let Some(rest) = text.strip_prefix("/usage") {
|
||||
if announce {
|
||||
self.emit(Event::MessageTaken {
|
||||
id: None,
|
||||
text: text.clone(),
|
||||
attachments,
|
||||
});
|
||||
}
|
||||
let said = self.usage.command(rest);
|
||||
self.emit(Event::AssistantText {
|
||||
delta: format!("{said}\n"),
|
||||
});
|
||||
self.emit(Event::Status {
|
||||
state: SessionStatus::Idle,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// The same word the real CLI takes, so a phone drives both the same
|
||||
// way. `Driver::compact` is what the manager's own route calls;
|
||||
// this is the typed path onto it.
|
||||
@@ -651,7 +686,7 @@ impl EchoDriver {
|
||||
});
|
||||
}
|
||||
|
||||
pub fn new(sink: EventSink, session_dir: PathBuf) -> Self {
|
||||
pub fn new(sink: EventSink, session_dir: PathBuf, usage: crate::usage::Fixture) -> Self {
|
||||
let driver = Self {
|
||||
sink,
|
||||
pending_questions: Mutex::new(Vec::new()),
|
||||
@@ -659,6 +694,7 @@ impl EchoDriver {
|
||||
busy: Arc::new(AtomicBool::new(false)),
|
||||
queued: Arc::new(Mutex::new(Vec::new())),
|
||||
session_dir,
|
||||
usage,
|
||||
};
|
||||
driver.emit(Event::Status {
|
||||
state: SessionStatus::Idle,
|
||||
|
||||
+147
-30
@@ -7,10 +7,24 @@
|
||||
//!
|
||||
//! **It is spawned but not spoken to over stdio.** The process is started
|
||||
//! through the same [`Transport`] as any other, and then reached over
|
||||
//! HTTP on a loopback port. That is the case the transport's doc comment
|
||||
//! flags: a remote llama-server would need its port forwarded as well as
|
||||
//! its command wrapped, which is not built, so a session on an ssh host
|
||||
//! is refused rather than silently talking to the wrong machine.
|
||||
//! HTTP on a loopback port. That is the second half of what a transport
|
||||
//! is -- "run this" plus "reach this port" -- and it is what lets a
|
||||
//! session run on another machine: [`Transport::reserve_port`] hands back
|
||||
//! a port the server binds *there* and a port that reaches it *here*, and
|
||||
//! the ssh connection carrying the command carries the tunnel between
|
||||
//! them. The far `llama-server` binds loopback only, so a model is never
|
||||
//! served to that machine's network.
|
||||
//!
|
||||
//! **The model file is the far machine's, not this one's.** A session
|
||||
//! serves a GGUF from the machine that runs `llama-server`, so a remote
|
||||
//! setup names its own models directory (`SshConfig::models_dir`,
|
||||
//! defaulting to the same place this backend keeps its own downloads).
|
||||
//! What this backend has downloaded is on that machine only when they are
|
||||
//! the same machine -- so the file is looked for *there*, and a session
|
||||
//! that names a model the machine does not have says so instead of
|
||||
//! starting a server that will never load one. Downloading to another
|
||||
//! machine is not built; the model gets there however anything else
|
||||
//! gets there.
|
||||
//!
|
||||
//! **The server is stateless between requests**, so the whole
|
||||
//! conversation goes with every one. It is rebuilt from the session's
|
||||
@@ -85,16 +99,10 @@ impl LlamaDriver {
|
||||
session_dir: &Path,
|
||||
sink: EventSink,
|
||||
) -> Result<Self> {
|
||||
if !matches!(transport, Transport::Here) {
|
||||
bail!(
|
||||
"llama.cpp sessions can only run on this machine for now: the model is served \
|
||||
over HTTP, and forwarding that port to another host isn't built yet."
|
||||
);
|
||||
}
|
||||
let model = meta.model.as_deref().context(
|
||||
"a llama.cpp session needs a model -- one of the downloaded ones, by its key",
|
||||
)?;
|
||||
let path = model_path(models_dir, model)?;
|
||||
let path = model_on(transport, models_dir, model)?;
|
||||
|
||||
// Already loaded and still running: keep talking to it. The
|
||||
// health poll below is what confirms it is really answering, so
|
||||
@@ -120,14 +128,21 @@ impl LlamaDriver {
|
||||
));
|
||||
}
|
||||
|
||||
let port = free_port().context("finding a port for llama-server")?;
|
||||
// Where it listens on its own machine, and where that is reached
|
||||
// from here -- the same number when that machine is this one.
|
||||
let forward = transport
|
||||
.reserve_port()
|
||||
.context("finding a port for llama-server")?;
|
||||
let mut args: Vec<String> = vec![
|
||||
"-m".into(),
|
||||
path.to_string_lossy().into_owned(),
|
||||
path.clone(),
|
||||
// Loopback there, whichever machine there is: what reaches it
|
||||
// from outside that machine is the ssh tunnel and nothing
|
||||
// else.
|
||||
"--host".into(),
|
||||
"127.0.0.1".into(),
|
||||
"--port".into(),
|
||||
port.to_string(),
|
||||
forward.there.to_string(),
|
||||
];
|
||||
// Settings that belong to the server because they decide how the
|
||||
// model is loaded; the sampling ones ride on each request instead,
|
||||
@@ -144,7 +159,7 @@ impl LlamaDriver {
|
||||
}
|
||||
|
||||
let program = provider.command.as_deref().unwrap_or("llama-server");
|
||||
let launch = Launch::new(program, args, meta.cwd.as_deref());
|
||||
let launch = Launch::new(program, args, meta.cwd.as_deref()).reaching(forward);
|
||||
// Its output goes to files, not pipes. Not only so the process can
|
||||
// outlive this server: nothing ever read those pipes, so a chatty
|
||||
// llama-server filled the 64 KB buffer and blocked mid-load with
|
||||
@@ -161,8 +176,12 @@ impl LlamaDriver {
|
||||
.id()
|
||||
.context("llama-server exited before it could be recorded")?;
|
||||
tracing::info!(
|
||||
"session {} running {program} for {model} on 127.0.0.1:{port} as pid {pid}",
|
||||
meta.id
|
||||
"session {} running {program} for {model} {} on 127.0.0.1:{} there, \
|
||||
reached at 127.0.0.1:{} here, as pid {pid}",
|
||||
meta.id,
|
||||
transport.describe(),
|
||||
forward.there,
|
||||
forward.here,
|
||||
);
|
||||
// Reaped so it does not become a zombie while this server is still
|
||||
// its parent; the health poll and the record are what actually say
|
||||
@@ -173,12 +192,18 @@ impl LlamaDriver {
|
||||
let _ = child.wait().await;
|
||||
});
|
||||
|
||||
let record = process::Record::of(pid, process::Detail::Http { port })
|
||||
// The *near* port, because that is the one anything reaching this
|
||||
// server has to dial -- including a later run of this backend,
|
||||
// which adopts the record without knowing which machine the server
|
||||
// is on. For a remote session the recorded pid is the ssh
|
||||
// client's, which is the process this machine owns and which holds
|
||||
// the tunnel open for exactly as long as the far server lives.
|
||||
let record = process::Record::of(pid, process::Detail::Http { port: forward.here })
|
||||
.context("llama-server was gone before its start time could be read")?;
|
||||
process::write(session_dir, &record);
|
||||
|
||||
Ok(Self::attached(
|
||||
format!("http://127.0.0.1:{port}"),
|
||||
format!("http://127.0.0.1:{}", forward.here),
|
||||
meta,
|
||||
model,
|
||||
transcript,
|
||||
@@ -212,7 +237,7 @@ impl LlamaDriver {
|
||||
let endpoint = endpoint.clone();
|
||||
let model = model.to_string();
|
||||
let session_dir = session_dir.to_path_buf();
|
||||
std::thread::spawn(move || match wait_until_ready(&endpoint) {
|
||||
std::thread::spawn(move || match wait_until_ready(&endpoint, &session_dir) {
|
||||
Ok(()) => {
|
||||
tracing::info!("{model} loaded and answering at {endpoint}");
|
||||
let _ = sink.send(Event::Status {
|
||||
@@ -518,19 +543,76 @@ fn model_path(models_dir: &Path, key: &str) -> Result<PathBuf> {
|
||||
Ok(path)
|
||||
}
|
||||
|
||||
/// An unused loopback port, by asking the OS for one and letting it go.
|
||||
/// The model file's path **on the machine that will serve it**, confirmed
|
||||
/// to be there.
|
||||
///
|
||||
/// Racy in principle: something else could take it between here and
|
||||
/// llama-server binding. In practice nothing on this machine is hunting
|
||||
/// for ports, and the alternative -- parsing the port back out of the
|
||||
/// server's log -- couples us to its output format for no real gain.
|
||||
fn free_port() -> Result<u16> {
|
||||
let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
|
||||
Ok(listener.local_addr()?.port())
|
||||
/// Local and remote answer the same question and it has to be asked of
|
||||
/// two different filesystems, which is why this is one function rather
|
||||
/// than a check beside the local path and hope for the other case. The
|
||||
/// remote answer is measured for the same reason the local one is: a
|
||||
/// missing file otherwise becomes a `llama-server` that starts, fails to
|
||||
/// load, and reports as a session that never became ready -- which reads
|
||||
/// as the machine being slow.
|
||||
///
|
||||
/// One blocking round trip on a remote spawn, which is the same cost the
|
||||
/// spawn is already paying to start ssh. The alternative is a path built
|
||||
/// here from a `~` this machine cannot expand.
|
||||
fn model_on(transport: &Transport, models_dir: &Path, key: &str) -> Result<String> {
|
||||
let Transport::Ssh { name, .. } = transport else {
|
||||
return Ok(model_path(models_dir, key)?.to_string_lossy().into_owned());
|
||||
};
|
||||
// The same directory the spawn screen listed for this machine, and
|
||||
// for the same reason it is one function: a list from one place and a
|
||||
// load from another is a model that appears and then fails.
|
||||
let dir = crate::models::dir_on(transport, models_dir);
|
||||
// Checked here rather than in the script: `..` in a key would walk
|
||||
// out of the models directory on a machine this server can start
|
||||
// processes on, and the phone is where the key comes from.
|
||||
for part in key.split('/') {
|
||||
if part.is_empty() || part == "." || part == ".." {
|
||||
bail!("\"{key}\" is not a model key this can resolve");
|
||||
}
|
||||
}
|
||||
let path = format!("{}/{key}", dir.trim_end_matches('/'));
|
||||
// `$HOME` on the far side, which is the only machine that knows what
|
||||
// it is -- and the resolved path is printed back so the launch below
|
||||
// hands `llama-server` something absolute.
|
||||
//
|
||||
// "the file is not there" is answered rather than failed, because the
|
||||
// two are different things to a reader and only one of them is a
|
||||
// fault: a machine that could not be asked at all has to say so in
|
||||
// its own words, and it would otherwise arrive as this same sentence
|
||||
// about a missing model.
|
||||
let script = "p=$1; case $p in \"~\") p=$HOME;; \"~/\"*) p=$HOME/${p#\"~/\"};; esac; \
|
||||
[ -f \"$p\" ] && printf 'at\\t%s\\n' \"$p\" || printf 'missing\\n'"
|
||||
.to_string();
|
||||
let launch = Launch::new(
|
||||
"sh",
|
||||
vec!["-c".to_string(), script, "sh".to_string(), path.clone()],
|
||||
None,
|
||||
);
|
||||
let answer = transport
|
||||
.capture_blocking(&launch)
|
||||
.with_context(|| format!("couldn't ask {name} where its models are"))?;
|
||||
match answer.trim().split_once('\t') {
|
||||
Some(("at", resolved)) => Ok(resolved.to_string()),
|
||||
_ => bail!(
|
||||
"{name} has no model at {path}. A llama.cpp session serves the file from the \
|
||||
machine it runs on, so the model has to be on {name} -- what this backend has \
|
||||
downloaded is somewhere else."
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
/// Polls until the server says it is ready, or gives up.
|
||||
fn wait_until_ready(endpoint: &str) -> Result<()> {
|
||||
///
|
||||
/// Watches the process as well as the port, because the two failures need
|
||||
/// different words and one of them is common: a model that will not load,
|
||||
/// a port already taken on the far machine, a `llama-server` too old for
|
||||
/// a flag. All of those exit within a second and none of them will ever
|
||||
/// answer `/health`, so waiting out the timeout turns a server that said
|
||||
/// exactly what was wrong into "gave up after 300s".
|
||||
fn wait_until_ready(endpoint: &str, session_dir: &Path) -> Result<()> {
|
||||
let deadline = std::time::Instant::now() + READY_TIMEOUT;
|
||||
let url = format!("{endpoint}/health");
|
||||
loop {
|
||||
@@ -539,13 +621,48 @@ fn wait_until_ready(endpoint: &str) -> Result<()> {
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
// `None` is the session having been stopped or deleted while this
|
||||
// waited, which is nobody's fault and still not worth waiting on.
|
||||
match process::recorded(session_dir) {
|
||||
Some((_, process::Liveness::Alive | process::Liveness::Unknown)) => {}
|
||||
Some((_, process::Liveness::Dead)) | None => {
|
||||
bail!("it exited before it answered.{}", log_tail(session_dir));
|
||||
}
|
||||
}
|
||||
if std::time::Instant::now() > deadline {
|
||||
bail!("gave up after {}s", READY_TIMEOUT.as_secs());
|
||||
bail!(
|
||||
"gave up after {}s.{}",
|
||||
READY_TIMEOUT.as_secs(),
|
||||
log_tail(session_dir)
|
||||
);
|
||||
}
|
||||
std::thread::sleep(std::time::Duration::from_millis(250));
|
||||
}
|
||||
}
|
||||
|
||||
/// The end of `llama-server`'s own log, for a failure message.
|
||||
///
|
||||
/// Its account of what went wrong is the useful half -- "failed to load
|
||||
/// model", "bind: Address already in use" -- and on a remote session it
|
||||
/// is the only half, since nobody reading the phone can open a file on
|
||||
/// that machine. Bounded, because this ends up in an event a phone draws.
|
||||
fn log_tail(session_dir: &Path) -> String {
|
||||
let Ok(text) = std::fs::read_to_string(session_dir.join(SERVER_LOG)) else {
|
||||
return String::new();
|
||||
};
|
||||
let tail: Vec<&str> = text.lines().rev().take(LOG_TAIL_LINES).collect();
|
||||
if tail.is_empty() {
|
||||
return String::new();
|
||||
}
|
||||
format!(
|
||||
" It last said: {}",
|
||||
tail.into_iter().rev().collect::<Vec<_>>().join(" / ")
|
||||
)
|
||||
}
|
||||
|
||||
/// How much of that log to carry into a message somebody reads on a phone.
|
||||
const LOG_TAIL_LINES: usize = 6;
|
||||
|
||||
/// One streamed completion: posts the conversation, emits each delta as it
|
||||
/// arrives. Emits rather than returns: the transcript those events land
|
||||
/// in is what the next turn reads back, so there is nothing to hand up.
|
||||
|
||||
+86
-24
@@ -163,6 +163,17 @@ pub struct SessionInfo {
|
||||
/// answers and only one of them stays true.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub max_image_edge: Option<u32>,
|
||||
/// Which of `GET /usage`'s snapshots reports on this session, and
|
||||
/// absent where nothing meters it -- see
|
||||
/// [`DriverKind::usage_provider`].
|
||||
///
|
||||
/// Reported for the same reason `keeps_own_transcript` is: it is a
|
||||
/// fact about the provider's *kind*, and the phone has only its name.
|
||||
/// Pairing by machine alone was the bug it exists to fix -- one
|
||||
/// machine runs echo and the Claude CLI, so every echo session drew
|
||||
/// the CLI's five-hour window as if it were its own.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub usage_provider: Option<&'static str>,
|
||||
/// Whether this session announces itself -- reported for the same
|
||||
/// reason `permission_mode` is: a switch that guesses its own position
|
||||
/// is how you turn something off while believing you are reading it.
|
||||
@@ -549,6 +560,7 @@ impl LiveSession {
|
||||
context_tokens: *self.shared.context_tokens.lock().unwrap(),
|
||||
notify: *self.shared.notify.lock().unwrap(),
|
||||
max_image_edge: kind.and_then(DriverKind::max_image_edge),
|
||||
usage_provider: kind.and_then(DriverKind::usage_provider),
|
||||
imported,
|
||||
keeps_own_transcript: kind.is_some_and(DriverKind::keeps_own_transcript),
|
||||
cwd: cwd.map(Path::to_path_buf),
|
||||
@@ -585,6 +597,11 @@ pub struct SessionManager {
|
||||
/// [`SessionManager::marking_new_sessions_throwaway`] and
|
||||
/// [`SessionConfig::throwaway`].
|
||||
spawn_throwaway: bool,
|
||||
/// The invented rate-limit answer an echo session's `/usage` sets,
|
||||
/// shared with the usage monitor that serves it. Held here because
|
||||
/// every echo driver this manager builds is handed a clone -- see
|
||||
/// [`SessionManager::reporting_usage_fixture`].
|
||||
usage_fixture: crate::usage::Fixture,
|
||||
inner: RwLock<Inner>,
|
||||
}
|
||||
|
||||
@@ -603,6 +620,11 @@ impl SessionManager {
|
||||
wg_app_link::private::create_dir(&data_dir)?;
|
||||
|
||||
let (notifications, _) = broadcast::channel(NOTIFICATION_BUFFER);
|
||||
// Made here rather than passed in, and handed *out* to the usage
|
||||
// monitor by whoever wires the two together: every echo driver
|
||||
// this manager builds gets a clone, including the ones built
|
||||
// below, so it has to exist before the first session does.
|
||||
let usage_fixture = crate::usage::Fixture::new();
|
||||
let mut live = HashMap::new();
|
||||
for meta in &config.sessions {
|
||||
// One unlaunchable session -- a corrupt transcript, an
|
||||
@@ -614,8 +636,11 @@ impl SessionManager {
|
||||
meta.clone(),
|
||||
&setup,
|
||||
&provider,
|
||||
&data_dir,
|
||||
&models_dir,
|
||||
Env {
|
||||
data_dir: &data_dir,
|
||||
models_dir: &models_dir,
|
||||
usage: &usage_fixture,
|
||||
},
|
||||
notifications.clone(),
|
||||
// Nothing is started here. See `Launching`: a restart
|
||||
// picks up the processes that are still running and
|
||||
@@ -638,11 +663,41 @@ impl SessionManager {
|
||||
notifications,
|
||||
pending: Arc::new(pending::Registry::default()),
|
||||
spawn_throwaway: false,
|
||||
usage_fixture,
|
||||
inner: RwLock::new(Inner { config, live }),
|
||||
};
|
||||
Ok(manager)
|
||||
}
|
||||
|
||||
/// Where this backend's own model downloads live. The machine a
|
||||
/// session runs on may keep its elsewhere -- see `models::dir_on`.
|
||||
pub fn models_dir(&self) -> &Path {
|
||||
&self.models_dir
|
||||
}
|
||||
|
||||
/// What this manager lends a session it launches. Borrowed from the
|
||||
/// manager rather than cloned, so there is one answer to where things
|
||||
/// are kept.
|
||||
fn env(&self) -> Env<'_> {
|
||||
Env {
|
||||
data_dir: &self.data_dir,
|
||||
models_dir: &self.models_dir,
|
||||
usage: &self.usage_fixture,
|
||||
}
|
||||
}
|
||||
|
||||
/// The invented rate-limit answer this manager's echo sessions set
|
||||
/// with `/usage`, for the usage monitor to serve.
|
||||
///
|
||||
/// Handed out rather than taken in because the drivers built inside
|
||||
/// the constructor need it, and because the direction is the one the
|
||||
/// layering allows: `usage` sits below the session layer, so a
|
||||
/// session can hold one of its types while it holds nothing of a
|
||||
/// session's.
|
||||
pub fn usage_fixture(&self) -> crate::usage::Fixture {
|
||||
self.usage_fixture.clone()
|
||||
}
|
||||
|
||||
/// Marks every session spawned from here on as one whose process is
|
||||
/// stopped when this server exits -- see [`SessionConfig::throwaway`]
|
||||
/// and [`SessionManager::stop_throwaway_sessions`].
|
||||
@@ -1035,6 +1090,8 @@ impl SessionManager {
|
||||
context_tokens: None,
|
||||
max_image_edge: kind_of(&inner.config, &meta.setup, &meta.provider)
|
||||
.and_then(DriverKind::max_image_edge),
|
||||
usage_provider: kind_of(&inner.config, &meta.setup, &meta.provider)
|
||||
.and_then(DriverKind::usage_provider),
|
||||
notify: meta.notify,
|
||||
imported: import::read_cursor(&self.data_dir.join(&meta.id)).is_some(),
|
||||
keeps_own_transcript: keeps_own_transcript(
|
||||
@@ -1162,8 +1219,7 @@ impl SessionManager {
|
||||
meta.clone(),
|
||||
&setup,
|
||||
&provider,
|
||||
&self.data_dir,
|
||||
&self.models_dir,
|
||||
self.env(),
|
||||
self.notifications.clone(),
|
||||
Launching::Asked(seed),
|
||||
)?;
|
||||
@@ -1605,7 +1661,7 @@ impl SessionManager {
|
||||
&meta,
|
||||
&setup,
|
||||
&provider,
|
||||
&self.models_dir,
|
||||
self.env(),
|
||||
session.dir(),
|
||||
session.transcript_path(),
|
||||
&session.sink,
|
||||
@@ -1619,8 +1675,7 @@ impl SessionManager {
|
||||
meta,
|
||||
&setup,
|
||||
&provider,
|
||||
&self.data_dir,
|
||||
&self.models_dir,
|
||||
self.env(),
|
||||
self.notifications.clone(),
|
||||
Launching::Asked(None),
|
||||
)?;
|
||||
@@ -2020,6 +2075,19 @@ enum Launching {
|
||||
Restart,
|
||||
}
|
||||
|
||||
/// What the server around a session lends it: where sessions and models
|
||||
/// are kept, and the usage fixture an echo session's `/usage` sets.
|
||||
///
|
||||
/// One parameter rather than three because they travel together through
|
||||
/// every launch path and none of them is a fact about the session --
|
||||
/// they are this server's belongings, handed down.
|
||||
#[derive(Clone, Copy)]
|
||||
struct Env<'a> {
|
||||
data_dir: &'a Path,
|
||||
models_dir: &'a Path,
|
||||
usage: &'a crate::usage::Fixture,
|
||||
}
|
||||
|
||||
/// Creates the session directory, opens its transcript (continuing the
|
||||
/// sequence numbering if one exists), settles what the session is doing,
|
||||
/// and spawns the event pump -- with a driver behind it where there is a
|
||||
@@ -2028,12 +2096,11 @@ fn launch(
|
||||
meta: SessionConfig,
|
||||
setup: &SetupConfig,
|
||||
provider: &ProviderConfig,
|
||||
data_dir: &Path,
|
||||
models_dir: &Path,
|
||||
env: Env<'_>,
|
||||
notifications: broadcast::Sender<Notification>,
|
||||
why: Launching,
|
||||
) -> Result<Arc<LiveSession>> {
|
||||
let dir = data_dir.join(&meta.id);
|
||||
let dir = env.data_dir.join(&meta.id);
|
||||
wg_app_link::private::create_dir(&dir)?;
|
||||
let transcript_path = dir.join("transcript.jsonl");
|
||||
let mut transcript = Transcript::open(&transcript_path)?;
|
||||
@@ -2163,17 +2230,7 @@ fn launch(
|
||||
|
||||
let driver = Arc::new(Mutex::new(
|
||||
driving
|
||||
.then(|| {
|
||||
make_driver(
|
||||
&meta,
|
||||
setup,
|
||||
provider,
|
||||
models_dir,
|
||||
&dir,
|
||||
&transcript_path,
|
||||
&sink,
|
||||
)
|
||||
})
|
||||
.then(|| make_driver(&meta, setup, provider, env, &dir, &transcript_path, &sink))
|
||||
.transpose()?,
|
||||
));
|
||||
|
||||
@@ -2216,18 +2273,22 @@ fn make_driver(
|
||||
meta: &SessionConfig,
|
||||
setup: &SetupConfig,
|
||||
provider: &ProviderConfig,
|
||||
models_dir: &Path,
|
||||
env: Env<'_>,
|
||||
dir: &Path,
|
||||
transcript_path: &Path,
|
||||
sink: &EventSink,
|
||||
) -> Result<Arc<dyn Driver>> {
|
||||
Ok(match provider.kind {
|
||||
DriverKind::Echo => Arc::new(EchoDriver::new(sink.clone(), dir.to_path_buf())),
|
||||
DriverKind::Echo => Arc::new(EchoDriver::new(
|
||||
sink.clone(),
|
||||
dir.to_path_buf(),
|
||||
env.usage.clone(),
|
||||
)),
|
||||
DriverKind::LlamaCpp => Arc::new(LlamaDriver::launch(
|
||||
meta,
|
||||
provider,
|
||||
&Transport::for_setup(setup),
|
||||
models_dir,
|
||||
env.models_dir,
|
||||
transcript_path,
|
||||
dir,
|
||||
sink.clone(),
|
||||
@@ -2584,6 +2645,7 @@ mod tests {
|
||||
driver: Arc::new(Mutex::new(Some(Arc::new(EchoDriver::new(
|
||||
sink.clone(),
|
||||
dir.path().to_path_buf(),
|
||||
crate::usage::Fixture::new(),
|
||||
))))),
|
||||
sink,
|
||||
waiting: Mutex::new(VecDeque::new()),
|
||||
|
||||
@@ -13,11 +13,14 @@
|
||||
//! this module decides *which* transport, that one knows what a correct
|
||||
//! ssh invocation is.
|
||||
//!
|
||||
//! Known second operation, not built because nothing needs it yet: a
|
||||
//! managed `llama-server` is spawned as a process but then spoken to over
|
||||
//! HTTP, so a remote one needs a forwarded port (`ssh -L`) as well. A
|
||||
//! transport is eventually "run this" plus "reach this port", where the
|
||||
//! second is a no-op locally. See PLAN.md's SSH section.
|
||||
//! A transport is therefore two operations rather than one: **run this**,
|
||||
//! and **reach this port**. The second is what a managed `llama-server`
|
||||
//! needs -- it is spawned as a process and then spoken to over HTTP -- and
|
||||
//! it is a no-op locally, where the port a program binds is already a port
|
||||
//! this machine can dial. Over ssh it is an `-L` tunnel carried by the
|
||||
//! same connection that runs the command, so the model server binds
|
||||
//! loopback on the far machine and is never exposed to its network. See
|
||||
//! [`Transport::reserve_port`] and PLAN.md's SSH section.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::Stdio;
|
||||
@@ -26,16 +29,28 @@ use anyhow::{Context, Result};
|
||||
use tokio::process::Child;
|
||||
|
||||
use crate::config::SshConfig;
|
||||
pub use crate::ssh::Forward;
|
||||
|
||||
/// What a driver needs run in order to exist as a process.
|
||||
///
|
||||
/// Deliberately just the three things every transport can carry. Anything
|
||||
/// a particular machine needs -- a port, a key, extra ssh options -- is
|
||||
/// Deliberately just what every transport can carry: the command, where
|
||||
/// it runs, and a port the caller needs to reach. Anything a particular
|
||||
/// machine needs -- a key, extra ssh options, which address to dial -- is
|
||||
/// the transport's own configuration, not something a driver states.
|
||||
pub struct Launch {
|
||||
pub program: String,
|
||||
pub args: Vec<String>,
|
||||
pub cwd: Option<PathBuf>,
|
||||
/// A port this program will listen on, and the port that reaches it
|
||||
/// from here -- see [`Transport::reserve_port`], which is the only
|
||||
/// thing that should produce one.
|
||||
///
|
||||
/// On the launch rather than in [`Transport::spawn`]'s signature
|
||||
/// because it is part of what is being run: a caller that needs to
|
||||
/// reach the process it is starting says so once, where it says
|
||||
/// everything else about it, and every transport reads it the same
|
||||
/// way.
|
||||
pub forward: Option<Forward>,
|
||||
}
|
||||
|
||||
impl Launch {
|
||||
@@ -44,8 +59,16 @@ impl Launch {
|
||||
program: program.into(),
|
||||
args,
|
||||
cwd: cwd.map(Path::to_path_buf),
|
||||
forward: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Says that this program serves `forward.there`, and that the caller
|
||||
/// will reach it at `forward.here`.
|
||||
pub fn reaching(mut self, forward: Forward) -> Self {
|
||||
self.forward = Some(forward);
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
/// How a launched process's standard streams are connected.
|
||||
@@ -113,6 +136,7 @@ impl Transport {
|
||||
&launch.program,
|
||||
&launch.args,
|
||||
launch.cwd.as_deref(),
|
||||
launch.forward,
|
||||
));
|
||||
match streams {
|
||||
Streams::Piped => {
|
||||
@@ -169,12 +193,15 @@ impl Transport {
|
||||
Self::Here => None,
|
||||
Self::Ssh { ssh, .. } => Some(ssh),
|
||||
};
|
||||
let output =
|
||||
crate::ssh::command(host, &launch.program, &launch.args, launch.cwd.as_deref())
|
||||
.output()
|
||||
.with_context(|| {
|
||||
format!("couldn't run \"{}\" {}", launch.program, self.describe())
|
||||
})?;
|
||||
let output = crate::ssh::command(
|
||||
host,
|
||||
&launch.program,
|
||||
&launch.args,
|
||||
launch.cwd.as_deref(),
|
||||
launch.forward,
|
||||
)
|
||||
.output()
|
||||
.with_context(|| format!("couldn't run \"{}\" {}", launch.program, self.describe()))?;
|
||||
if !output.status.success() {
|
||||
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
|
||||
anyhow::bail!(if stderr.is_empty() {
|
||||
@@ -237,6 +264,34 @@ impl Transport {
|
||||
})
|
||||
}
|
||||
|
||||
/// Picks a port for a launched program to serve on, and the port that
|
||||
/// reaches it from here.
|
||||
///
|
||||
/// The "reach this port" half of what a transport is. Locally there is
|
||||
/// one port and the OS chooses it, by binding and letting go -- racy
|
||||
/// in principle, and nothing on this machine is hunting for ports.
|
||||
///
|
||||
/// Over ssh the near end is chosen the same way and the far end is a
|
||||
/// guess, because there is no portable way to ask a machine for a free
|
||||
/// port that does not race with binding it anyway. It is taken from
|
||||
/// [`FAR_PORTS`], below the range Linux hands out to outgoing
|
||||
/// connections, so a collision means something else deliberately
|
||||
/// listening there. That is not silent: the program fails to bind and
|
||||
/// exits, and `session::llama` reports what its log said rather than
|
||||
/// waiting out its readiness timeout.
|
||||
pub fn reserve_port(&self) -> Result<Forward> {
|
||||
let listener = std::net::TcpListener::bind("127.0.0.1:0")
|
||||
.context("asking this machine for a free port")?;
|
||||
let here = listener.local_addr()?.port();
|
||||
Ok(match self {
|
||||
Self::Here => Forward { there: here, here },
|
||||
Self::Ssh { .. } => Forward {
|
||||
there: rand::random_range(FAR_PORTS),
|
||||
here,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
/// How to say where this runs, for a log line a person reads.
|
||||
pub fn describe(&self) -> String {
|
||||
match self {
|
||||
@@ -246,6 +301,11 @@ impl Transport {
|
||||
}
|
||||
}
|
||||
|
||||
/// Where a port on another machine is guessed from: high enough to be out
|
||||
/// of the way of services, and below the 32768-60999 Linux hands out to
|
||||
/// outgoing connections, which is where a guess would most often collide.
|
||||
const FAR_PORTS: std::ops::Range<u16> = 20000..30000;
|
||||
|
||||
/// What a command is given on its standard input.
|
||||
///
|
||||
/// Three cases rather than an `Option<Stdio>` because they are three
|
||||
|
||||
Reference in new issue
Block a user