diff --git a/CLAUDE.md b/CLAUDE.md index 0995f757b..67d937e41 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -504,7 +504,7 @@ When `session.enabled` is true (default) and `listener.session_api_addr` is non- | Method & Path | Format | Purpose | |---|---|---| | `GET /` | text | One-line-per-endpoint index. Answers "is this the session API, and on the right port?" — the reason a 404 here was worth replacing. | -| `GET /v1/sessions` | `application/json` | List active sessions: `{sessions: [{id, createdAt, updatedAt, eventCount, title, agent, adopted, totalTokens, costMicros, avoidedMicros, saturated, active, promptContext, inferenceHost, inferenceHostFromHistory}]}`. `id`, `createdAt`, `updatedAt`, `eventCount` and `active` are always present; every other field is `omitempty` — absent rather than zero, on the standing rule that an unknown value must not render as a real one. (Do not read that off the position of `active`: it sits fourth from the end, between two `omitempty` fields.) **`inferenceHost`** is the host of the session's most recent request that carried an inference parse — after a redirect, the server it was sent to rather than the host the client named — folded at append time, latest wins, and left alone by tunnel rows, MCP calls, responses and denials; agentop's sessions-table `SERVER` column names a session's inference server from it, and both `S`'s result line and `agentop server remove`'s warning count running sessions by it. **`inferenceHostFromHistory`** is `true` when that host came only from the archive's fold, because no request the proxy's entry for the session holds has set it — a session resumed after a restart that has sent no inference since. The inference-router's pins live in memory, so after a restart nothing holds such a session where its history says: both of agentop's counts leave it out. **`title` is a suggestion, not an identifier:** the proxy derives it from the session's own events (a `/rename`, else a ``, else ordinary user prose, with `` blocks excised), so it is a display convenience and nothing addresses a session by it. Absent when nothing in the events named it. **Folded at append time and FIRST-WINS, except that a `/rename` always overrides** — so ordinary conversation does not re-title a session on every turn, and a `/rename` survives eviction of the event that carried it. agentop reads this field as a FALLBACK: its TITLE column prefers a harvested Claude Code transcript title and uses the served title only for a session the harvest cannot name. That precedence is fixed rather than a judgement about which string is better — both sides rank candidates their own way and do not agree on every session. That is the case worth having: an agent with no transcript tree on the operator's disk still routes through the proxy, so a row that used to render blank now has a name. agentop deliberately still treats such a row as unnamed for its own re-harvest backoff, so a served title does not stop it looking for a harvested one. **`?archived=true` adds history** where the proxy runs a session archive (`session.archive`, laptop only — see the chatty-traffic gotcha): every session the archive holds and memory does not, marked `resident: false`, plus a top-level `archive` object — `{bytes, maxBytes, retentionDays}` and, only when nonzero, `droppedEvents`, `writeErrors`, `droppedRenames`, `paused`, any of which means the history has gaps. A session in both is listed once, as its resident row. **That row covers the session's whole history**: `eventCount`, `totalTokens`, `costMicros`, `avoidedMicros`, `currencies`, `saturated`, `title`, `agent`, `createdAt`, `promptContext` and `inferenceHost` continue from the archive's fold of what came before this process's entry, each by its own rule, in the default list too — so with an archive and `session.max_events` unset (the default) a restart changes no figure, and without an archive they start over. Without an archive the parameter is ignored and no `archive` object is sent, which is how a client tells "no history here" from "no history yet". | +| `GET /v1/sessions` | `application/json` | List active sessions: `{sessions: [{id, createdAt, updatedAt, eventCount, title, agent, adopted, totalTokens, costMicros, avoidedMicros, saturated, active, promptContext, inferenceHost, inferenceHostFromHistory}]}`. `id`, `createdAt`, `updatedAt`, `eventCount` and `active` are always present; every other field is `omitempty` — absent rather than zero, on the standing rule that an unknown value must not render as a real one. (Do not read that off the position of `active`: it sits fourth from the end, between two `omitempty` fields.) **`inferenceHost`** is the host of the session's most recent request that carried an inference parse — after a redirect, the server it was sent to rather than the host the client named — folded at append time, latest wins, and left alone by tunnel rows, MCP calls, responses and denials; agentop's sessions-table `SERVER` column names a session's inference server from it, and both `S`'s result line and `agentop server remove`'s warning count running sessions by it. **`inferenceHostFromHistory`** is `true` when that host came only from the archive's fold, because no request the proxy's entry for the session holds has set it — a session resumed after a restart that has sent no inference since. On a local install the inference-router keeps its pins in the plugin store (`~/.cortex/plugin-state.json`), which a restart keeps, so both of agentop's counts include such a session. **`title` is a suggestion, not an identifier:** the proxy derives it from the session's own events (a `/rename`, else a ``, else ordinary user prose, with `` blocks excised), so it is a display convenience and nothing addresses a session by it. Absent when nothing in the events named it. **Folded at append time and FIRST-WINS, except that a `/rename` always overrides** — so ordinary conversation does not re-title a session on every turn, and a `/rename` survives eviction of the event that carried it. agentop reads this field as a FALLBACK: its TITLE column prefers a harvested Claude Code transcript title and uses the served title only for a session the harvest cannot name. That precedence is fixed rather than a judgement about which string is better — both sides rank candidates their own way and do not agree on every session. That is the case worth having: an agent with no transcript tree on the operator's disk still routes through the proxy, so a row that used to render blank now has a name. agentop deliberately still treats such a row as unnamed for its own re-harvest backoff, so a served title does not stop it looking for a harvested one. **`?archived=true` adds history** where the proxy runs a session archive (`session.archive`, laptop only — see the chatty-traffic gotcha): every session the archive holds and memory does not, marked `resident: false`, plus a top-level `archive` object — `{bytes, maxBytes, retentionDays}` and, only when nonzero, `droppedEvents`, `writeErrors`, `droppedRenames`, `paused`, any of which means the history has gaps. A session in both is listed once, as its resident row. **That row covers the session's whole history**: `eventCount`, `totalTokens`, `costMicros`, `avoidedMicros`, `currencies`, `saturated`, `title`, `agent`, `createdAt`, `promptContext` and `inferenceHost` continue from the archive's fold of what came before this process's entry, each by its own rule, in the default list too — so with an archive and `session.max_events` unset (the default) a restart changes no figure, and without an archive they start over. Without an archive the parameter is ignored and no `archive` object is sent, which is how a client tells "no history here" from "no history yet". | | `GET /v1/sessions/{id}` | `application/json` | The session's most recent events. `?limit=N` (default 500, max 2000) sets the window; `?before=` returns the page ending just before that event, so the whole session is reachable by paging backward from the tail. `totalEvents` is the session's true length and `oldestSeq` the oldest event the store still holds — both present only when this response is not the whole session, so a client can tell "this is the beginning" from "there is more behind me" without a second request. 404 if unknown/expired. **One response is still not a full snapshot:** with `session.max_events` unset a session can hold thousands of events, and one real session's whole history encoded to 1.1GB — 17s to write, against clients that time out in 10. That cap is why `before` exists — until it did, a session past 2000 events had a beginning no request could reach at any limit, while still costing memory. The response is written one event at a time rather than encoded whole, so serving it costs the proxy heap proportional to one event; see the chatty-traffic gotcha below. **With a session archive, a page continues from disk** below the oldest event memory holds: a session resumed after a restart pages back through its whole history, one evicted from memory is served instead of 404'd, and `totalEvents`/`oldestSeq` describe the merged history. `GET /v1/sessions/{id}/events/{seq}` falls back to the archive the same way, and answers 404 rather than a neighbour when neither holds that seq. | | `DELETE /v1/sessions` | `application/json` | Clear every session: the store, and the session archive where one runs. **The cost ledger is kept** — spend totals, no content. Answers `{sessions, archivedSessions, bytes}`; a failed disk half is a 500 adding `archiveError`, with memory cleared regardless (the store clears under its lock before the archive is waited on, up to 10s). **The one state-changing route on this unauthenticated API, so it is guarded against a browser**, each guard pinned by its own test in `core/sessionapi/clear_test.go`: DELETE only (a cross-site DELETE needs a preflight this server never approves; a simple cross-site POST needs none); a loopback `Host` — exactly `localhost`, `127.0.0.1` or `::1`, which defeats DNS rebinding; no `Origin` header; and `WithClearAllowed`, which `cmd/cortex` sets from `listener.bind_loopback_only`. A refusal is a 403 `{"error": ...}`. agentop's `X` drives it. | | `GET /v1/events` | `text/event-stream` | SSE stream of new events. Optional `?session=` filters to one session. Heartbeat every 30s. | diff --git a/cmd/agentop/README.md b/cmd/agentop/README.md index 1a9f72249..013033450 100644 --- a/cmd/agentop/README.md +++ b/cmd/agentop/README.md @@ -658,8 +658,7 @@ With several LiteLLM servers, each with its own URL and key, `agentop server` chooses which one an agent's **new** sessions use, through the [`inference-router`](../../docs/plugin-catalog.md#inference-router) plugin. A session stays on the server it started on, including one already running when its -agent is first routed, until the proxy restarts: a restarted proxy treats every -session it sees as a new one. Routing is opt-in per agent: until an agent is given +agent is first routed. Routing is opt-in per agent: until an agent is given a server, its traffic goes where the agent sends it. ```sh @@ -702,22 +701,17 @@ With Claude Code's `ANTHROPIC_BASE_URL` at `ete`, the listing after `use` reads: before replacing it says which mapping it will have. - **`use --agent `** routes the agent's new sessions to the server, including an agent that has not run yet. **`reset --agent `** stops - routing it. Either way a session already running stays where it is if it has sent - a request to one of the servers since the proxy started, which is how the router - knows where it is — unless the proxy has since dropped it from memory, which keeps - the most recently used sessions (100 by default), before the router pinned it. + routing it. Either way a session already running stays where it is: its pin holds + it, kept in `~/.cortex/plugin-state.json` so a proxy restart forgets none. - **`remove `** refuses while an agent is routed to the server, and names the command that takes the agent off it; it also refuses the last server. A session that started on a removed server gets a 503 asking for a new session - until the server is added back, or until the proxy restarts, which forgets - which server each session started on. That holds while its requests are + until the server is added back. That holds while its requests are addressed to a server that is left: one addressed to the removed server's own host is not routed at all, and goes there with the agent's own key. So when sessions the proxy is running last sent their inference to the server, `remove` warns how many, in those terms, before it goes ahead. It leaves - out the default and `pending:` buckets, which the router never pins, and a session - resumed after a restart that has sent no inference since, which no pin survived - to hold. + out the default and `pending:` buckets, which the router never pins. - **`agentop server`** lists each server's host (its whole URL when it is plain `http`), its model mapping and the agents routed to it, then checks `~/.claude/settings.json`: that `ANTHROPIC_BASE_URL` names one of the servers, @@ -751,8 +745,7 @@ the same way, says the proxy stopped answering, and exits 1. A file someone edit while the proxy was reloading is not put back but left as found, with an error saying so. If the proxy reports neither a reload nor a refusal within 30 seconds, the command exits 1 and leaves the file as written. With no proxy running the file is still written, and the change applies at -the next start, to every session from then: a proxy that starts knows where no -session is. Nothing is written unless the result loads as a Cortex config and +the next start. Nothing is written unless the result loads as a Cortex config and passes the router's own rules; what only the running proxy can check, such as whether its build includes the router, is what a refused reload reports. A change that changes nothing writes nothing. `--config PATH` points every form at another @@ -1495,12 +1488,9 @@ New claude-code sessions → glm. 3 running sessions stay where they are. The count is of the agent's sessions this proxy holds in memory whose inference last went somewhere other than the new choice (for a switch back to its own choice, to any of the servers), and is left out when there are none. A switch moves none of them: -the router keeps a running session on the server it started on until the proxy -restarts, unless the proxy dropped the session from memory before the router pinned -it (see [`agentop server`](#choosing-an-inference-server-agentop-server)). So a -session resumed after a restart that has sent no inference since is not counted: no -pin survived the restart, and its next request goes to the new choice. The sessions -table's `SERVER` column shows where each one is. +the router keeps a running session on the server it started on, across a proxy +restart too (see [`agentop server`](#choosing-an-inference-server-agentop-server)). +The sessions table's `SERVER` column shows where each one is. When the switch does not go through, a panel over the pane says so, whole, and where it left the config file; `Esc`, `q`, `Ctrl+C` or `↵` closes it. A reload the proxy diff --git a/cmd/agentop/cmd_server.go b/cmd/agentop/cmd_server.go index 77c2c9a2a..3d1c9cfb4 100644 --- a/cmd/agentop/cmd_server.go +++ b/cmd/agentop/cmd_server.go @@ -41,8 +41,9 @@ that Claude Code's settings let it be routed. A server is a gateway, such as LiteLLM, with its own URL and key. Routing is per agent and opt-in: until "use" gives an agent a server, its traffic goes wherever the agent sends it. A session stays on the server it started on, so a change -applies to new sessions only. A proxy restart forgets where each session is, and -treats each one it then sees as new. +applies to new sessions only, and a proxy restart moves none, since each +session's server is kept in ~/.cortex. A Claude Code conversation that began +before routing was set up stays where Claude Code sends it. "add" reads the server's API key at a prompt that does not echo, or from stdin with --key-stdin; it is never an argument. Adding a name that exists asks before diff --git a/cmd/agentop/cmd_server_write.go b/cmd/agentop/cmd_server_write.go index 34e3270fa..1e6c711c9 100644 --- a/cmd/agentop/cmd_server_write.go +++ b/cmd/agentop/cmd_server_write.go @@ -298,10 +298,11 @@ func serverRemove(args []string, stdout, stderr io.Writer) int { // agent's current server, which cannot be name, since remove refuses a server an agent is // routed to. None of them can get the error, whatever host it last sent to. // -// IN MEMORY, NOT IN THE ARCHIVE, AND BY A REQUEST THIS PROXY SAW: the router pins a session by -// the first request it routes and keeps the pin in memory, so a session only the archive holds -// has no pin left to break, and neither has a resident one whose host is only its history's. -// SessionSummary.Active would not do — it marks the one most recently updated session. +// IN MEMORY, NOT IN THE ARCHIVE: running means a session this proxy has seen a request from. +// One resumed after a restart counts, its host its history's: the router keeps its pins in the +// plugin store, so its pin survived. One only the archive holds has sent nothing since and is +// not running. SessionSummary.Active would not do — it marks the one most recently updated +// session. // // 0 when the proxy does not answer, which is the quiet side for a warning: the remove itself // is what matters, and it goes ahead either way. @@ -317,9 +318,7 @@ func runningOn(sessionsURL string, c routerconfig.Config, name string) int { } n := 0 for _, s := range list { - // Its host only its history's: resumed after a restart, nothing sent since, so no pin - // survived to hold it to name (see SessionSummary.InferenceHostFromHistory). - if s.InferenceHostFromHistory || s.ID == session.DefaultSessionID || strings.HasPrefix(s.ID, session.PendingPrefix) { + if s.ID == session.DefaultSessionID || strings.HasPrefix(s.ID, session.PendingPrefix) { continue } if on, ok := servers.ForHost(c, s.InferenceHost); ok && on == name { @@ -423,21 +422,15 @@ func serverReset(args []string, stdout, stderr io.Writer) int { // runningSessionsStay is what use and reset mean for sessions already running, // said once the running proxy has the change. It holds a session by its pin or by -// the request to a server its history records. One that has sent nothing since the -// proxy started has neither, and a quiet one the store has evicted (session.max_sessions, -// least recently used first, or a configured session.ttl) has no history, so either -// looks new unless the router pinned it; hence "can be". -const runningSessionsStay = "Sessions already running stay where they are, except one that has sent nothing " + - "since the proxy last started, or that the proxy has dropped from memory (it keeps the most recently used, " + - "100 by default), which can be treated as a new one." +// the request to a server its history records. +const runningSessionsStay = "Sessions already running stay where they are." // writeReport is what runServerWrite says about a change, by how it landed. type writeReport struct { // done is the change, said once the proxy has it or will load it at start. done string // live is what the change means for sessions already running. Said only when - // the running proxy reloaded it: that proxy's pins and history are what hold a - // running session where it is, and a proxy that starts later has neither. + // the running proxy reloaded it. live string // unchanged is said when there was nothing to write. unchanged string @@ -470,11 +463,6 @@ func runServerWrite(stdout, stderr io.Writer, path, statsURL string, ch edit.Con say(r.done, r.live, r.note) case edit.WriteNotRunning: next := fmt.Sprintf("Written to %s. No Cortex answered at its stats address, so this applies when the proxy next starts.", shown) - if r.live != "" { - // Not r.live: a proxy that starts holds no pins and no history. - next += " A proxy that starts knows where no session is, so it treats every session it sees as a new one, " + - "the ones running now included." - } say(r.done, next, r.note) case edit.WriteReloadFailed: // WritePluginConfig has put the file back, so it does not hold the change; diff --git a/cmd/agentop/cmd_server_write_test.go b/cmd/agentop/cmd_server_write_test.go index 5ef6e3fe5..cd4614ab7 100644 --- a/cmd/agentop/cmd_server_write_test.go +++ b/cmd/agentop/cmd_server_write_test.go @@ -347,11 +347,8 @@ func TestServerUse_RoutesTheAgentsNewSessions(t *testing.T) { path := serverEnv(t, newFakeStats(t, 0).addr(), routerBlock) code, out, errOut := runServerCmd(t, "", "use", "ete", "--agent", "opencode", "--config", path) // A running session stays because the proxy holds its pin or the request its - // history records: it has neither for a session it has not seen since it - // started, and no history for a quiet one its store has since evicted. - if code != 0 || out != "New opencode sessions → ete.\nSessions already running stay where they are, except one that has "+ - "sent nothing since the proxy last started, or that the proxy has dropped from memory (it keeps the most recently "+ - "used, 100 by default), which can be treated as a new one.\n" { + // history records. + if code != 0 || out != "New opencode sessions → ete.\nSessions already running stay where they are.\n" { t.Fatalf("exit %d, stdout:\n%s%s", code, out, errOut) } if !strings.Contains(readConfig(t, path), " claude-code: glm\n opencode: ete\n") { @@ -573,9 +570,8 @@ func TestServerRemove_TheOnlyServerWithAnAgentNamesBothSteps(t *testing.T) { } } -// A proxy that is not running holds no pins and, once started, no history, so it -// routes every session it then sees as a new one. Written for the next start, the -// change must not promise that running sessions stay where they are. +// Written for the next start, the change must not promise that running sessions +// stay where they are. func TestServerWrites_ForTheNextStartPromiseNoRunningSessionStays(t *testing.T) { for _, args := range [][]string{ {"use", "ete", "--agent", "opencode"}, @@ -591,8 +587,8 @@ func TestServerWrites_ForTheNextStartPromiseNoRunningSessionStays(t *testing.T) if strings.Contains(out, "stay where they are") || strings.Contains(out, "now gets an error") { t.Errorf("promises what a restart does not keep:\n%s", out) } - if !strings.Contains(flat(out), "treats every session it sees as a new one, the ones running now included") { - t.Errorf("does not say what the next start does:\n%s", out) + if strings.Contains(out, "new one") { + t.Errorf("says the next start takes running sessions for new ones:\n%s", out) } }) } @@ -716,10 +712,9 @@ func TestServerRemove_WithNoProxyRunningCountsNothing(t *testing.T) { } // A session resumed after a restart that has sent no inference since lists its history's host, -// but the router's pins did not survive the restart: nothing holds it to ete, so its next request -// goes to its agent's current server and gets no error. It is left out of the count; the session -// whose own request this proxy saw is not. -func TestServerRemove_LeavesOutASessionNoPinHolds(t *testing.T) { +// and its pin survived the restart in the plugin store: it is held where it was, as the session +// whose own request this proxy saw is, so both are counted. +func TestServerRemove_CountsASessionResumedAfterARestart(t *testing.T) { earlier := session.NewSummaryFold() e := inferenceTo("ete.example.com") earlier.Add("resumed", &e) @@ -729,7 +724,7 @@ func TestServerRemove_LeavesOutASessionNoPinHolds(t *testing.T) { store.Append("seen", inferenceTo("ete.example.com")) path := serverEnvWithSessions(t, newFakeStats(t, 0).addr(), serveSessions(t, store), routerBlock) code, out, errOut := runServerCmd(t, "", "remove", "ete", "--config", path) - if code != 0 || !strings.Contains(errOut, "warning: 1 running session ") { + if code != 0 || !strings.Contains(errOut, "warning: 2 running sessions ") { t.Errorf("exit %d, stdout:\n%s\nstderr:\n%s", code, out, errOut) } } diff --git a/cmd/agentop/tui/server_picker.go b/cmd/agentop/tui/server_picker.go index 9d40f0b17..6554e5cb8 100644 --- a/cmd/agentop/tui/server_picker.go +++ b/cmd/agentop/tui/server_picker.go @@ -314,21 +314,17 @@ func serverSwitchedText(msg serverSwitchedMsg, staying int) string { // too, but it is where new sessions go, so naming it would say nothing; one with no inference // yet is not anywhere, and its first request goes to the new choice. // -// RUNNING MEANS A REQUEST THIS PROXY SAW, because that is what the router pins a session by, in -// memory. A row only the archive holds has no pin, and neither has a resident one whose host is -// only its history's (InferenceHostFromHistory: resumed after a restart, no inference sent -// since) — its next request is decided as a new one's and goes to the new choice, so it does not -// stay. Neither is counted. After an eviction alone such a row's pin can survive, and leaving it -// out errs low, the quiet side. SessionSummary.Active would not do — it marks the one most -// recently updated session. +// RUNNING MEANS A SESSION THIS PROXY HAS SEEN A REQUEST FROM. One resumed after a restart counts, +// its host its history's (InferenceHostFromHistory): the router keeps its pins in the plugin +// store, so its pin survived. A row only the archive holds has sent nothing since and is not +// counted. SessionSummary.Active would not do — it marks the one most recently updated session. // // Read with the router as it was: a switch changes agents, not servers, so the hosts are the same. func (m *model) sessionsStaying(agent, server string) int { router, _ := m.activeRouter() n := 0 for _, s := range m.sessions { - if pipeline.AgentName(s.Agent) != agent || (s.Resident != nil && !*s.Resident) || s.InferenceHost == "" || - s.InferenceHostFromHistory { + if pipeline.AgentName(s.Agent) != agent || (s.Resident != nil && !*s.Resident) || s.InferenceHost == "" { continue } on, ok := servers.ForHost(router, s.InferenceHost) diff --git a/cmd/agentop/tui/server_picker_test.go b/cmd/agentop/tui/server_picker_test.go index bf2f31054..b4e2f4276 100644 --- a/cmd/agentop/tui/server_picker_test.go +++ b/cmd/agentop/tui/server_picker_test.go @@ -224,9 +224,9 @@ func TestServerPicker_TheFlashCountsTheRunningSessionsThatStay(t *testing.T) { } // A session resumed after a restart that has sent no inference since shows its history's host, -// but no pin holds it there: its next request is decided as a new one's and goes to the new -// choice. So it is not among those that stay — for a switch to a server, or back to own choice. -func TestServerPicker_TheCountLeavesOutASessionNoPinHolds(t *testing.T) { +// and its pin survived the restart in the plugin store, so it stays where it is like any other — +// for a switch to a server, or back to own choice. +func TestServerPicker_TheCountIncludesASessionResumedAfterARestart(t *testing.T) { for _, server := range []string{"glm", ""} { m := serverModel(t, newFakeProxy(t, false)) m.sessions = []session.SessionSummary{ @@ -234,8 +234,8 @@ func TestServerPicker_TheCountLeavesOutASessionNoPinHolds(t *testing.T) { {ID: "resumed", Agent: "claude-code", InferenceHost: "ete.example.com", InferenceHostFromHistory: true}, } switchTo(t, m, "claude-code/2.1.270", server) - if !strings.Contains(m.flash, " 1 running session stays where it is.") { - t.Errorf("switch to %q: flash %q, want the one session a pin holds", server, m.flash) + if !strings.Contains(m.flash, " 2 running sessions stay where they are.") { + t.Errorf("switch to %q: flash %q, want both sessions their pins hold", server, m.flash) } } } diff --git a/cmd/cortex/main.go b/cmd/cortex/main.go index 0579d821d..729a40b7d 100644 --- a/cmd/cortex/main.go +++ b/cmd/cortex/main.go @@ -279,6 +279,9 @@ func fatalf(format string, args ...any) { if closeArchiveOnFatal != nil { closeArchiveOnFatal() } + if closePluginStoreOnFatal != nil { + closePluginStoreOnFatal() + } log.Fatalf(format, args...) } @@ -408,6 +411,12 @@ func main() { } localInstall = startedFromLocalInstall(*configPath) + // Opened before the first pipeline build and kept for the process's life: every + // build, the initial one and each reload's, injects this same store, so what a + // plugin saves survives a reload as well as a restart. Nil off a local install. + pluginStore := openPluginStore(*configPath) + history := &archiveHistory{} + // Build the SPIFFE Provider when the spiffe block is configured. The // Provider drives both mTLS (via X509Source) and token-exchange's // spiffe identity (via JWTSource). Construction blocks until the first @@ -510,8 +519,8 @@ func main() { // cannot honor fails here — on startup and on every reload — instead of // running to no effect. The reverse proxy honors no redirect, which is what // keeps a WritesDestination plugin off the inbound chain. - inDeps := plugins.Deps{SPIFFE: provider, Pricing: pricingRegistry, Listener: reverseproxy.Support()} - outDeps := plugins.Deps{SPIFFE: provider, Pricing: pricingRegistry, Listener: forwardproxy.Support(c.MTLS != nil)} + inDeps := plugins.Deps{SPIFFE: provider, Pricing: pricingRegistry, Store: pluginStore, History: history, Listener: reverseproxy.Support()} + outDeps := plugins.Deps{SPIFFE: provider, Pricing: pricingRegistry, Store: pluginStore, History: history, Listener: forwardproxy.Support(c.MTLS != nil)} in, err := plugins.BuildWithDeps(c.Pipeline.Inbound.Plugins, inDeps) if err != nil { return nil, nil, nil, fmt.Errorf("inbound: %w", err) @@ -696,6 +705,7 @@ func main() { // sessions survive a restart and the store's eviction. See core/session/archive. sessArchive = openSessionArchive(cfg, *configPath, sessions) if sessArchive != nil { + history.open(sessArchive) // Before any listener starts: see replayUsage for why that is the whole design. replayUsageAtStartup(sessArchive, usageAgg) } @@ -1104,6 +1114,15 @@ func main() { } } + // The plugin store once no pipeline can still write to it: both have stopped. Close saves + // what the last interval held, so an orderly stop loses nothing; a SIGKILL loses at most + // that interval. + if pluginStore != nil { + if err := pluginStore.Close(); err != nil { + slog.Warn("plugin store: final save failed; changes since the last save are lost", "error", err) + } + } + if sessions != nil { sessions.Close() } diff --git a/cmd/cortex/plugin_store.go b/cmd/cortex/plugin_store.go new file mode 100644 index 000000000..bbb60ab33 --- /dev/null +++ b/cmd/cortex/plugin_store.go @@ -0,0 +1,51 @@ +package main + +import ( + "log/slog" + "path/filepath" + + "github.com/rossoctl/cortex/core/storage" + "github.com/rossoctl/cortex/core/storage/filestore" +) + +// pluginStoreFileName is the plugins' saved state under ~/.cortex, beside the cost +// ledger and the session archive. +const pluginStoreFileName = "plugin-state.json" + +// closePluginStoreOnFatal saves the plugin store before a fatal exit, for the reason +// closeLedgerOnFatal flushes the ledger. Nil whenever no store is open. +var closePluginStoreOnFatal func() + +// openPluginStore opens the store plugins keep state in across restarts (see +// storage.StoreConsumer): ~/.cortex/plugin-state.json, on a local install only, the +// one place a restart is routine and the state can live in the user's own directory. +// Elsewhere it returns nil, and so it does when the store cannot open — a warning, +// not a failed start: plugins then keep their state in memory, as they did before +// there was a store. +// +// The result is an interface that is nil when there is no store, never a nil +// *filestore.Store: plugins.Deps injects whatever is non-nil, and a typed nil would +// pass that check and panic in the plugin. +func openPluginStore(configPath string) storage.Store { + if !startedFromLocalInstall(configPath) { + return nil + } + dir, err := defaultCortexDir() + if err != nil { + slog.Warn("plugin store disabled — cannot determine where to write it; plugins keep their state in memory", "error", err) + return nil + } + path := filepath.Join(dir, pluginStoreFileName) + st, err := filestore.Open(path) + if err != nil { + slog.Warn("plugin store disabled — plugins keep their state in memory and lose it on restart", "path", path, "error", err) + return nil + } + closePluginStoreOnFatal = func() { + if cerr := st.Close(); cerr != nil { + slog.Warn("plugin store: final save failed during a fatal startup error", "error", cerr) + } + } + slog.Info("plugin store enabled — plugins' saved state survives a restart", "path", path) + return st +} diff --git a/cmd/cortex/plugin_store_test.go b/cmd/cortex/plugin_store_test.go new file mode 100644 index 000000000..b5a27aa16 --- /dev/null +++ b/cmd/cortex/plugin_store_test.go @@ -0,0 +1,46 @@ +package main + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" +) + +// Only a local install gets a store: it is the one place a restart is routine and +// the state can live in the user's own ~/.cortex. +func TestOpenPluginStore_OnlyOnALocalInstall(t *testing.T) { + inside, outside := localConfig(t) + if st := openPluginStore(outside); st != nil { + _ = st.Close() + t.Error("opened a store for a config outside ~/.cortex") + } + st := openPluginStore(inside) + if st == nil { + t.Fatal("no store for a local install") + } + if err := st.Set(context.Background(), "k", "v", time.Hour); err != nil { + t.Fatal(err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(filepath.Join(filepath.Dir(inside), pluginStoreFileName)); err != nil { + t.Errorf("the store is not at ~/.cortex/%s: %v", pluginStoreFileName, err) + } +} + +// A store that cannot open is a warning, not a failed start: plugins then keep their +// state in memory, as they did before there was a store. +func TestOpenPluginStore_AFailureToOpenIsNoStore(t *testing.T) { + inside, _ := localConfig(t) + // A directory where the file should be: nothing can be read or written there. + if err := os.MkdirAll(filepath.Join(filepath.Dir(inside), pluginStoreFileName), 0o700); err != nil { + t.Fatal(err) + } + if st := openPluginStore(inside); st != nil { + _ = st.Close() + t.Error("opened a store over a directory") + } +} diff --git a/cmd/cortex/session_history.go b/cmd/cortex/session_history.go new file mode 100644 index 000000000..2ebcd0a27 --- /dev/null +++ b/cmd/cortex/session_history.go @@ -0,0 +1,25 @@ +package main + +import ( + "sync/atomic" + + "github.com/rossoctl/cortex/core/pipeline" + "github.com/rossoctl/cortex/core/session/archive" +) + +// archiveHistory is the session archive as plugins read it (session.History). The first +// pipeline build runs before the archive opens, so every build is handed this instead, and it +// reads the archive once open sets it; until then, and with no archive, it reads nothing. +type archiveHistory struct { + archive atomic.Pointer[archive.Archive] +} + +func (h *archiveHistory) open(a *archive.Archive) { h.archive.Store(a) } + +func (h *archiveHistory) Earlier(id string, before uint64, fn func(*pipeline.SessionEvent) bool) error { + a := h.archive.Load() + if a == nil { + return nil + } + return a.Earlier(id, before, fn) +} diff --git a/cmd/cortex/session_history_test.go b/cmd/cortex/session_history_test.go new file mode 100644 index 000000000..133dab5c6 --- /dev/null +++ b/cmd/cortex/session_history_test.go @@ -0,0 +1,42 @@ +package main + +import ( + "testing" + "time" + + "github.com/rossoctl/cortex/core/pipeline" + "github.com/rossoctl/cortex/core/session/archive" +) + +func historySeqs(t *testing.T, h *archiveHistory, id string) []uint64 { + t.Helper() + var seqs []uint64 + if err := h.Earlier(id, 0, func(e *pipeline.SessionEvent) bool { + seqs = append(seqs, e.Seq) + return true + }); err != nil { + t.Fatal(err) + } + return seqs +} + +// Plugins hold the history from the first build, before the archive opens: it reads nothing +// until then, and the archive's events once it has. +func TestArchiveHistory_ReadsTheArchiveOnceItOpens(t *testing.T) { + root := t.TempDir() + recordTraffic(t, root, nil, time.Now()) + h := &archiveHistory{} + if seqs := historySeqs(t, h, "a"); len(seqs) != 0 { + t.Fatalf("before open: %v, want nothing", seqs) + } + + arch, err := archive.Open(root) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = arch.Close() }) + h.open(arch) + if seqs := historySeqs(t, h, "a"); len(seqs) != 4 || seqs[0] != 4 { + t.Errorf("after open: %v, want session a's 4 events newest first", seqs) + } +} diff --git a/core/plugins/deps.go b/core/plugins/deps.go index 2e7238454..478c01fef 100644 --- a/core/plugins/deps.go +++ b/core/plugins/deps.go @@ -8,7 +8,9 @@ import ( "github.com/rossoctl/cortex/core/config" "github.com/rossoctl/cortex/core/cost/pricing" "github.com/rossoctl/cortex/core/pipeline" + "github.com/rossoctl/cortex/core/session" "github.com/rossoctl/cortex/core/spiffe" + "github.com/rossoctl/cortex/core/storage" ) // Deps are the process-wide dependencies a pipeline build can inject into plugins. @@ -29,6 +31,15 @@ type Deps struct { // place, so a plugin from any build sees current rates. See pricing.Registry. Pricing *pricing.Registry + // Store is injected into plugins implementing storage.StoreConsumer: the process's + // state that outlives it. Long-lived like Pricing: every build gets the same one, + // so state a plugin saves survives a reload as well as a restart. + Store storage.Store + + // History is injected into plugins implementing session.HistoryConsumer: the + // session archive's events, for a process that keeps one. + History session.History + // Listener says which listener-dependent capabilities the listener this pipeline // serves can honor; a plugin declaring one it cannot is refused. The zero value // honors none of them, so a build that names no listener admits no such plugin. @@ -89,6 +100,12 @@ func BuildWithDeps(entries []config.PluginEntry, deps Deps, opts ...pipeline.Opt if c, ok := p.(pricing.ResolverConsumer); ok && deps.Pricing != nil { c.SetPricingResolver(deps.Pricing) } + if c, ok := p.(storage.StoreConsumer); ok && deps.Store != nil { + c.SetStore(deps.Store) + } + if c, ok := p.(session.HistoryConsumer); ok && deps.History != nil { + c.SetHistory(deps.History) + } if c, ok := p.(pipeline.Configurable); ok { if err := c.Configure(e.Config); err != nil { diff --git a/core/plugins/deps_test.go b/core/plugins/deps_test.go index 0fb227dc6..422808bf5 100644 --- a/core/plugins/deps_test.go +++ b/core/plugins/deps_test.go @@ -3,12 +3,16 @@ package plugins import ( "context" "encoding/json" + "path/filepath" "strings" "testing" "github.com/rossoctl/cortex/core/config" "github.com/rossoctl/cortex/core/cost/pricing" "github.com/rossoctl/cortex/core/pipeline" + "github.com/rossoctl/cortex/core/session" + "github.com/rossoctl/cortex/core/storage" + "github.com/rossoctl/cortex/core/storage/filestore" ) // pricedPlugin records what was injected and when, so the tests can assert the @@ -85,6 +89,56 @@ func TestBuildWithDeps_NoResolverLeavesConsumerUncalled(t *testing.T) { } } +// storedPlugin records the store it was handed and whether it had it at Configure. +type storedPlugin struct { + pricedPlugin + store storage.Store + hadAtConfigure bool +} + +func (p *storedPlugin) SetStore(s storage.Store) { p.store = s } +func (p *storedPlugin) Configure(json.RawMessage) error { + p.hadAtConfigure = p.store != nil + return nil +} + +func registerStored(t *testing.T, name string) *storedPlugin { + t.Helper() + p := &storedPlugin{pricedPlugin: pricedPlugin{name: name}} + RegisterPlugin(name, func() pipeline.Plugin { return p }) + t.Cleanup(func() { UnregisterPlugin(name) }) + return p +} + +func TestBuildWithDeps_InjectsTheStoreBeforeConfigure(t *testing.T) { + p := registerStored(t, "test-stored-order") + st, err := filestore.Open(filepath.Join(t.TempDir(), "s.json")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = st.Close() }) + + if _, err := BuildWithDeps(entriesFor("test-stored-order"), Deps{Store: st}); err != nil { + t.Fatalf("BuildWithDeps: %v", err) + } + if p.store != storage.Store(st) { + t.Errorf("store = %v, want the Deps store", p.store) + } + if !p.hadAtConfigure { + t.Error("the store was not injected before Configure ran") + } +} + +func TestBuildWithDeps_NoStoreLeavesConsumerUncalled(t *testing.T) { + p := registerStored(t, "test-stored-none") + if _, err := BuildWithDeps(entriesFor("test-stored-none"), Deps{}); err != nil { + t.Fatalf("BuildWithDeps: %v", err) + } + if p.store != nil { + t.Errorf("store = %v, want nil when no Deps.Store was supplied", p.store) + } +} + func TestBuild_IsBuildWithDepsWithNoDeps(t *testing.T) { // Build and BuildWithSPIFFE had near-identical 40-line bodies. They now // delegate, so this asserts the delegation preserved their behaviour. @@ -223,3 +277,36 @@ func TestBuildWithDeps_DoesNotCheckAnEntryThatIsOff(t *testing.T) { t.Fatalf("BuildWithDeps checked an entry that is off: %v", err) } } + +// historyPlugin records the history it was handed and whether it had it at Configure. +type historyPlugin struct { + pricedPlugin + history session.History + hadAtConfigure bool +} + +func (p *historyPlugin) SetHistory(h session.History) { p.history = h } +func (p *historyPlugin) Configure(json.RawMessage) error { + p.hadAtConfigure = p.history != nil + return nil +} + +type noHistory struct{} + +func (noHistory) Earlier(string, uint64, func(*pipeline.SessionEvent) bool) error { return nil } + +func TestBuildWithDeps_InjectsTheHistoryBeforeConfigure(t *testing.T) { + p := &historyPlugin{pricedPlugin: pricedPlugin{name: "test-history-order"}} + RegisterPlugin(p.name, func() pipeline.Plugin { return p }) + t.Cleanup(func() { UnregisterPlugin(p.name) }) + + if _, err := BuildWithDeps(entriesFor(p.name), Deps{History: noHistory{}}); err != nil { + t.Fatalf("BuildWithDeps: %v", err) + } + if p.history != session.History(noHistory{}) { + t.Errorf("history = %v, want the Deps history", p.history) + } + if !p.hadAtConfigure { + t.Error("the history was not injected before Configure ran") + } +} diff --git a/core/plugins/inferencerouter/history_test.go b/core/plugins/inferencerouter/history_test.go new file mode 100644 index 000000000..2b93a22d2 --- /dev/null +++ b/core/plugins/inferencerouter/history_test.go @@ -0,0 +1,261 @@ +package inferencerouter + +import ( + "encoding/json" + "errors" + "net/http" + "strings" + "testing" + + "github.com/rossoctl/cortex/core/config" + "github.com/rossoctl/cortex/core/pipeline" + "github.com/rossoctl/cortex/core/plugins" +) + +// fakeHistory is a session archive: each session's events in the order they were +// recorded. It notes the before of every read and how many events it handed out. +type fakeHistory struct { + events map[string][]pipeline.SessionEvent + err error + befores []uint64 + read int +} + +func (h *fakeHistory) Earlier(id string, before uint64, fn func(*pipeline.SessionEvent) bool) error { + h.befores = append(h.befores, before) + if h.err != nil { + return h.err + } + evs := h.events[id] + for i := len(evs) - 1; i >= 0; i-- { + if before > 0 && evs[i].Seq >= before { + continue + } + h.read++ + if !fn(&evs[i]) { + return nil + } + } + return nil +} + +// archived numbers events from seq 1 as the archive holds them. +func archived(events ...pipeline.SessionEvent) []pipeline.SessionEvent { + for i := range events { + events[i].Seq = uint64(i + 1) + } + return events +} + +// buildWithHistory is build with h injected before Configure, as plugins.BuildWithDeps does. +func buildWithHistory(t *testing.T, config string, h *fakeHistory) *pipeline.Pipeline { + t.Helper() + r := New() + r.SetHistory(h) + if err := r.Configure(json.RawMessage(config)); err != nil { + t.Fatalf("Configure: %v", err) + } + p, err := pipeline.New([]pipeline.Plugin{r}) + if err != nil { + t.Fatalf("pipeline.New: %v", err) + } + return p +} + +// The restart that first keeps pins on disk starts with none, and a restart with the +// plugin state lost does the same: the proxy's memory holds nothing of a running +// session either. Its archived events still say where it went, and that history keeps +// a Claude Code continuation on glm rather than leaving it where Claude Code sends it. +// Another agent's later row, filed under the session, is no evidence of its server. +func TestRouter_ASessionsArchivedHistoryKeepsItsServerAcrossARestart(t *testing.T) { + h := &fakeHistory{events: map[string][]pipeline.SessionEvent{ + "running": archived(sent(glmHost, claudeUA), sent(eteHost, "curl/8.7.1")), + }} + store := newStore(t) + p := buildWithHistory(t, routerConfig(`"claude-code": "glm"`), h) + pctx := claudeCode(request(store, eteHost, claudeUA, "running"), pipeline.AgentRoleMain, 26, 3) + run(t, p, pctx) + + assertRouted(t, pctx, glmHost, "glm-key") + assertRecord(t, pctx, pipeline.ActionModify, "routed", map[string]string{"server": "glm", "pin": pinNew}) + if pin, _ := pinOf(t, store, "running"); pin != "glm" { + t.Errorf("pin = %q, want glm", pin) + } +} + +// Memory is read first, and the archive only below the oldest event memory holds: +// above it the archive holds the same events, which memory already showed are no +// evidence. +func TestRouter_TheArchiveIsReadBelowTheOldestEventInMemory(t *testing.T) { + tunnel := pipeline.SessionEvent{Seq: 5, Direction: pipeline.Outbound, Phase: pipeline.SessionRequest, + Host: eteHost + ":443", Tunnel: true, HTTPMethod: http.MethodConnect, Client: pipeline.ParseUserAgent(claudeUA)} + h := &fakeHistory{events: map[string][]pipeline.SessionEvent{ + "running": archived(sent(glmHost, claudeUA), sent(eteHost, claudeUA), sent(glmHost, claudeUA), sent(eteHost, opencodeUA), tunnel), + }} + p := buildWithHistory(t, routerConfig(`"claude-code": "glm"`), h) + pctx := withHistory(claudeCode(request(newStore(t), eteHost, claudeUA, "running"), pipeline.AgentRoleMain, 26, 3), tunnel) + run(t, p, pctx) + + assertRouted(t, pctx, glmHost, "glm-key") + if len(h.befores) != 1 || h.befores[0] != 5 { + t.Errorf("archive read with before = %v, want one read below seq 5", h.befores) + } +} + +// What memory shows is not read again from the archive. +func TestRouter_HistoryInMemoryNeedsNoArchiveRead(t *testing.T) { + h := &fakeHistory{} + p := buildWithHistory(t, routerConfig(`"claude-code": "glm"`), h) + pctx := withHistory(request(newStore(t), eteHost, claudeUA, "running"), sent(eteHost, claudeUA)) + run(t, p, pctx) + + assertRouted(t, pctx, eteHost, "ete-key") + if len(h.befores) != 0 { + t.Errorf("archive read %d times, want none", len(h.befores)) + } +} + +// With nothing in the archive to go on — no session, no request of this agent's to a +// server, or a read that failed — the request's own turn decides, as without an +// archive. +func TestRouter_WithNoArchivedEvidenceTheTurnDecides(t *testing.T) { + for _, tc := range []struct { + name string + h *fakeHistory + }{ + {"nothing archived", &fakeHistory{}}, + {"only another agent's request", &fakeHistory{events: map[string][]pipeline.SessionEvent{ + "running": archived(sent(glmHost, opencodeUA)), + }}}, + {"a failed read", &fakeHistory{err: errors.New("segment unreadable")}}, + } { + t.Run(tc.name, func(t *testing.T) { + p := buildWithHistory(t, routerConfig(`"claude-code": "glm"`), tc.h) + pctx := claudeCode(request(newStore(t), eteHost, claudeUA, "running"), pipeline.AgentRoleMain, 26, 3) + run(t, p, pctx) + + assertUntouched(t, pctx, eteHost) + assertRecord(t, pctx, pipeline.ActionSkip, "not_routed", map[string]string{"pin": pinNew, "turn": turnContinuation}) + }) + } +} + +// The archive is read on the request path, so the search stops after historyLimit +// events; past it the request's own turn decides. +func TestRouter_TheArchiveSearchStopsAtItsLimit(t *testing.T) { + events := []pipeline.SessionEvent{sent(glmHost, claudeUA)} + for range historyLimit { + events = append(events, sent(glmHost, opencodeUA)) + } + h := &fakeHistory{events: map[string][]pipeline.SessionEvent{"running": archived(events...)}} + p := buildWithHistory(t, routerConfig(`"claude-code": "glm"`), h) + pctx := claudeCode(request(newStore(t), eteHost, claudeUA, "running"), pipeline.AgentRoleMain, 26, 3) + run(t, p, pctx) + + assertUntouched(t, pctx, eteHost) + if h.read != historyLimit { + t.Errorf("read %d archived events, want %d", h.read, historyLimit) + } +} + +// A request no parser read as inference — Claude Code's count_tokens, a /v1/models +// list — says nothing about where a conversation is, so it pins nothing: it goes to +// its agent's current server, and the session's next inference request decides. +// Claude Code sends count_tokens mid-conversation, and as the first request after an +// idle gap. +func TestRouter_ARequestWithNoInferenceParseDecidesNothing(t *testing.T) { + unparsed := func(store pipeline.SharedStore, ua string) *pipeline.Context { + pctx := request(store, eteHost, ua, "s1") + pctx.Path, pctx.Extensions.Inference = "/v1/messages/count_tokens", nil + return pctx + } + + t.Run("a Claude Code continuation after it is still one", func(t *testing.T) { + store := newStore(t) + p := build(t, routerConfig(`"claude-code": "glm"`)) + first := unparsed(store, claudeUA) + run(t, p, first) + assertRouted(t, first, glmHost, "glm-key") + assertRecord(t, first, pipeline.ActionModify, "routed", map[string]string{"server": "glm", "pin": pinNone}) + if _, pinned := pinOf(t, store, "s1"); pinned { + t.Fatal("an unparsed request pinned the session") + } + + next := claudeCode(request(store, eteHost, claudeUA, "s1"), pipeline.AgentRoleMain, 26, 3) + run(t, p, next) + assertUntouched(t, next, eteHost) + assertRecord(t, next, pipeline.ActionSkip, "not_routed", map[string]string{"pin": pinNew, "turn": turnContinuation}) + }) + + t.Run("another agent's session starts on the server it was on at its first inference request", func(t *testing.T) { + store := newStore(t) + run(t, build(t, routerConfig(`"opencode": "glm"`)), unparsed(store, opencodeUA)) + if _, pinned := pinOf(t, store, "s1"); pinned { + t.Fatal("an unparsed request pinned the session") + } + + next := request(store, eteHost, opencodeUA, "s1") + run(t, build(t, routerConfig(`"opencode": "ete"`)), next) + assertRouted(t, next, eteHost, "ete-key") + assertRecord(t, next, pipeline.ActionModify, "routed", map[string]string{"server": "ete", "pin": pinNew}) + }) + + t.Run("history still decides", func(t *testing.T) { + store := newStore(t) + p := build(t, routerConfig(`"claude-code": "glm"`)) + pctx := withHistory(unparsed(store, claudeUA), sent(eteHost, claudeUA)) + run(t, p, pctx) + assertRouted(t, pctx, eteHost, "ete-key") + if pin, _ := pinOf(t, store, "s1"); pin != "ete" { + t.Errorf("pin = %q, want ete", pin) + } + }) +} + +// The router decides from inference-parser's record of each request, so a chain +// without the parser before it is refused rather than run unable to pin a session. +func TestRouter_RequiresInferenceParserBeforeIt(t *testing.T) { + _, err := plugins.BuildWithDeps([]config.PluginEntry{{Name: Name, Config: json.RawMessage(routerConfig(``))}}, + plugins.Deps{Listener: pipeline.ListenerSupport{Listener: "forward proxy", Destination: true}}) + if err == nil || !strings.Contains(err.Error(), `requires "inference-parser" earlier in the chain`) { + t.Fatalf("err = %v, want the router refused for want of inference-parser", err) + } +} + +// A side request decides nothing, and so its row is no evidence of where the session +// is, in memory or in the archive: the conversation's own requests decide. A row of +// the conversation's own, older than the side requests, still does. +func TestRouter_ASideRequestsRowIsNoEvidence(t *testing.T) { + oneShot := claudeCode(request(nil, eteHost, claudeUA, "s1"), pipeline.AgentRoleMain, 0, 0) + subagent := claudeCode(request(nil, eteHost, claudeUA, "s1"), pipeline.AgentRoleSubagent, 11, 2) + turn := claudeCode(request(nil, glmHost, claudeUA, "s1"), pipeline.AgentRoleMain, 26, 1) + for _, tc := range []struct { + name string + memory, archive []pipeline.SessionEvent + turns int // assistant turns in the request + host string // where it goes + pin string + }{ + {"opening turn after side rows in memory", []pipeline.SessionEvent{recorded(oneShot), recorded(subagent)}, nil, 0, glmHost, "glm"}, + {"opening turn after side rows in the archive", nil, []pipeline.SessionEvent{recorded(oneShot), recorded(subagent)}, 0, glmHost, "glm"}, + {"continuation after side rows only", nil, []pipeline.SessionEvent{recorded(subagent), recorded(oneShot)}, 3, eteHost, ""}, + {"continuation whose own turn is older than the side rows", nil, + []pipeline.SessionEvent{recorded(turn), recorded(oneShot), recorded(subagent)}, 3, glmHost, "glm"}, + } { + t.Run(tc.name, func(t *testing.T) { + store := newStore(t) + h := &fakeHistory{events: map[string][]pipeline.SessionEvent{"s1": archived(tc.archive...)}} + p := buildWithHistory(t, routerConfig(`"claude-code": "glm"`), h) + pctx := withHistory(claudeCode(request(store, eteHost, claudeUA, "s1"), pipeline.AgentRoleMain, 26, tc.turns), + tc.memory...) + run(t, p, pctx) + + if pctx.Host != tc.host { + t.Errorf("Host = %q, want %q", pctx.Host, tc.host) + } + if pin, pinned := pinOf(t, store, "s1"); !pinned || pin != tc.pin { + t.Errorf("pin = %q (pinned %v), want %q", pin, pinned, tc.pin) + } + }) + } +} diff --git a/core/plugins/inferencerouter/models_test.go b/core/plugins/inferencerouter/models_test.go index 0af8c826a..da91b0c12 100644 --- a/core/plugins/inferencerouter/models_test.go +++ b/core/plugins/inferencerouter/models_test.go @@ -463,10 +463,11 @@ func TestRouter_RunsLastInTheLaptopChain(t *testing.T) { // since it would read the server's model rather than the client's. func TestRouter_IsRefusedBeforeABodyReader(t *testing.T) { _, err := plugins.BuildWithDeps([]config.PluginEntry{ - {Name: Name, Config: json.RawMessage(mappedConfig(``))}, {Name: "inference-parser"}, + {Name: Name, Config: json.RawMessage(mappedConfig(``))}, + {Name: "mcp-parser"}, }, plugins.Deps{Listener: pipeline.ListenerSupport{Listener: "forward proxy", Destination: true}}) - if err == nil || !strings.Contains(err.Error(), `"inference-parser" reads body after mutator "inference-router"`) { + if err == nil || !strings.Contains(err.Error(), `"mcp-parser" reads body after mutator "inference-router"`) { t.Fatalf("err = %v, want the reader rule naming both", err) } } diff --git a/core/plugins/inferencerouter/pins.go b/core/plugins/inferencerouter/pins.go new file mode 100644 index 000000000..e144ba28f --- /dev/null +++ b/core/plugins/inferencerouter/pins.go @@ -0,0 +1,109 @@ +package inferencerouter + +import ( + "context" + "encoding/json" + "log/slog" + "time" + + "github.com/rossoctl/cortex/core/pipeline" + "github.com/rossoctl/cortex/core/storage" +) + +// pins is where the router keeps its pins: the process's durable store when one was +// injected, so a restart forgets none, else the process store, which a reload keeps +// and a restart empties. +type pins interface { + // load is key's pin, and whether there is one. + load(key string) (pin, bool) + // save pins key to pn for pinTTL. + save(key string, pn pin) + // keep renews key's pin, pn, for another pinTTL: its session is still in use. + keep(key string, pn pin) +} + +// pinsFor is where pctx's pins are kept, nil when there is nowhere. +func (p *Router) pinsFor(pctx *pipeline.Context) pins { + if p.store != nil { + now := p.now + if now == nil { + now = time.Now + } + return storedPins{st: p.store, now: now} + } + if pctx.Shared != nil { + return memPins{st: pctx.Shared} + } + return nil +} + +// memPins keeps pins in the process store, as the pin struct itself. +type memPins struct{ st pipeline.SharedStore } + +func (m memPins) load(key string) (pin, bool) { + v, ok := m.st.Get(key) + pn, isPin := v.(pin) + return pn, ok && isPin +} + +func (m memPins) save(key string, pn pin) { m.st.Put(key, pn, pinTTL) } +func (m memPins) keep(key string, pn pin) { m.st.Put(key, pn, pinTTL) } + +// pinRecord is a pin as the durable store holds it. +type pinRecord struct { + Agent string `json:"agent"` + Server string `json:"server"` // "" for not routed + // Renewed is the Unix time, in seconds, the pin's TTL was last set. + Renewed int64 `json:"renewed"` +} + +// renewAfter is how old a stored pin's TTL gets before a request renews it. Every +// request keeps its session's pin, and a store write each time would have the store +// save its file every interval while any session is busy; renewed once a day, a +// pin's TTL runs between 29 and 30 days from its session's last request. +const renewAfter = 24 * time.Hour + +// storedPins keeps pins in the durable store, as JSON pin records. +type storedPins struct { + st storage.Store + now func() time.Time +} + +// record is key's stored record. A value that is not one is no pin: the session is +// decided as unpinned, and its pin is written over it. +func (d storedPins) record(key string) (pinRecord, bool) { + raw, err := d.st.Get(context.Background(), key) + if err != nil { + slog.Warn("inference-router: reading a pin from the plugin store failed; the session is decided as unpinned", + "error", err) + return pinRecord{}, false + } + var rec pinRecord + if raw == "" || json.Unmarshal([]byte(raw), &rec) != nil { + return pinRecord{}, false + } + return rec, true +} + +func (d storedPins) load(key string) (pin, bool) { + rec, ok := d.record(key) + return pin{agent: rec.Agent, server: rec.Server}, ok +} + +func (d storedPins) save(key string, pn pin) { + b, err := json.Marshal(pinRecord{Agent: pn.agent, Server: pn.server, Renewed: d.now().Unix()}) + if err == nil { + err = d.st.Set(context.Background(), key, string(b), pinTTL) + } + if err != nil { + slog.Warn("inference-router: saving a pin to the plugin store failed; the session's next request decides again", + "error", err) + } +} + +func (d storedPins) keep(key string, pn pin) { + if rec, ok := d.record(key); ok && d.now().Sub(time.Unix(rec.Renewed, 0)) < renewAfter { + return + } + d.save(key, pn) +} diff --git a/core/plugins/inferencerouter/plugin.go b/core/plugins/inferencerouter/plugin.go index f39ff3fbb..6a78a0ab5 100644 --- a/core/plugins/inferencerouter/plugin.go +++ b/core/plugins/inferencerouter/plugin.go @@ -4,8 +4,11 @@ // // The choice is plugin config — servers by name, and agents by name to a server — // changed by editing it and letting the proxy hot-reload. Sessions are pinned in the -// process-scoped store (pctx.Shared), which a reload does not replace, so changing -// an agent's server moves only the sessions it has not started yet. +// process's durable store (storage.StoreConsumer) — on a local install a file under +// ~/.cortex — which neither a reload nor a restart loses, so changing an agent's +// server moves only the sessions it has not started yet. A process with no durable +// store pins in its process-scoped store (pctx.Shared), which a reload keeps and a +// restart empties. // // Place it last in the outbound chain, which is where agentop puts it. A redirect // moves pctx.Host, so the plugins before it decide on the host the client asked for @@ -25,33 +28,48 @@ // nothing, so a key set on the strength of that nil would go to the host the client // named. // -// A session is pinned by where its first request went, not by the choice made for -// it: to the server when the request was redirected there; to "not routed" when it -// stayed where the client sent it, because the agent is not routed or because the -// router runs under on_error: observe; and not at all when the redirect failed or -// the request was refused for its model, so the session's next request decides -// again. A pin by choice would hold a session started under observe to a server it -// never used, and turning enforce on would then move it there mid-conversation, -// which is the switch the pin exists to prevent. +// A session is pinned by where the request that decides it went, not by the choice +// made for it: to the server when the request was redirected there; to "not routed" +// when it stayed where the client sent it, because the agent is not routed or +// because the router runs under on_error: observe; and not at all when the redirect +// failed or the request was refused for its model, so the session's next request +// decides again. A pin by choice would hold a session started under observe to a +// server it never used, and turning enforce on would then move it there +// mid-conversation, which is the switch the pin exists to prevent. // // The first request the router sees is not always a session's first. A session // already running when routing is first configured, quiet while it was, has no pin, -// and neither has one whose pin lapsed. The session store's history tells such a -// session from a new one: on a pin miss for a routed agent, the latest earlier -// inference request that agent sent in the session to a server's host keeps the -// session on that server, and only a session with no such request is new and goes -// to its agent's current server. Without this, the documented first setup — add -// the servers, then route Claude Code — would move every conversation that sent +// and neither has one whose pin lapsed. The session's history tells such a session +// from a new one: on a pin miss for a routed agent, the latest earlier inference +// request, not a side request, that agent sent in the session to a server's host +// keeps the session on that server, and only a session with no such request is new +// and goes to its agent's current server. Without this, the documented first setup — +// add the servers, then route Claude Code — would move every conversation that sent // nothing between the router's arrival and the route to the new server on its next // turn. A request to any other host is no evidence: the router never routes one, so // it says nothing about which server the session is on. Reading it as "not routed" // would leave an agent that switches providers inside a session, as OpenCode does, -// unrouted for good once its first request went elsewhere, still sending its own -// key to the server it then addresses. Only a non-empty history is evidence: a -// view is empty for a session the store has recorded nothing of yet. After a -// restart the store holds nothing from before it, and the pins, which live in -// memory, are gone too, so a running session's next request then looks new and can -// move; so can a quiet one the store has evicted before the router pinned it. +// unrouted for good once its first request went elsewhere, still sending its own key +// to the server it then addresses. Only a non-empty history is evidence: a view is +// empty for a session the store has recorded nothing of yet. +// +// With no pin and no history, a Claude Code inference request says itself whether +// the router is seeing its conversation begin. Its system prompt states the caller, +// and the main conversation carries its tool manifest: an opening turn has no +// assistant message yet and goes to the agent's current server, and a continuation +// has some, so the router did not see it begin — it started before routing was set +// up, or lost its pin to a restart of a process that keeps pins only in memory — and +// it is not routed and pinned so, rather than taken for new and moved +// mid-conversation. The one-shots Claude Code interleaves with a conversation carry +// no tools, and a subagent's requests are its own conversation's; neither says +// whether the session began, so neither decides: each goes where Claude Code sent it +// and leaves the session unpinned. Only Claude Code is read this way, because no +// other agent states its role, and one that switches providers, as OpenCode does, +// sends earlier turns that went elsewhere. A session routed to a server keeps it by +// its pin and its history, so a restart that keeps neither sends such a conversation +// back to where Claude Code sends it. So does a move to a new session id — Claude +// Code's continued-in hand-off, or a background job forked from a conversation — +// which neither follows. // // A pin is its agent's. The session id is the listener's answer, and another // agent's request can be filed under it — by process attribution, which files a @@ -102,6 +120,7 @@ import ( "github.com/rossoctl/cortex/core/plugins" "github.com/rossoctl/cortex/core/plugins/inferencerouter/routerconfig" "github.com/rossoctl/cortex/core/session" + "github.com/rossoctl/cortex/core/storage" "github.com/tidwall/gjson" ) @@ -116,6 +135,11 @@ const ( // request renews it, so only a session idle this long forgets its server. pinTTL = 30 * 24 * time.Hour + // historyLimit is how many of a session's archived events wentTo reads, newest + // first, before it stops looking: the archive is read from disk on the request + // path. + historyLimit = 2000 + // codeUnavailable maps to 503 in pipeline's code table. A session whose server // cannot be reached through the router is the server being unavailable to it, // not the client's mistake. @@ -136,7 +160,14 @@ const ( const ( pinNew = "new" // this request pinned the session pinExisting = "existing" // the session was already pinned - pinNone = "none" // nothing was pinned: no session, a synthetic one, no store, a failed redirect, a refused model, or another agent's pin + pinNone = "none" // nothing was pinned: no session, a synthetic one, no store, a failed redirect, a refused model, another agent's pin, a side request, or an unparsed request +) + +// The turn detail on a not-routed record, where the request's own turn decided an +// unpinned Claude Code session. +const ( + turnContinuation = "continuation" // a conversation the router did not see begin + turnAside = "aside" // a one-shot or a subagent's request, which decides nothing ) // route is one configured server, ready to redirect to. @@ -160,13 +191,30 @@ type Router struct { byHost map[string]string // every server's Endpoint.Hostname, to its name agents map[string]string // agent name to server name + // store is the process's durable store (storage.StoreConsumer), where pins are + // kept when there is one; nil keeps them in pctx.Shared. + store storage.Store + // history is the session archive (session.HistoryConsumer), nil in a process with + // none. + history session.History + // now is the clock a stored pin's renewal is judged by. + now func() time.Time + // noStore makes the "pins are off" warning once per instance rather than once // per request. noStore sync.Once } // New constructs an unconfigured plugin. -func New() *Router { return &Router{} } +func New() *Router { return &Router{now: time.Now} } + +// SetStore implements storage.StoreConsumer: the router keeps its pins in st, so a +// restart forgets none. +func (p *Router) SetStore(st storage.Store) { p.store = st } + +// SetHistory implements session.HistoryConsumer: the router reads a session's archived +// events for where it went before this process (see wentTo). +func (p *Router) SetHistory(h session.History) { p.history = h } func init() { plugins.RegisterPlugin(Name, func() pipeline.Plugin { return New() }) @@ -178,6 +226,7 @@ func (p *Router) Capabilities() pipeline.PluginCapabilities { return pipeline.PluginCapabilities{ WritesDestination: true, WritesRequestBody: true, // SetRequestModel, for a server with models of its own + Requires: []string{"inference-parser"}, Description: "Sends each agent's new sessions to its chosen inference server.", } } @@ -244,8 +293,11 @@ func (p *Router) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.A if name == "" { // Not routed: the request, its key included, stays exactly as the client sent it. pin.settle("") - pctx.Record(pipeline.Invocation{Action: pipeline.ActionSkip, Reason: "not_routed", - Details: map[string]string{"pin": pin.state}}) + details := map[string]string{"pin": pin.state} + if pin.turn != "" { + details["turn"] = pin.turn + } + pctx.Record(pipeline.Invocation{Action: pipeline.ActionSkip, Reason: "not_routed", Details: details}) return cont } // Built where each record is made, since a failed redirect changes the pin. @@ -376,8 +428,8 @@ func (p *Router) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Act return pipeline.Action{Type: pipeline.Continue} } -// pin is a session's pin as the shared store holds it: the agent whose session it -// is, and the server the session is on, "" for not routed. +// pin is a session's pin: the agent whose session it is, and the server the session +// is on, "" for not routed. See pins for where it is kept. type pin struct { agent string server string @@ -387,9 +439,12 @@ type pin struct { // session's first request where to store the outcome once OnRequest knows it. type pinning struct { state string // pinNew, pinExisting or pinNone - store pipeline.SharedStore + store pins key string agent string + // turn is turnContinuation or turnAside when the request's own turn left an + // unpinned session unrouted, "" otherwise. + turn string } // settle pins a session on its first request to where that request went: server @@ -397,7 +452,7 @@ type pinning struct { // session already pinned keeps its pin, which serverFor renewed. func (pn *pinning) settle(server string) { if pn.state == pinNew { - pn.store.Put(pn.key, pin{agent: pn.agent, server: server}, pinTTL) + pn.store.save(pn.key, pin{agent: pn.agent, server: server}) } } @@ -417,7 +472,8 @@ func (pn *pinning) forgo() { // renews the pin. Otherwise the session is decided by unpinned, and the pin comes // back pinNew for OnRequest to settle by where the request went (see the package // doc) — except when the pin is another agent's, which this request must neither -// follow nor overwrite, so it comes back pinNone. Without a session to pin, whether +// follow nor overwrite, or when the request decides nothing, and then it comes back +// pinNone. Without a session to pin, whether // none, a synthetic one or no store, the agent's choice applies and nothing is // pinned: a synthetic session's history is many conversations', not this one's. // @@ -430,7 +486,8 @@ func (p *Router) serverFor(pctx *pipeline.Context) (string, pinning) { if pctx.Session == nil || pctx.Session.ID == "" || synthetic(pctx.Session.ID) { return choice, pinning{state: pinNone} } - if pctx.Shared == nil { + store := p.pinsFor(pctx) + if store == nil { p.noStore.Do(func() { slog.Warn("inference-router: this binary wires no process store, so sessions are not pinned: " + "each request follows its agent's current server, and changing it moves running sessions") @@ -438,59 +495,138 @@ func (p *Router) serverFor(pctx *pipeline.Context) (string, pinning) { return choice, pinning{state: pinNone} } key := pinPrefix + pctx.Session.ID - if v, ok := pctx.Shared.Get(key); ok { - if pn, ok := v.(pin); ok { - if pn.agent != agent { - return p.unpinned(pctx.Session, agent, choice), pinning{state: pinNone} - } - pctx.Shared.Put(key, pn, pinTTL) - return pn.server, pinning{state: pinExisting} + if pn, ok := store.load(key); ok { + if pn.agent != agent { + server, _, _ := p.unpinned(pctx, agent, choice) + return server, pinning{state: pinNone} } + store.keep(key, pn) + return pn.server, pinning{state: pinExisting} } - return p.unpinned(pctx.Session, agent, choice), pinning{state: pinNew, store: pctx.Shared, key: key, agent: agent} + server, turn, decides := p.unpinned(pctx, agent, choice) + if !decides { + return server, pinning{state: pinNone, turn: turn} + } + return server, pinning{state: pinNew, store: store, key: key, agent: agent, turn: turn} } -// unpinned is the server for a request in a session its agent holds no pin on: the -// server the session's history shows it already uses, or for a new session — no -// earlier request to any server — the agent's current choice. An agent that is not -// routed is left alone, whatever its history. -func (p *Router) unpinned(s *pipeline.SessionView, agent, choice string) string { +// unpinned is the server for a request in a session its agent holds no pin on, and +// whether the request decides the session's pin. In order: +// +// 1. An agent that is not routed is left alone, whatever its history. +// 2. The server the session's history shows the agent already uses. +// 3. A request no parser read as inference goes to the agent's current choice and +// decides nothing. +// 4. A Claude Code request's own turn (see turnOf): a conversation's opening turn +// goes to the agent's current choice; a continuation, a conversation the router +// did not see begin, is not routed; a side request is not routed and decides +// nothing. turn names the last two, for the record. +// 5. The agent's current choice. +func (p *Router) unpinned(pctx *pipeline.Context, agent, choice string) (server, turn string, decides bool) { if choice == "" { + return "", "", true + } + if server, ok := p.wentTo(pctx.Session, agent); ok { + return server, "", true + } + if pctx.Extensions.Inference == nil { + return choice, "", false + } + switch turnOf(pctx.Extensions.Inference) { + case turnContinuation: + return "", turnContinuation, true + case turnAside: + return "", turnAside, false + } + return choice, "", true +} + +// turnOf is a Claude Code request's place in its session's conversation, read from +// inference-parser's parse: turnContinuation for the main conversation with earlier +// assistant turns, turnAside for any other request, and "" for the main +// conversation's opening turn or a request that states no role. +// +// The role is stated only by Claude Code's system prompt (see +// pipeline.InferenceExtension.AgentRole), and only Claude Code is read this way: +// another agent's earlier turns may have gone to another provider — OpenCode starts +// on Zen and addresses a server later — so its turns say nothing about whether its +// use of the servers began before this request. The main conversation carries its +// tool manifest; the one-shots Claude Code interleaves with it, its title request, +// auto mode's classifier and the permission monitor, carry none, and neither they +// nor a subagent's requests say whether the session began now. +func turnOf(ext *pipeline.InferenceExtension) string { + if ext == nil || ext.AgentRole == "" { return "" } - if server, ok := p.wentTo(s.Events, agent); ok { - return server + if ext.AgentRole != pipeline.AgentRoleMain || len(ext.Tools) == 0 { + return turnAside + } + for _, m := range ext.Messages { + if m.Role == "assistant" { + return turnContinuation + } } - return choice + return "" } -// wentTo is the server the latest earlier inference request agent sent in events -// to a server's host went to; ok is false when there is no such request, which is -// what makes a session new. +// wentTo is the server the latest earlier inference request agent sent in the +// session to a server's host went to; ok is false when there is no such request, +// which is what makes a session new. The session's events in memory are read first, +// then, where the process has a session archive, its archived events below the +// oldest one memory holds, up to historyLimit of them; an archive read that fails is +// no evidence. // // Only an outbound request row a parser read as inference counts, and only one that // reached a server. A tunnel row, a count_tokens or a /v1/models request no parser -// claimed, and a denied request, which went nowhere, say nothing about where the -// conversation is; nor does a request to any other host, which the router never -// routes (see the package doc). Another agent's row says nothing about this agent's -// conversation, and following it would hand this request that agent's server and -// key. A row's Host is where the bytes went, the server's host for a routed request, -// because the listener records it after the redirect; RequestedHost, where the -// client asked to go, is not where the session is. -func (p *Router) wentTo(events []pipeline.SessionEvent, agent string) (server string, ok bool) { +// claimed, a side request (see turnOf), which decides nothing, and a denied request, +// which went nowhere, say nothing about where the conversation is; nor does a +// request to any other host, which the router never routes (see the package doc). +// Another agent's row says nothing about this agent's conversation, and following it +// would hand this request that agent's server and key. A row's Host is where the +// bytes went, the server's host for a routed request, because the listener records +// it after the redirect; RequestedHost, where the client asked to go, is not where +// the session is. +func (p *Router) wentTo(view *pipeline.SessionView, agent string) (server string, ok bool) { + events := view.Events for i := len(events) - 1; i >= 0; i-- { - e := &events[i] - if e.Direction != pipeline.Outbound || e.Phase != pipeline.SessionRequest || e.Inference == nil || - pipeline.AgentName(e.Client.Label()) != agent { - continue - } - if server, ok := p.byHost[routerconfig.Hostname(e.Host)]; ok { + if server, ok := p.serverOf(&events[i], agent); ok { return server, true } } + if p.history == nil { + return "", false + } + var before uint64 + if len(events) > 0 { + before = events[0].Seq + } + read := 0 + err := p.history.Earlier(view.ID, before, func(e *pipeline.SessionEvent) bool { + read++ + server, ok = p.serverOf(e, agent) + return !ok && read < historyLimit + }) + if ok { + return server, true + } + if err != nil { + slog.Warn("inference-router: could not read the session's archived events; deciding it without them", + "session", view.ID, "error", err) + } return "", false } +// serverOf is the server e shows agent's session on, by the rule wentTo gives; ok is +// false when e is no evidence. +func (p *Router) serverOf(e *pipeline.SessionEvent, agent string) (server string, ok bool) { + if e.Direction != pipeline.Outbound || e.Phase != pipeline.SessionRequest || e.Inference == nil || + turnOf(e.Inference) == turnAside || pipeline.AgentName(e.Client.Label()) != agent { + return "", false + } + server, ok = p.byHost[routerconfig.Hostname(e.Host)] + return server, ok +} + // synthetic reports a session id the listener files traffic under when it knows no // one conversation: the default bucket, or an agent's pending bucket. See the // package doc for why those are never pinned. @@ -529,4 +665,6 @@ var ( _ pipeline.Plugin = (*Router)(nil) _ pipeline.Configurable = (*Router)(nil) _ pipeline.SchemaProvider = (*Router)(nil) + _ storage.StoreConsumer = (*Router)(nil) + _ session.HistoryConsumer = (*Router)(nil) ) diff --git a/core/plugins/inferencerouter/plugin_test.go b/core/plugins/inferencerouter/plugin_test.go index b4233e2a4..a790dd0a1 100644 --- a/core/plugins/inferencerouter/plugin_test.go +++ b/core/plugins/inferencerouter/plugin_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "fmt" "log/slog" "net/http" "strings" @@ -59,7 +60,7 @@ func newStore(t *testing.T) *memstore.Store { // request is a context the forward proxy would build for a decrypted request to // host from an agent with User-Agent ua, in session (none when ""), carrying the -// client's own key as a bearer token. +// client's own key as a bearer token, that inference-parser read as inference. func request(store pipeline.SharedStore, host, ua, session string) *pipeline.Context { pctx := &pipeline.Context{ Direction: pipeline.Outbound, @@ -75,6 +76,7 @@ func request(store pipeline.SharedStore, host, ua, session string) *pipeline.Con if session != "" { pctx.Session = &pipeline.SessionView{ID: session} } + pctx.Extensions.Inference = &pipeline.InferenceExtension{} pctx.ResolveClient() pctx.MarkRedirectable() return pctx @@ -222,7 +224,7 @@ func TestRouter_RoutesAListedAgentsSessionToItsServer(t *testing.T) { func TestRouter_RoutesEveryPathOnAServersHost(t *testing.T) { p := build(t, routerConfig(`"claude-code": "glm"`)) pctx := request(newStore(t), eteHost, claudeUA, "s1") - pctx.Method, pctx.Path = http.MethodGet, "/v1/models" + pctx.Method, pctx.Path, pctx.Extensions.Inference = http.MethodGet, "/v1/models", nil run(t, p, pctx) assertRouted(t, pctx, glmHost, "glm-key") @@ -565,6 +567,15 @@ func sent(host, ua string) pipeline.SessionEvent { } } +// recorded is the request row the forward proxy records for pctx once the pipeline +// has run: where it went, and the parse it carried. +func recorded(pctx *pipeline.Context) pipeline.SessionEvent { + return pipeline.SessionEvent{ + Direction: pipeline.Outbound, Phase: pipeline.SessionRequest, Host: pctx.Host, + Inference: pctx.Extensions.Inference, Client: pipeline.ParseUserAgent(pctx.Headers.Get("User-Agent")), + } +} + // withHistory gives pctx's session the events the store recorded before it. func withHistory(pctx *pipeline.Context, events ...pipeline.SessionEvent) *pipeline.Context { pctx.Session.Events = events @@ -748,3 +759,152 @@ func TestRouter_APinHoldsOnlyForTheAgentItWasMadeFor(t *testing.T) { assertRouted(t, owner, glmHost, "glm-key") assertRecord(t, owner, pipeline.ActionModify, "routed", map[string]string{"server": "glm", "pin": pinExisting}) } + +// claudeCode gives pctx inference-parser's reading of a Claude Code request: role is +// the caller its system prompt declares, tools how many tools it carries, and turns +// how many assistant turns came earlier in its conversation. +func claudeCode(pctx *pipeline.Context, role pipeline.AgentRole, tools, turns int) *pipeline.Context { + ext := &pipeline.InferenceExtension{Model: "claude-opus-5-5", AgentRole: role, Messages: []pipeline.InferenceMessage{ + {Role: "system", Content: "x-anthropic-billing-header: cc_version=2.1.286; cc_entrypoint=cli;"}, + {Role: "user", Content: "hello"}, + }} + for range turns { + ext.Messages = append(ext.Messages, + pipeline.InferenceMessage{Role: "assistant", Content: "Hi."}, + pipeline.InferenceMessage{Role: "user", Content: "go on"}) + } + for i := range tools { + ext.Tools = append(ext.Tools, pipeline.InferenceTool{Name: fmt.Sprintf("Tool%d", i)}) + } + pctx.Extensions.Inference = ext + return pctx +} + +// The live case behind this rule: a Claude Code conversation running on ete before +// the router knew it — started before routing was set up, or forgotten by a restart — +// was taken for a new session, sent to glm and held there, 298 assistant turns in. +// Its request carries those turns, so the router knows it did not see the +// conversation begin, and leaves it where Claude Code sent it. +func TestRouter_AClaudeCodeConversationItDidNotSeeBeginIsNotRouted(t *testing.T) { + store := newStore(t) + p := build(t, routerConfig(`"claude-code": "glm"`)) + pctx := claudeCode(request(store, eteHost, claudeUA, "running"), pipeline.AgentRoleMain, 26, 3) + run(t, p, pctx) + + assertUntouched(t, pctx, eteHost) + assertRecord(t, pctx, pipeline.ActionSkip, "not_routed", map[string]string{"pin": pinNew, "turn": turnContinuation}) + if pin, pinned := pinOf(t, store, "running"); !pinned || pin != "" { + t.Errorf("pin = %q (pinned %v), want the session pinned as not routed", pin, pinned) + } + + // Pinned now: the session stays unrouted whatever its later requests look like. + next := claudeCode(request(store, eteHost, claudeUA, "running"), pipeline.AgentRoleMain, 26, 4) + run(t, p, next) + assertUntouched(t, next, eteHost) + assertRecord(t, next, pipeline.ActionSkip, "not_routed", map[string]string{"pin": pinExisting}) +} + +// A conversation's opening turn is the beginning the router assigns a server at. +func TestRouter_AClaudeCodeOpeningTurnGoesToItsAgentsServer(t *testing.T) { + store := newStore(t) + p := build(t, routerConfig(`"claude-code": "glm"`)) + pctx := claudeCode(request(store, eteHost, claudeUA, "s1"), pipeline.AgentRoleMain, 26, 0) + run(t, p, pctx) + + assertRouted(t, pctx, glmHost, "glm-key") + assertRecord(t, pctx, pipeline.ActionModify, "routed", map[string]string{"server": "glm", "pin": pinNew}) + if pin, _ := pinOf(t, store, "s1"); pin != "glm" { + t.Errorf("pin = %q, want glm", pin) + } +} + +// Claude Code's one-shots — the title request, auto mode's classifier, the +// permission monitor — carry no tools and come at any point in a conversation, and +// a subagent's requests are its own conversation's, not the session's. None says +// whether the session began now, so none decides: it goes where Claude Code sent +// it, and the session stays unpinned for its conversation's next request. +func TestRouter_AClaudeCodeSideRequestDecidesNothing(t *testing.T) { + for _, tc := range []struct { + name string + role pipeline.AgentRole + tools, turns int + }{ + {"a one-shot with no tools", pipeline.AgentRoleMain, 0, 0}, + {"a subagent's first turn", pipeline.AgentRoleSubagent, 11, 0}, + {"a subagent mid-task", pipeline.AgentRoleSubagent, 11, 4}, + } { + t.Run(tc.name, func(t *testing.T) { + store := newStore(t) + p := build(t, routerConfig(`"claude-code": "glm"`)) + pctx := claudeCode(request(store, eteHost, claudeUA, "s1"), tc.role, tc.tools, tc.turns) + run(t, p, pctx) + + assertUntouched(t, pctx, eteHost) + assertRecord(t, pctx, pipeline.ActionSkip, "not_routed", map[string]string{"pin": pinNone, "turn": turnAside}) + if _, pinned := pinOf(t, store, "s1"); pinned { + t.Error("a side request pinned the session") + } + }) + } +} + +// After a side request, the conversation's own next request decides: an opening turn +// starts the session on its agent's server, a continuation leaves it unrouted. +func TestRouter_AfterASideRequestTheConversationDecides(t *testing.T) { + for _, tc := range []struct { + name string + turns int + host string // where the conversation's request goes + pin string + }{ + {"an opening turn", 0, glmHost, "glm"}, + {"a continuation", 2, eteHost, ""}, + } { + t.Run(tc.name, func(t *testing.T) { + store := newStore(t) + p := build(t, routerConfig(`"claude-code": "glm"`)) + side := claudeCode(request(store, eteHost, claudeUA, "s1"), pipeline.AgentRoleMain, 0, 0) + run(t, p, side) + // The forward proxy records the side request's row, which the next request sees. + pctx := withHistory(claudeCode(request(store, eteHost, claudeUA, "s1"), pipeline.AgentRoleMain, 26, tc.turns), + recorded(side)) + run(t, p, pctx) + + if pctx.Host != tc.host { + t.Errorf("Host = %q, want %q", pctx.Host, tc.host) + } + if pin, pinned := pinOf(t, store, "s1"); !pinned || pin != tc.pin { + t.Errorf("pin = %q (pinned %v), want %q", pin, pinned, tc.pin) + } + }) + } +} + +// The request's turn is read only where the request states Claude Code's role. Any +// other agent's earlier turns may have gone to another provider — OpenCode starts on +// Zen and addresses a server later — so its first request to a server's host is +// where its use of the servers begins, and it goes to its agent's server as before. +func TestRouter_ARequestThatStatesNoRoleIsDecidedAsBefore(t *testing.T) { + store := newStore(t) + p := build(t, routerConfig(`"opencode": "glm"`)) + pctx := claudeCode(request(store, eteHost, opencodeUA, "s1"), "", 12, 5) + run(t, p, pctx) + + assertRouted(t, pctx, glmHost, "glm-key") + assertRecord(t, pctx, pipeline.ActionModify, "routed", map[string]string{"server": "glm", "pin": pinNew}) +} + +// Where the session's recorded history says which server it is on, that is evidence +// and outranks the request: the continuation follows its history to glm. +func TestRouter_RecordedHistoryOutranksTheRequestsTurn(t *testing.T) { + store := newStore(t) + p := build(t, routerConfig(`"claude-code": "glm"`)) + pctx := withHistory(claudeCode(request(store, eteHost, claudeUA, "running"), pipeline.AgentRoleMain, 26, 3), + sent(glmHost, claudeUA)) + run(t, p, pctx) + + assertRouted(t, pctx, glmHost, "glm-key") + if pin, _ := pinOf(t, store, "running"); pin != "glm" { + t.Errorf("pin = %q, want glm", pin) + } +} diff --git a/core/plugins/inferencerouter/reload_test.go b/core/plugins/inferencerouter/reload_test.go index d64e16977..e4b10844e 100644 --- a/core/plugins/inferencerouter/reload_test.go +++ b/core/plugins/inferencerouter/reload_test.go @@ -9,9 +9,10 @@ import ( "github.com/rossoctl/cortex/core/plugins" ) -// entries is the outbound chain a config file with only the router would build. +// entries is the outbound chain a config file with inference-parser and the router +// would build. func entries(routerConfig string) []config.PluginEntry { - return []config.PluginEntry{{Name: Name, Config: json.RawMessage(routerConfig)}} + return []config.PluginEntry{{Name: "inference-parser"}, {Name: Name, Config: json.RawMessage(routerConfig)}} } // reloadBuild builds the chain the way cmd/cortex does on startup and on every diff --git a/core/plugins/inferencerouter/store_test.go b/core/plugins/inferencerouter/store_test.go new file mode 100644 index 000000000..54599de77 --- /dev/null +++ b/core/plugins/inferencerouter/store_test.go @@ -0,0 +1,133 @@ +package inferencerouter + +import ( + "context" + "encoding/json" + "path/filepath" + "testing" + "time" + + "github.com/rossoctl/cortex/core/pipeline" + "github.com/rossoctl/cortex/core/storage" + "github.com/rossoctl/cortex/core/storage/filestore" +) + +// openStore opens a plugin store at path, as cmd/cortex does on a local install. +func openStore(t *testing.T, path string) *filestore.Store { + t.Helper() + st, err := filestore.Open(path) + if err != nil { + t.Fatalf("filestore.Open: %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + return st +} + +// buildWithStore is build with st injected before Configure, as plugins.BuildWithDeps +// does, and the router's clock set to now when it is non-nil. +func buildWithStore(t *testing.T, config string, st storage.Store, now func() time.Time) *pipeline.Pipeline { + t.Helper() + r := New() + r.SetStore(st) + if now != nil { + r.now = now + } + if err := r.Configure(json.RawMessage(config)); err != nil { + t.Fatalf("Configure: %v", err) + } + p, err := pipeline.New([]pipeline.Plugin{r}) + if err != nil { + t.Fatalf("pipeline.New: %v", err) + } + return p +} + +// storedPin is the record the store holds for session, and whether there is one. +func storedPin(t *testing.T, st storage.Store, session string) (pinRecord, bool) { + t.Helper() + raw, err := st.Get(context.Background(), pinPrefix+session) + if err != nil { + t.Fatalf("store Get: %v", err) + } + if raw == "" { + return pinRecord{}, false + } + var rec pinRecord + if err := json.Unmarshal([]byte(raw), &rec); err != nil { + t.Fatalf("stored pin %q does not decode: %v", raw, err) + } + return rec, true +} + +// The live case: a session started on glm, the proxy restarted, and its next turn — +// a continuation, which the router would leave unrouted — went to ete and failed. With +// a store, the pin outlives the process, and the restarted router keeps the session +// on glm. +func TestRouter_WithAStoreAPinSurvivesARestart(t *testing.T) { + path := filepath.Join(t.TempDir(), "plugin-state.json") + before := openStore(t, path) + shared := newStore(t) + p := buildWithStore(t, routerConfig(`"claude-code": "glm"`), before, nil) + run(t, p, claudeCode(request(shared, eteHost, claudeUA, "s1"), pipeline.AgentRoleMain, 26, 0)) + if _, inMemory := pinOf(t, shared, "s1"); inMemory { + t.Error("the pin was kept in memory as well as in the store") + } + if err := before.Close(); err != nil { + t.Fatal(err) + } + + // A new process: a fresh router, a fresh process store, and the store reopened. + after := openStore(t, path) + restarted := buildWithStore(t, routerConfig(`"claude-code": "glm"`), after, nil) + pctx := claudeCode(request(newStore(t), eteHost, claudeUA, "s1"), pipeline.AgentRoleMain, 26, 3) + run(t, restarted, pctx) + + assertRouted(t, pctx, glmHost, "glm-key") + assertRecord(t, pctx, pipeline.ActionModify, "routed", map[string]string{"server": "glm", "pin": pinExisting}) +} + +// Every request renews its session's pin, and a store write per request would rewrite +// the file every interval while any session is busy. The stored pin is renewed once +// it is a day old: its TTL then runs between 29 and 30 days from the session's last +// request, rather than exactly 30. +func TestRouter_AStoredPinIsRenewedAtMostOnceADay(t *testing.T) { + st := openStore(t, filepath.Join(t.TempDir(), "plugin-state.json")) + now := time.Date(2026, 10, 8, 17, 0, 0, 0, time.UTC) + p := buildWithStore(t, routerConfig(`"claude-code": "glm"`), st, func() time.Time { return now }) + send := func() { + run(t, p, claudeCode(request(newStore(t), eteHost, claudeUA, "s1"), pipeline.AgentRoleMain, 26, 0)) + } + + send() + first, ok := storedPin(t, st, "s1") + if !ok || first.Agent != "claude-code" || first.Server != "glm" || first.Renewed != now.Unix() { + t.Fatalf("stored pin = %+v (found %v), want claude-code on glm renewed now", first, ok) + } + now = now.Add(23 * time.Hour) + send() + if rec, _ := storedPin(t, st, "s1"); rec.Renewed != first.Renewed { + t.Errorf("renewed at %d within a day of %d, want it left alone", rec.Renewed, first.Renewed) + } + now = now.Add(2 * time.Hour) + send() + if rec, _ := storedPin(t, st, "s1"); rec.Renewed != now.Unix() { + t.Errorf("renewed = %d past a day, want %d", rec.Renewed, now.Unix()) + } +} + +// A stored value that is not a pin record says nothing about the session, which is +// decided as an unpinned one and pinned over it. +func TestRouter_AStoredValueThatIsNoPinIsIgnored(t *testing.T) { + st := openStore(t, filepath.Join(t.TempDir(), "plugin-state.json")) + if err := st.Set(context.Background(), pinPrefix+"s1", "not a pin", 0); err != nil { + t.Fatal(err) + } + p := buildWithStore(t, routerConfig(`"claude-code": "glm"`), st, nil) + pctx := claudeCode(request(newStore(t), eteHost, claudeUA, "s1"), pipeline.AgentRoleMain, 26, 0) + run(t, p, pctx) + + assertRouted(t, pctx, glmHost, "glm-key") + if rec, ok := storedPin(t, st, "s1"); !ok || rec.Server != "glm" { + t.Errorf("stored pin = %+v (found %v), want glm", rec, ok) + } +} diff --git a/core/session/archive/read.go b/core/session/archive/read.go index ff0d799da..56454f5d6 100644 --- a/core/session/archive/read.go +++ b/core/session/archive/read.go @@ -177,3 +177,27 @@ func (l *lastN) items() []*pipeline.SessionEvent { } return append(slices.Clone(l.buf[l.next:]), l.buf[:l.next]...) } + +var _ session.History = (*Archive)(nil) + +// earlierPage is how many events Earlier asks Page for at a time, which bounds the events one +// read holds. +var earlierPage = 500 + +// Earlier implements session.History: Page after Page, each below the oldest event the last +// one returned, until fn returns false or the session's oldest archived event. +func (a *Archive) Earlier(id string, before uint64, fn func(*pipeline.SessionEvent) bool) error { + for { + n, last, stopped := 0, uint64(0), false + _, ok, err := a.Page(id, before, earlierPage, func(e *pipeline.SessionEvent) bool { + n, last = n+1, e.Seq + stopped = !fn(e) + return !stopped + }) + // last <= 1: nothing is numbered below it, and a before of 0 would start over. + if err != nil || !ok || stopped || n < earlierPage || last <= 1 { + return err + } + before = last + } +} diff --git a/core/session/archive/read_test.go b/core/session/archive/read_test.go index acab97e32..85245ad35 100644 --- a/core/session/archive/read_test.go +++ b/core/session/archive/read_test.go @@ -171,3 +171,37 @@ func equalSeqs(a, b []uint64) bool { } return true } + +func earlier(t *testing.T, a *Archive, id string, before uint64, stopAt uint64) []uint64 { + t.Helper() + var seqs []uint64 + if err := a.Earlier(id, before, func(e *pipeline.SessionEvent) bool { + seqs = append(seqs, e.Seq) + return e.Seq != stopAt + }); err != nil { + t.Fatal(err) + } + return seqs +} + +// Earlier reads a session's archived events newest first, a page at a time across segments, +// from below before to the oldest, and stops when fn says so. +func TestEarlier_ReadsNewestFirstPageByPageUntilToldToStop(t *testing.T) { + defer func(n int) { earlierPage = n }(earlierPage) + earlierPage = 5 + a, _ := twoSegments(t) + tick(a) + + if seqs := earlier(t, a, "s1", 0, 0); !equalSeqs(seqs, []uint64{12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1}) { + t.Errorf("all: %v, want 12 down to 1", seqs) + } + if seqs := earlier(t, a, "s1", 8, 0); !equalSeqs(seqs, []uint64{7, 6, 5, 4, 3, 2, 1}) { + t.Errorf("before=8: %v, want 7 down to 1", seqs) + } + if seqs := earlier(t, a, "s1", 0, 4); !equalSeqs(seqs, []uint64{12, 11, 10, 9, 8, 7, 6, 5, 4}) { + t.Errorf("stopped at 4: %v, want 12 down to 4", seqs) + } + if seqs := earlier(t, a, "nope", 0, 0); len(seqs) != 0 { + t.Errorf("unknown session: %v, want nothing", seqs) + } +} diff --git a/core/session/history.go b/core/session/history.go index 20acf2a1b..ea517fc10 100644 --- a/core/session/history.go +++ b/core/session/history.go @@ -114,6 +114,22 @@ type PriorKeeper interface { Prior(sessionID string, after uint64) (Prior, bool) } +// History is the events a session recorded on disk, the session archive's, for a plugin that +// needs the events from before this process: a store entry starts empty after a restart. +type History interface { + // Earlier calls fn, newest first, for the events recorded under sessionID numbered below + // before, every one when before is 0, until fn returns false or none are left. It reads + // from disk. + Earlier(sessionID string, before uint64, fn func(*pipeline.SessionEvent) bool) error +} + +// HistoryConsumer is implemented by plugins that read a session's archived events. +// plugins.BuildWithDeps injects the process's History before Configure runs; a process with no +// session archive passes nothing. +type HistoryConsumer interface { + SetHistory(History) +} + // priorLocked is the first PriorKeeper's answer for id's entry numbered after `after`. s.mu held. func (s *Store) priorLocked(id string, after uint64) (Prior, bool) { for _, r := range s.recorders { diff --git a/core/session/store.go b/core/session/store.go index 3bdbb009c..4bdb5e4b9 100644 --- a/core/session/store.go +++ b/core/session/store.go @@ -1006,12 +1006,12 @@ type SessionSummary struct { // request since. Absent otherwise, and on a row only the archive holds (Resident false), all of // whose figures are history. // - // WHAT A CLIENT NEEDS TO COUNT WHERE RUNNING SESSIONS STAY. The inference-router pins a session - // in process memory by the first request it routes, so after a restart nothing holds such a - // session where its history says it was: its next request is decided as a new one's, and goes - // to its agent's current server. After an eviction alone the pin can survive, so a count that - // leaves these rows out errs low there, which is the quiet side. InferenceHost itself is - // unchanged by this: it still says where the session's inference last went. + // WHAT IT TELLS A CLIENT: the host is the archive's word, from before this process, and not a + // request this process saw. Whether something still holds the session there is the router's + // business: on a local install it keeps its pins in the plugin store, which a restart keeps, and + // agentop counts such a session as staying; a process with no plugin store keeps them in memory + // and forgets them on restart. InferenceHost itself is unchanged by this: it still says where + // the session's inference last went. InferenceHostFromHistory bool `json:"inferenceHostFromHistory,omitempty"` // Resident is false on a row the session archive served because the store no longer holds diff --git a/core/storage/consumer.go b/core/storage/consumer.go new file mode 100644 index 000000000..3e9b43518 --- /dev/null +++ b/core/storage/consumer.go @@ -0,0 +1,12 @@ +package storage + +// StoreConsumer is implemented by plugins that keep state which must outlive the +// process. plugins.BuildWithDeps injects the process Store before Configure runs, +// as it injects spiffe.ProviderConsumer's and pricing.ResolverConsumer's. +// +// A plugin must tolerate never being called: a process with no store to offer — any +// that is not a local install — passes nothing, and the plugin then keeps its state +// in memory as before. +type StoreConsumer interface { + SetStore(Store) +} diff --git a/core/storage/filestore/store.go b/core/storage/filestore/store.go new file mode 100644 index 000000000..3e35ca194 --- /dev/null +++ b/core/storage/filestore/store.go @@ -0,0 +1,393 @@ +// Package filestore is a storage.Store kept in memory and saved to one file, for a +// single process that needs its plugins' state to outlive it: the laptop proxy, +// where a restart is routine and there is no redis. +// +// It answers like the redis driver: a missing key reads as "", Set replaces a key's +// TTL and Incr keeps it, Expire with a TTL of zero or less deletes the key, and a +// key used as the other type — a string as a hash, or a hash as a string — is an +// error. +// +// Every change is held in memory and saved within one flush interval, and Close +// saves what is left, so a process that dies without closing the store loses at +// most that interval. The file is replaced whole, by rename, so it is never read +// half-written. One that does not parse is set aside as .corrupt and the store +// starts empty: what it holds is worth keeping, not worth failing a start over. +// +// It is not shared between processes: two opening one file overwrite each other's +// saves. +package filestore + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "time" + + "github.com/rossoctl/cortex/core/storage" +) + +func init() { + storage.Register("file", func(url string) (storage.Store, error) { + return Open(strings.TrimPrefix(url, "file://")) + }) +} + +var _ storage.Store = (*Store)(nil) + +// flushEvery is how long a change can wait in memory before it is saved. +const flushEvery = 2 * time.Second + +var ( + errClosed = errors.New("filestore: store is closed") + errWrongType = errors.New("filestore: key holds the wrong type of value") +) + +// entry is one key: a hash when Hash is non-nil, else the string Value. A hash +// always holds a field, as in redis, so a nil and an empty one never need telling +// apart. +type entry struct { + Value string `json:"v,omitempty"` + Hash map[string]string `json:"h,omitempty"` + // Expires is the Unix time in nanoseconds the key expires at, 0 for never. + Expires int64 `json:"exp,omitempty"` +} + +// snapshot is the file's format. +type snapshot struct { + Version int `json:"version"` + Entries map[string]*entry `json:"entries"` +} + +// Store is the file-backed storage.Store. +type Store struct { + path string + now func() time.Time + every time.Duration + + mu sync.Mutex + entries map[string]*entry + dirty bool + closed bool + + stop chan struct{} + done chan struct{} +} + +type option func(*Store) + +func withClock(now func() time.Time) option { return func(s *Store) { s.now = now } } +func withFlushEvery(d time.Duration) option { return func(s *Store) { s.every = d } } + +// Open loads the store saved at path, or starts an empty one when there is none, +// creating path's directory if it is missing. +func Open(path string) (*Store, error) { return openWith(path) } + +func openWith(path string, opts ...option) (*Store, error) { + s := &Store{path: path, now: time.Now, every: flushEvery, entries: make(map[string]*entry), + stop: make(chan struct{}), done: make(chan struct{})} + for _, o := range opts { + o(s) + } + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return nil, fmt.Errorf("filestore: %w", err) + } + if err := s.load(); err != nil { + return nil, err + } + go s.flushLoop() + return s, nil +} + +func (s *Store) load() error { + b, err := os.ReadFile(s.path) //nolint:gosec // the path is the operator's + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return fmt.Errorf("filestore: %w", err) + } + var snap snapshot + if err := json.Unmarshal(b, &snap); err != nil { + aside := s.path + ".corrupt" + slog.Warn("filestore: the saved store does not parse, so it is set aside and the store starts empty", + "path", s.path, "setAside", aside, "error", err) + if rerr := os.Rename(s.path, aside); rerr != nil { + return fmt.Errorf("filestore: setting aside %s: %w", s.path, rerr) + } + return nil + } + now := s.now().UnixNano() + for k, e := range snap.Entries { + if e != nil && (e.Expires == 0 || e.Expires > now) { + s.entries[k] = e + } + } + return nil +} + +// live is key's entry, or nil when there is none or it has expired. An expired +// entry is deleted. Called with mu held. +func (s *Store) live(key string) *entry { + e, ok := s.entries[key] + if !ok { + return nil + } + if e.Expires != 0 && e.Expires <= s.now().UnixNano() { + delete(s.entries, key) + s.dirty = true + return nil + } + return e +} + +// expiry is the Expires value for a TTL from now, 0 for none. +func (s *Store) expiry(ttl time.Duration) int64 { + if ttl <= 0 { + return 0 + } + return s.now().Add(ttl).UnixNano() +} + +// locked runs f with mu held, after refusing a closed store. +func (s *Store) locked(f func() error) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return errClosed + } + return f() +} + +func (s *Store) Get(_ context.Context, key string) (string, error) { + var v string + err := s.locked(func() error { + e := s.live(key) + if e == nil { + return nil + } + if e.Hash != nil { + return errWrongType + } + v = e.Value + return nil + }) + return v, err +} + +func (s *Store) Set(_ context.Context, key, value string, ttl time.Duration) error { + return s.locked(func() error { + s.entries[key] = &entry{Value: value, Expires: s.expiry(ttl)} + s.dirty = true + return nil + }) +} + +func (s *Store) Incr(_ context.Context, key string, delta int64) (int64, error) { + var n int64 + err := s.locked(func() error { + e := s.live(key) + if e == nil { + e = &entry{Value: "0"} + s.entries[key] = e + } + if e.Hash != nil { + return errWrongType + } + cur, err := strconv.ParseInt(e.Value, 10, 64) + if err != nil { + return fmt.Errorf("filestore: value is not an integer: %w", err) + } + n = cur + delta + e.Value = strconv.FormatInt(n, 10) + s.dirty = true + return nil + }) + return n, err +} + +// hash is key's hash for writing, created when the key is missing. Called with mu +// held. +func (s *Store) hash(key string) (*entry, error) { + e := s.live(key) + if e == nil { + e = &entry{Hash: make(map[string]string, 1)} + s.entries[key] = e + return e, nil + } + if e.Hash == nil { + return nil, errWrongType + } + return e, nil +} + +func (s *Store) HashIncr(_ context.Context, key, field string, delta int64) (int64, error) { + var n int64 + err := s.locked(func() error { + e, err := s.hash(key) + if err != nil { + return err + } + var cur int64 + if v, ok := e.Hash[field]; ok { + if cur, err = strconv.ParseInt(v, 10, 64); err != nil { + return fmt.Errorf("filestore: hash field is not an integer: %w", err) + } + } + n = cur + delta + e.Hash[field] = strconv.FormatInt(n, 10) + s.dirty = true + return nil + }) + return n, err +} + +func (s *Store) HashGet(_ context.Context, key string) (map[string]string, error) { + out := map[string]string{} + err := s.locked(func() error { + e := s.live(key) + if e == nil { + return nil + } + if e.Hash == nil { + return errWrongType + } + for k, v := range e.Hash { + out[k] = v + } + return nil + }) + return out, err +} + +func (s *Store) HashSetNX(_ context.Context, key, field, value string) (bool, error) { + var set bool + err := s.locked(func() error { + e, err := s.hash(key) + if err != nil { + return err + } + if _, ok := e.Hash[field]; ok { + return nil + } + e.Hash[field] = value + s.dirty, set = true, true + return nil + }) + return set, err +} + +func (s *Store) Expire(_ context.Context, key string, ttl time.Duration) error { + return s.locked(func() error { + e := s.live(key) + if e == nil { + return nil + } + if ttl <= 0 { + delete(s.entries, key) + } else { + e.Expires = s.expiry(ttl) + } + s.dirty = true + return nil + }) +} + +// Close saves what the store holds and stops its saving. Operations after it fail; +// a second Close does nothing. +func (s *Store) Close() error { + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return nil + } + s.closed = true + s.mu.Unlock() + close(s.stop) + <-s.done + return s.flush() +} + +func (s *Store) flushLoop() { + defer close(s.done) + tick := time.NewTicker(s.every) + defer tick.Stop() + for { + select { + case <-s.stop: + return + case <-tick.C: + if err := s.flush(); err != nil { + slog.Warn("filestore: saving the store failed; the next interval retries", "path", s.path, "error", err) + } + } + } +} + +// flush saves the store when it has changed since the last save. The file is +// written beside its final name and renamed over it, so a reader sees the old file +// or the new one, never part of one. +func (s *Store) flush() error { + s.mu.Lock() + if !s.dirty { + s.mu.Unlock() + return nil + } + now := s.now().UnixNano() + snap := snapshot{Version: 1, Entries: make(map[string]*entry, len(s.entries))} + for k, e := range s.entries { + if e.Expires == 0 || e.Expires > now { + c := *e + if e.Hash != nil { + c.Hash = make(map[string]string, len(e.Hash)) + for f, v := range e.Hash { + c.Hash[f] = v + } + } + snap.Entries[k] = &c + } + } + s.dirty = false + s.mu.Unlock() + + if err := s.write(snap); err != nil { + s.mu.Lock() + s.dirty = true + s.mu.Unlock() + return err + } + return nil +} + +func (s *Store) write(snap snapshot) error { + b, err := json.Marshal(snap) + if err != nil { + return fmt.Errorf("filestore: %w", err) + } + tmp := s.path + ".tmp" + f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600) //nolint:gosec // the path is the operator's + if err != nil { + return fmt.Errorf("filestore: %w", err) + } + if _, err := f.Write(b); err != nil { + _ = f.Close() + return fmt.Errorf("filestore: %w", err) + } + if err := f.Sync(); err != nil { + _ = f.Close() + return fmt.Errorf("filestore: %w", err) + } + if err := f.Close(); err != nil { + return fmt.Errorf("filestore: %w", err) + } + if err := os.Rename(tmp, s.path); err != nil { + return fmt.Errorf("filestore: %w", err) + } + return nil +} diff --git a/core/storage/filestore/store_test.go b/core/storage/filestore/store_test.go new file mode 100644 index 000000000..5d738f241 --- /dev/null +++ b/core/storage/filestore/store_test.go @@ -0,0 +1,299 @@ +package filestore + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/rossoctl/cortex/core/storage" +) + +var ctx = context.Background() + +// clock is a settable time source. +type clock struct{ t time.Time } + +func (c *clock) now() time.Time { return c.t } +func (c *clock) advance(d time.Duration) { c.t = c.t.Add(d) } + +// openT opens a store at path on clk, flushing only on Close unless the test sets an +// interval, and closes it when the test ends. +func openT(t *testing.T, path string, clk *clock, opts ...option) *Store { + t.Helper() + opts = append([]option{withClock(clk.now), withFlushEvery(time.Hour)}, opts...) + s, err := openWith(path, opts...) + if err != nil { + t.Fatalf("Open: %v", err) + } + t.Cleanup(func() { _ = s.Close() }) + return s +} + +func newClock() *clock { return &clock{t: time.Date(2026, 10, 8, 17, 0, 0, 0, time.UTC)} } + +func get(t *testing.T, s *Store, key string) string { + t.Helper() + v, err := s.Get(ctx, key) + if err != nil { + t.Fatalf("Get(%q): %v", key, err) + } + return v +} + +func TestStore_AMissingKeyReadsEmpty(t *testing.T) { + s := openT(t, filepath.Join(t.TempDir(), "s.json"), newClock()) + if v := get(t, s, "nope"); v != "" { + t.Errorf("Get = %q, want empty", v) + } + if h, err := s.HashGet(ctx, "nope"); err != nil || len(h) != 0 { + t.Errorf("HashGet = %v, %v; want an empty map", h, err) + } +} + +func TestStore_ATTLExpiresTheKeyAndSetReplacesIt(t *testing.T) { + clk := newClock() + s := openT(t, filepath.Join(t.TempDir(), "s.json"), clk) + if err := s.Set(ctx, "k", "v1", time.Minute); err != nil { + t.Fatal(err) + } + clk.advance(59 * time.Second) + if v := get(t, s, "k"); v != "v1" { + t.Fatalf("Get before expiry = %q, want v1", v) + } + // Set with no TTL replaces the value and the expiry, as redis SET does. + if err := s.Set(ctx, "k", "v2", 0); err != nil { + t.Fatal(err) + } + clk.advance(time.Hour) + if v := get(t, s, "k"); v != "v2" { + t.Fatalf("Get = %q, want v2 kept with no expiry", v) + } + if err := s.Set(ctx, "k", "v3", time.Minute); err != nil { + t.Fatal(err) + } + clk.advance(time.Minute) + if v := get(t, s, "k"); v != "" { + t.Errorf("Get at expiry = %q, want the key gone", v) + } +} + +func TestStore_ExpireSetsATTLDeletesAtZeroAndIgnoresAMissingKey(t *testing.T) { + clk := newClock() + s := openT(t, filepath.Join(t.TempDir(), "s.json"), clk) + _ = s.Set(ctx, "k", "v", 0) + if err := s.Expire(ctx, "k", time.Minute); err != nil { + t.Fatal(err) + } + clk.advance(time.Minute) + if v := get(t, s, "k"); v != "" { + t.Errorf("Get after Expire's TTL = %q, want the key gone", v) + } + _ = s.Set(ctx, "k", "v", 0) + if err := s.Expire(ctx, "k", 0); err != nil { + t.Fatal(err) + } + if v := get(t, s, "k"); v != "" { + t.Errorf("Get after Expire(0) = %q, want the key deleted", v) + } + if err := s.Expire(ctx, "missing", time.Minute); err != nil { + t.Errorf("Expire on a missing key = %v, want nil", err) + } + if v := get(t, s, "missing"); v != "" { + t.Errorf("Expire created %q", v) + } +} + +func TestStore_IncrCountsFromZeroAndKeepsTheTTL(t *testing.T) { + clk := newClock() + s := openT(t, filepath.Join(t.TempDir(), "s.json"), clk) + if n, err := s.Incr(ctx, "n", 3); err != nil || n != 3 { + t.Fatalf("Incr = %d, %v; want 3", n, err) + } + _ = s.Expire(ctx, "n", time.Minute) + if n, err := s.Incr(ctx, "n", -1); err != nil || n != 2 { + t.Fatalf("Incr = %d, %v; want 2", n, err) + } + clk.advance(time.Minute) + if v := get(t, s, "n"); v != "" { + t.Errorf("Get = %q, want the TTL Expire set kept by Incr", v) + } + _ = s.Set(ctx, "word", "abc", 0) + if _, err := s.Incr(ctx, "word", 1); err == nil { + t.Error("Incr on a non-integer = nil error, want one") + } +} + +func TestStore_HashOperations(t *testing.T) { + s := openT(t, filepath.Join(t.TempDir(), "s.json"), newClock()) + if n, err := s.HashIncr(ctx, "h", "a", 2); err != nil || n != 2 { + t.Fatalf("HashIncr = %d, %v; want 2", n, err) + } + if n, err := s.HashIncr(ctx, "h", "a", 5); err != nil || n != 7 { + t.Fatalf("HashIncr = %d, %v; want 7", n, err) + } + if set, err := s.HashSetNX(ctx, "h", "b", "x"); err != nil || !set { + t.Fatalf("HashSetNX new field = %v, %v; want true", set, err) + } + if set, err := s.HashSetNX(ctx, "h", "b", "y"); err != nil || set { + t.Fatalf("HashSetNX existing field = %v, %v; want false", set, err) + } + h, err := s.HashGet(ctx, "h") + if err != nil || h["a"] != "7" || h["b"] != "x" || len(h) != 2 { + t.Fatalf("HashGet = %v, %v", h, err) + } + h["a"] = "changed" + if again, _ := s.HashGet(ctx, "h"); again["a"] != "7" { + t.Error("HashGet returned the store's own map") + } +} + +// A key holds a string or a hash, and using it as the other is an error, as redis +// answers WRONGTYPE. +func TestStore_UsingAKeyAsTheOtherTypeIsAnError(t *testing.T) { + s := openT(t, filepath.Join(t.TempDir(), "s.json"), newClock()) + _ = s.Set(ctx, "str", "v", 0) + _, _ = s.HashIncr(ctx, "hash", "f", 1) + if _, err := s.Get(ctx, "hash"); err == nil { + t.Error("Get on a hash = nil error") + } + if _, err := s.Incr(ctx, "hash", 1); err == nil { + t.Error("Incr on a hash = nil error") + } + if _, err := s.HashGet(ctx, "str"); err == nil { + t.Error("HashGet on a string = nil error") + } + if _, err := s.HashIncr(ctx, "str", "f", 1); err == nil { + t.Error("HashIncr on a string = nil error") + } + if _, err := s.HashSetNX(ctx, "str", "f", "v"); err == nil { + t.Error("HashSetNX on a string = nil error") + } +} + +// The point of the store: what it holds survives the process. A key keeps its TTL +// across the reopen, and one that expired while the store was closed is gone. +func TestStore_ReopeningRestoresWhatCloseSaved(t *testing.T) { + path := filepath.Join(t.TempDir(), "s.json") + clk := newClock() + s, err := openWith(path, withClock(clk.now), withFlushEvery(time.Hour)) + if err != nil { + t.Fatal(err) + } + _ = s.Set(ctx, "kept", "v", 0) + _ = s.Set(ctx, "short", "v", time.Minute) + _ = s.Set(ctx, "long", "v", time.Hour) + _, _ = s.HashIncr(ctx, "h", "f", 4) + if err := s.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + clk.advance(2 * time.Minute) + r := openT(t, path, clk) + for key, want := range map[string]string{"kept": "v", "short": "", "long": "v"} { + if v := get(t, r, key); v != want { + t.Errorf("after reopen Get(%q) = %q, want %q", key, v, want) + } + } + if h, _ := r.HashGet(ctx, "h"); h["f"] != "4" { + t.Errorf("after reopen HashGet = %v, want f=4", h) + } + clk.advance(time.Hour) + if v := get(t, r, "long"); v != "" { + t.Errorf("Get(long) = %q, want its TTL kept across the reopen", v) + } +} + +// A change is saved while the store is open, not only at Close, so a process that +// dies without closing it loses at most one interval. +func TestStore_SavesAChangeWithinTheFlushInterval(t *testing.T) { + path := filepath.Join(t.TempDir(), "s.json") + s := openT(t, path, newClock(), withFlushEvery(10*time.Millisecond)) + _ = s.Set(ctx, "k", "v", 0) + + deadline := time.Now().Add(5 * time.Second) + for { + if r, err := openWith(path, withFlushEvery(time.Hour)); err == nil { + v, _ := r.Get(ctx, "k") + _ = r.Close() + if v == "v" { + return + } + } + if time.Now().After(deadline) { + t.Fatal("the change was not saved while the store was open") + } + time.Sleep(10 * time.Millisecond) + } +} + +// A file that does not parse is set aside rather than refusing to open: the store +// holds state worth keeping, not state worth failing a proxy start over. +func TestStore_AnUnreadableFileIsSetAsideAndTheStoreStartsEmpty(t *testing.T) { + path := filepath.Join(t.TempDir(), "s.json") + if err := os.WriteFile(path, []byte("{not json"), 0o600); err != nil { + t.Fatal(err) + } + s := openT(t, path, newClock()) + if v := get(t, s, "k"); v != "" { + t.Errorf("Get = %q, want an empty store", v) + } + if b, err := os.ReadFile(path + ".corrupt"); err != nil || string(b) != "{not json" { + t.Errorf("set-aside copy = %q, %v; want the unreadable file kept", b, err) + } +} + +// The file names sessions, so it is the user's alone, as the session archive is. +func TestStore_TheFileIsPrivate(t *testing.T) { + path := filepath.Join(t.TempDir(), "dir", "s.json") + s := openT(t, path, newClock()) + _ = s.Set(ctx, "k", "v", 0) + if err := s.Close(); err != nil { + t.Fatal(err) + } + fi, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if perm := fi.Mode().Perm(); perm != 0o600 { + t.Errorf("file mode = %v, want 0600", perm) + } + di, _ := os.Stat(filepath.Dir(path)) + if perm := di.Mode().Perm(); perm != 0o700 { + t.Errorf("dir mode = %v, want 0700", perm) + } +} + +func TestStore_AfterCloseOperationsFail(t *testing.T) { + s := openT(t, filepath.Join(t.TempDir(), "s.json"), newClock()) + if err := s.Close(); err != nil { + t.Fatal(err) + } + if err := s.Close(); err != nil { + t.Errorf("second Close = %v, want nil", err) + } + if err := s.Set(ctx, "k", "v", 0); err == nil { + t.Error("Set after Close = nil error") + } +} + +// storage.Open reaches the driver by the file scheme, as it reaches redis by +// "redis". +func TestOpen_IsRegisteredForTheFileScheme(t *testing.T) { + path := filepath.Join(t.TempDir(), "s.json") + st, err := storage.Open("file", "file://"+path) + if err != nil { + t.Fatalf("storage.Open: %v", err) + } + defer st.Close() + if err := st.Set(ctx, "k", "v", 0); err != nil { + t.Fatal(err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(path); err != nil { + t.Errorf("the scheme's path was not the store's file: %v", err) + } +} diff --git a/core/storage/store.go b/core/storage/store.go index 0984a5ea4..718be9afc 100644 --- a/core/storage/store.go +++ b/core/storage/store.go @@ -7,8 +7,9 @@ import ( // Store provides key-value and hash operations for plugins that need // persistent, cross-pod state (budget counters, session data, caches, -// rate limiting). Implementations live in separate modules to isolate -// their dependencies. +// rate limiting). Implementations with dependencies live in separate +// modules to isolate them, as redis does; filestore, a single process's +// store saved to one file, has none and lives in core. type Store interface { // Get returns the value for a simple key. Returns "" and nil if the key does not exist. Get(ctx context.Context, key string) (string, error) diff --git a/docs/agents/claude-code.md b/docs/agents/claude-code.md index ef4cf3357..077db6ac6 100644 --- a/docs/agents/claude-code.md +++ b/docs/agents/claude-code.md @@ -178,22 +178,25 @@ events table shows where each request went and, under METHOD, the model it was sent for, and the detail pane's `redirected:` and `model:` lines name the host and the model Claude Code asked for when they were others. -A session is pinned on the first request the router sees from it, so switch -**before** `/clear` or a new `claude`, not after. A conversation already running -when you first route Claude Code keeps the server its requests were going to, even -if it sent nothing while you ran `agentop server`: Cortex reads which server its -last request reached from the session's history, and from then on sends it there -with that server's key. That history is in the proxy's memory, which keeps the most -recently used sessions (100 by default), so a conversation quiet long enough to be -dropped before you route Claude Code is taken for a new session. A proxy restart -forgets the pins and that history, so a running conversation's next request after a -restart is taken for a new session's and goes to Claude Code's current server; see -[Known issues](#known-issues). A conversation -carried to a new session id is pinned as new, to its agent's current server, which -happens in three ways: Claude Code's continued-in hand-off copies a live -conversation to a new id; `claude daemon` starts background jobs with -`--session-id --fork-session --resume `; and `/clear` with background -subagents running moves their traffic to the new id. +A session is pinned on its conversation's opening turn, so switch **before** +`/clear` or a new `claude`, not after. Claude Code's request says which turn it +is: an opening turn carries no assistant reply yet. A conversation Cortex did not +see begin — one already running when you first route Claude Code — is never moved: +if the session's history shows which server its conversation's last request +reached, Cortex keeps it there with that server's key, and otherwise it goes where +Claude Code sends it, unrouted. Claude Code's one-shot requests (the title, auto +mode's classifier) and its subagents' requests decide nothing; they follow the +session's pin once there is one. Pins are kept in `~/.cortex/plugin-state.json`, +so a proxy restart moves no conversation. + +A conversation carried to a new session id is one Cortex did not see begin, so it +goes where Claude Code sends it, and one routed to another server moves back; see +[Known issues](#known-issues). That happens in three ways: Claude Code's +continued-in hand-off copies a live conversation to a new id; `claude daemon` +starts background jobs with `--session-id --fork-session --resume `; +and `/clear` with background subagents running moves their traffic to the new id, +where it goes where Claude Code sends it until the new conversation's opening turn +pins the session. Nothing fails over. When a session's server is down its requests fail, with a 502 from the proxy when the server cannot be reached; route the agent to another @@ -244,14 +247,13 @@ whose servers do not resolve. - **`disable` discards hand-edits.** If you change one of the seven keys by hand after `enable`, `disable` still puts back the value from before `enable`, without a warning ([#1289](https://github.com/rossoctl/cortex/issues/1289)). -- **A proxy restart can move a running conversation to another server.** Cortex - keeps a conversation on its server by a pin and by the session's history, and - both live in the proxy's memory; the session archive on disk is not read for it. - After a restart, a conversation's next request looks like a new session's and - goes to the server Claude Code is routed to now, so one started on a different - server moves, mid-conversation. To keep it where it was, route Claude Code back - to that server with `agentop server use --agent claude-code` before - continuing it, and switch again once it has sent a request, which pins it there. +- **A conversation moved to a new session id leaves its server.** Cortex keeps a + conversation on its server by its session id's pin, and Claude Code's + continued-in hand-off and `claude daemon`'s forked background jobs carry a + conversation to a new id, which no pin follows. Cortex then sees a conversation + it did not see begin and sends it where Claude Code sends it, so one routed to + another server moves back to the server `ANTHROPIC_BASE_URL` names, + mid-conversation. - **A continued conversation can show twice.** Claude Code sometimes moves a live conversation to a new session id. agentop then shows two rows with the same title, and the older row keeps the copied title. Cortex does not yet read Claude Code's record of diff --git a/docs/laptop-service.md b/docs/laptop-service.md index ba9c1388a..a9ec842c2 100644 --- a/docs/laptop-service.md +++ b/docs/laptop-service.md @@ -346,9 +346,9 @@ first by the time they were last written: A full disk does not stop the proxy. The archive stops writing until its next hourly pass, and requests are served as before; only the history has a gap. -**What it does not promise.** Writing happens off the request path, so a request never waits -on the disk — and the cost of that is that a burst the writer cannot keep up with is dropped -*from the archive*, never from memory. An unclean kill loses at most the last second. +**What it does not promise.** Writing happens off the request path — and the cost of that is +that a burst the writer cannot keep up with is dropped *from the archive*, never from memory. An +unclean kill loses at most the last second. `GET /v1/sessions?archived=true` reports both the archive's size and what it lost, under `archive` (`bytes`, `maxBytes`, `retentionDays`, and `droppedEvents`, `writeErrors`, `droppedRenames`, `paused` when any is nonzero); nonzero means what you are reading has gaps. @@ -386,6 +386,14 @@ session: That stops new writes and leaves what is on disk; clear first with `X` if you want it gone, or `rm -rf ~/.cortex/sessions` with the proxy stopped. +## Plugin state is kept in `~/.cortex/plugin-state.json` + +Plugins that need state to outlive a restart keep it here. Today that is the inference router: +which server each session is on, as its session id, its agent's name and the server's name. It +holds no prompts and no keys. A session's entry expires 29 to 30 days after its last request; `X` +does not clear it. The proxy saves changes within two seconds and on stop, and sets aside a file it +cannot read as `plugin-state.json.corrupt` and starts empty. + ## `agentop: command not found` Setup puts both binaries in `~/.local/bin`. If that is not on your PATH, adding it to diff --git a/docs/plugin-catalog.md b/docs/plugin-catalog.md index 74ce201de..986726903 100644 --- a/docs/plugin-catalog.md +++ b/docs/plugin-catalog.md @@ -187,39 +187,50 @@ transparently redirected connection is dialed where the client chose, and is `skip/not_an_inference_server`. On a server's host every path is handled, `/v1/models` and `count_tokens` included, and the session decides: -- The first request the router sees from a session pins it, in the process store, +- A session is pinned by the first request that decides it, by the order below, to where that request went: a server when the redirect took effect, or "not routed" when the request stayed where the client sent it — because the agent is not routed, or because the router runs under `on_error: observe`, even for an agent with a server. A session started under observe therefore stays unrouted - after a switch to enforce. A first request whose redirect failed, or that was - refused for its model, pins nothing, and the next one decides again. Every - request renews the pin, which lapses 30 days after the last. -- Where that request goes is decided by the session's history, for a routed - agent: a session already running when routing was first configured, quiet while - it was, has no pin but is not new. The latest earlier inference request that - agent sent in the session to a server's host, as the session store holds it, - keeps the session on that server, from then with that server's key. A session - with no such request is new and goes to its agent's current server. A request to - any other host is no evidence, since the router never routes one: an OpenCode - session that used another provider first is new when it addresses a server. - The history is read from the `inference` record `inference-parser` puts on each - request, so the rule needs `inference-parser` in the outbound chain; without it, a - session the router has not pinned is always treated as new. + after a switch to enforce. A request whose redirect failed, or that was refused + for its model, pins nothing, and the next one decides again. Every request renews + the pin, which lapses 30 days after the last. On a local install pins are kept in + `~/.cortex/plugin-state.json`, which a restart keeps; elsewhere in the process + store, which a restart empties. +- A request in a session its agent has not pinned is decided in this order, for a + routed agent: + 1. The session's history: the latest earlier inference request, other than a + side request (3), that agent sent in the session to a server's host keeps + the session on that server, from then with that server's key. The history + is the session store's and, where the proxy keeps a session archive, up to + 2000 of the archive's events from before the oldest the store holds. A + request to any other host is no evidence, since the router never routes + one: an OpenCode session that used another provider first is new when it + addresses a server. + 2. A request `inference-parser` did not read as inference — `count_tokens`, + `/v1/models` — goes to its agent's current server and pins nothing. + 3. A Claude Code request's own turn: an opening turn, with no assistant reply + yet, goes to its agent's current server; a continuation, a conversation the + router did not see begin, is not routed, and is pinned so; a one-shot with no + tools, or a subagent's request, is not routed and pins nothing. + 4. Otherwise its agent's current server. + + The router reads the `inference` record `inference-parser` puts on each request, + so a chain without `inference-parser` before the router is refused. - A pin is its agent's. A request from another agent filed under the same session id — by process attribution, which files a command an agent runs under that agent's session and is on by default on a laptop; by the active-session fallback with client affinity off; or by a header id two clients share — is decided as if the session were unpinned, from that agent's own history and server, and leaves the pin alone. -- Changing `agents` therefore moves only new sessions while the proxy runs: the - pins and the history survive a hot reload. A restart loses both — the router reads - the store's history in memory, not the session archive on disk — so a running - session's next request after a restart is treated as a new session's, and moves - if its agent's server is another. The store also drops quiet sessions while the - proxy runs — `session.max_sessions` (100 by default) evicts the least recently - used, and a configured `session.ttl` expires them — and a session dropped before - the router pinned it is treated as new too; a pin outlives that. A request with no session follows its agent's +- Changing `agents` therefore moves only new sessions: the pins and the history + survive a hot reload, and on a local install a restart too. A proxy with neither + a plugin store nor a session archive loses both on restart, and a running + session's next request is then decided as an unpinned one's. The store also drops + quiet sessions while the proxy runs — `session.max_sessions` (100 by default) + evicts the least recently used, and a configured `session.ttl` expires them — and + a session dropped before the router pinned it has only the archive's history, + where there is one; a pin outlives that. A request with no session follows its agent's current server, unpinned, and so does one filed under a synthetic session — the `default` bucket or a `pending:` id — since each holds many conversations, not one.