Merge branch 'sidecars-claude' into sidecars
This commit is contained in:
@@ -23,8 +23,15 @@ One Bun process (`src/server.tsx`) serves everything:
|
||||
|
||||
Long-running and privileged work lives in **sidecars**: separate processes that dial back in over
|
||||
`/api/sidecar/register` and are tracked in `src/servers/sidecar-registry.ts`. PM2 runs them
|
||||
(`ecosystem.config.cjs`): `officer` (the server), `officer-claude`, `officer-opencode`,
|
||||
`officer-email`, `officer-pty`, `officer-vnc`.
|
||||
(`ecosystem.config.cjs`): `officer` (the server), `officer-anthropic-proxy`, `officer-agent`,
|
||||
`officer-opencode`, `officer-email`, `officer-pty`, `officer-vnc`, `officer-music`, `officer-vault`,
|
||||
`officer-slskd`.
|
||||
|
||||
**`officer-anthropic-proxy` and `officer-agent` are not the same thing.** The proxy holds the Anthropic
|
||||
credential and forwards API traffic; the agent is the process that spawns `claude`. They were one entry
|
||||
named `officer-claude` until the sidecar-isolation work — which is exactly how the false claim that
|
||||
"restarting officer doesn't disturb the agent" survived so long. Every sidecar is a PM2 peer of
|
||||
`officer`, so **no sidecar is a child of the server and restarting the server does not kill one.**
|
||||
|
||||
Agents run **unsandboxed as the server owner**, with `--dangerously-skip-permissions`. This is
|
||||
deliberate — it is the owner's own machine. Do not add a jail without being asked.
|
||||
|
||||
@@ -0,0 +1,207 @@
|
||||
# Frontend Navigation Audit — the "opaque click" anti-pattern
|
||||
|
||||
**Date:** 2026-07-30 · **Branch:** `sidecars-claude` · **Status:** exploration + documentation only (no code changed)
|
||||
|
||||
Prep work for the upcoming **full navigation refactor**. This catalogues every place the frontend
|
||||
navigates to / selects an **addressable entity** using a `<button onClick>` / `<div onClick>` /
|
||||
imperative `navigate()` / global-channel setter **instead of a real `<Link to>` / `<a href>`**.
|
||||
|
||||
> Line numbers were captured on `sidecars-claude` at audit time and may drift as other work lands —
|
||||
> treat them as anchors, re-grep the symbol if a line looks off.
|
||||
|
||||
---
|
||||
|
||||
## The anti-pattern (definition)
|
||||
|
||||
A clickable element selects/opens something that has (or should have) a URL, but:
|
||||
- **(a)** the entity id/slug is **not in the DOM** (no `href`, no `data-*`) — it lives only in an onClick closure;
|
||||
- **(b)** clicking **doesn't change the URL** (or does so only via an indirect state→URL effect);
|
||||
- **(c)** selection is held in **JS state / a global channel** (`usePanelChannel`, `useGlobal`), not the URL;
|
||||
- **(d)** **no anchor semantics** — can't cmd/ctrl-click to a new tab, no middle-click, not link-focusable.
|
||||
|
||||
**Exemplar (already fixed):** the `/chat` session list. Rows were `<button onClick={() => selectById(id)}>`
|
||||
(id only in the closure) → converted to `<Link to={`/chat/${session.id}`}>` (committed to master `f35c145`).
|
||||
That fix is the template for the HIGH items below. **Caveat:** the fix only did the *rows* — the chat
|
||||
**detail panel** still selects via channel, not the URL (finding **C1**), so `/chat` is the model for both
|
||||
"done right" (rows) and "still to do" (detail).
|
||||
|
||||
## Route map (from `App.tsx`)
|
||||
|
||||
**Existing entity routes:** `/chat/:sessionId`, `/jobs/:id`, `/dashboards/:id`, `/projects/:id`, `/email/:emailId`.
|
||||
**Flat screens:** `/` `/files` `/music` `/soulseek` `/tasks` `/skills` `/processes` `/activity`
|
||||
`/system-monitor` `/task-logs` `/plans` `/terminal` `/desktop` `/browser` `/code-editor` `/settings/{profile,ai,system,integrations,apps}`.
|
||||
|
||||
**What's already correct** (lean on these in the refactor): the **Dock**, **Header** (logo + mobile sheet),
|
||||
**UserMenu**, **JobsIndicator** are all real `<Link>`s. Shared `NavLink.tsx` (query-string-appending `<Link>`
|
||||
wrapper — note: *not* react-router's NavLink, gives no active state) and `BackButton.tsx` (`<Link>` back arrow)
|
||||
are good building blocks. The **Workspace/Panel framework** contains **zero** route navigation — it's orthogonal.
|
||||
|
||||
---
|
||||
|
||||
## Consolidated findings — severity ranked
|
||||
|
||||
### 🔴 HIGH — addressable route already exists; just needs a `<Link>` / URL-as-source-of-truth
|
||||
|
||||
| ID | file:line | Entity | Current impl | Fix |
|
||||
|----|-----------|--------|--------------|-----|
|
||||
| H1 | `Screens/Dashboard/Jobs/JobsPage.tsx:108` | a job | `<button onClick={() => navigate(`/jobs/${job.id}`)}>` — id in closure | → `<Link to={`/jobs/${job.id}`}>`. Active-row already keys off `useParams().id`; keep the stop/delete button. **The exact twin of the /chat fix.** |
|
||||
| H2 | `workspaces/…/apps/Dashboards/DashboardListApp.tsx:133` | a dashboard | `<div onClick={handleClick}>` → `useGlobal(SELECTED_DASHBOARD_KEY)` on-page (**no URL change**), `navigate()` off-page | rows → `<Link to={`/dashboards/${ws.id}`}>`; drop the global as selection source (derive from `useParams`). Header is already a `<Link>` — app is internally inconsistent. |
|
||||
| H3 | `workspaces/…/apps/Projects/ProjectListApp.tsx:161` | a project | `<div onClick={handleClick}>` → `useGlobal(SELECTED_PROJECT)` on-page (**no URL change**), `navigate()` off-page | identical to H2 → `<Link to={`/projects/${p.id}`}>`; retire `SELECTED_PROJECT` as source of truth. |
|
||||
| H4 | `workspaces/…/apps/ChatHistory/ChatDetailPanel.tsx:136` | which chat to render | reads `usePanelChannel('chat:selected-session')`, **not** `useParams` | make the route the source of truth: read `sessionId` from `useParams`, fetch by id, retire `chat:selected-session` (or make it a derived cache). **Finishes the /chat fix.** |
|
||||
| H5 | `Screens/Dashboard/Email/EmailList.tsx:336` | an email | `<button data-email-id onClick={setSelectedId}>` + `EMAIL_SELECTED` channel + a state↔URL sync effect (`EmailScreen.tsx:27-33`) | rows → `<Link to={`/email/${msg.id}`}>`, drive selection **from** `useParams().emailId`, delete the two sync effects. (Already round-trips the URL + has `data-email-id`, but still isn't an anchor.) |
|
||||
|
||||
> **Verify:** `Dashboards/DashboardPreview.tsx:355,377` and `Projects/ProjectPreview.tsx:430,457` also
|
||||
> `navigate('/…/:id')`. The nav auditor read these as HIGH "open entity"; the apps auditor read the Project
|
||||
> ones as *correct* post-create/edit flows. Eyeball them during H2/H3 — convert the ones that are "open an
|
||||
> existing entity from a preview", leave genuine post-mutation redirects.
|
||||
|
||||
### 🟠 MEDIUM — navigable entity with **no route yet** (add a route, then link)
|
||||
|
||||
| ID | file:line | Entity | Proposed route | Note |
|
||||
|----|-----------|--------|----------------|------|
|
||||
| M1 | `Screens/Dashboard/CapabilityPage.tsx:431` | task / skill / process | `/tasks/:dir`, `/skills/:dir`, `/processes/:dir` | one component backs **three** screens (Tasks/Skills/Processes). Highest-value MEDIUM. Also drops the auto-select-`items[0]` effect. |
|
||||
| M2 | `Screens/Dashboard/TaskLogs/index.tsx:104` | a task-log run | `/task-logs/:id` | detail fetch already keys off the id — clean move. |
|
||||
| M3 | `Screens/Dashboard/Activity/ActivityScreen.tsx:63,73` | background task / detached job | `/activity/:id` | two row types; unify under one param, screen re-derives `task=`/`path=`. |
|
||||
| M4 | `FileBrowser/.../useFileBrowserApp.ts:269`, `FileItem.tsx:516`, `Breadcrumb.tsx:16`, search-hit `:249` | a folder | `/files?path=<dir>` | **files** already open via `?view=`; **folders** are pure `currentPath` state — no URL, no back/forward. Folder rows + crumbs → `<Link>` on a `?path=` param. |
|
||||
| M5 | `CodeEditor/FileTree.tsx:59`, `EditorTabs.tsx:33` | open source file / active tab | `/code-editor?file=<path>` (+ `open=` for tabs) | tree file rows → `<Link>`; active file in a param. |
|
||||
| M6 | `Layout/Settings/SettingsPanel.tsx:39` | a settings sub-section | `/settings/:page/:section` or `?section=` | `<button onClick={setSelectedKey}>` → `useGlobal`; nothing in URL, not deep-linkable. **One change covers all settings pages** (all go through `createSettingsPanelComponents`). |
|
||||
| M7 | `workspaces/components/Combobox.tsx:53` | caller-supplied route | (existing) | shared widget: `href`-bearing options do `navigate(option.href)`; render them as `<Link>`. Every caller inherits the opaque click. |
|
||||
| M8 | `Layout/Header/UserMenu.tsx:51` | — | — | **dead target:** `<Link to="/settings/resources">` — route doesn't exist (falls through to `/`). Add the route or remove the item. |
|
||||
| M9 | `Screens/Dashboard/Plans/index.tsx:36` | a plan document | `/plans/:name` | native `<select>` → local state; a plan is a real addressable doc. Have the select `navigate()` or use a link list. |
|
||||
| M10 | `SystemMonitor/ScopeList.tsx:14` | monitor scope (btop/pm2/docker) | `/system-monitor/:scope` | small fixed set; low urgency but genuinely addressable. |
|
||||
|
||||
**Music** (M-music) and **Soulseek** (M-slsk) are whole-workspace channel apps — pulled out below because each
|
||||
is **one design decision** that cascades across many files:
|
||||
|
||||
- **Music** — `Music/{MusicBrowser:182,209 · MusicDetail:138,174,234 · FavoritesView:85,107}` hold the entire
|
||||
library location in `usePanelChannel('music:cwd')`; every drill-in is `setCwd(path)`. No URL, no deep-link to
|
||||
an album. → back `music:cwd` with `/music?path=<rel>`; rows become `<Link>`s.
|
||||
- **Soulseek** — an entire workspace with **no URL at all**: `soulseek:section` nav (`SoulseekNav:40`,
|
||||
`SoulseekDashboard:88/96/110/121`), `soulseek:user` peer channel (`UserMenu:32`, `SoulseekUsers:42`), and
|
||||
per-panel `useState` for past-search (`SearchView:160`, a textbook `role="button"` div with `s.id` in the
|
||||
closure), rooms (`SoulseekRooms:153`), conversations (`SoulseekChat:135`). → decide a nested tree
|
||||
`/soulseek/:section` + `/soulseek/{users/:name, search/:id, rooms/:name, chat/:name}`. **S1 (the section nav)
|
||||
decides S2–S7.**
|
||||
|
||||
### 🟡 LOW / borderline
|
||||
|
||||
- **Dock / Header active styling** (`Dock.tsx:43,78` · `Header.tsx:53,112`) — destinations are already real
|
||||
`<Link>`s; only the *active* class is JS-derived from `useLocation`. Switch to react-router `<NavLink>` and drop the hand-rolled `isActive`.
|
||||
- **Browser tabs** (`Browser/TabList.tsx:93`) — ephemeral CDP targets (polled, come/go every few seconds); a durable `/browser/:targetId` is dubious. Probably leave as selection.
|
||||
- **Jobs step/iteration** (`Jobs/JobDetail.tsx:234,255`) — intra-detail sub-selection; only if we want `/jobs/:id/:stepKey` deep links.
|
||||
- **Preview slug** (`Preview/PreviewApp.tsx:23`) — Preview usually derives its target from the surrounding dashboard/project `cwd`; the manual picker is a fallback. Probably leave URL-less.
|
||||
- **FileBrowser widget** (`FileBrowserWidget/.../BrowseTab.tsx:36`) — compact sidebar widget; keeping its browse cursor local is defensible.
|
||||
- **Music favorites view** (`music:fav`) & **SystemMonitor scope** — view toggles over tiny fixed sets; channel-state is arguably fine.
|
||||
- **"New Chat" button** (`ChatHistory/SessionList.tsx:68`) — a create-action that also `navigate('/chat/new')`; could be `<Link to="/chat/new">` if the state-set moves into the route. Its `chat:selected-session` write is part of the H4 cleanup.
|
||||
|
||||
---
|
||||
|
||||
## Channel-based selection map (the core of the refactor)
|
||||
|
||||
Every place an **addressable entity** is selected through a global channel / global state instead of the URL.
|
||||
This is the primary surface to convert to URL-driven selection.
|
||||
|
||||
| Channel / global key | Entity held | Should map to | Files |
|
||||
|----------------------|-------------|---------------|-------|
|
||||
| `chat:selected-session` | open chat session | `/chat/:sessionId` | `ChatDetailPanel.tsx:136`, `SessionList.tsx:16` (H4) |
|
||||
| `chat:active-cwd` | chat working dir | query param on `/chat` | `ChatDetailPanel.tsx:102`, `SessionList.tsx:14` |
|
||||
| `SELECTED_DASHBOARD_KEY` (`useGlobal`) | selected dashboard | `/dashboards/:id` | `DashboardListApp.tsx:36`, `DashboardPreview.tsx:287` (H2) |
|
||||
| `SELECTED_PROJECT` (`useGlobal`) | selected project | `/projects/:id` | `ProjectListApp.tsx:45`, `ProjectPreview.tsx:386` (H3) |
|
||||
| `EMAIL_SELECTED` | open email | `/email/:emailId` | `EmailList.tsx:336`, `EmailScreen.tsx` sync (H5) |
|
||||
| `music:cwd` | library location (album/artist/folder) | `/music?path=` | `MusicBrowser/MusicDetail/FavoritesView` |
|
||||
| `soulseek:section` | active Soulseek section | `/soulseek/:section` | `SoulseekNav/View/Dashboard/UserMenu` |
|
||||
| `soulseek:user` | selected peer | `/soulseek/users/:name` | `UserMenu.tsx:28`, `SoulseekUsers.tsx:34` |
|
||||
| `monitor:scope` | monitor scope | `/system-monitor/:scope` | `ScopeList.tsx` (via `shared.ts:15`) |
|
||||
| settings `useGlobal(*_SELECTED)` | settings section | `/settings/:page/:section` | `SettingsPanel.tsx:56` (M6) |
|
||||
| local `useState` selections | search / room / conversation / peer / open file / preview slug | respective nested routes | `SearchView`, `SoulseekRooms`, `SoulseekChat`, `SoulseekUsers`, `CodeEditor/useEditorState`, `PreviewProvider` |
|
||||
|
||||
**Excluded — event-bus / refresh signals, NOT selection:** `preview:refresh`, `files:refresh-signal`,
|
||||
`SLSKD_REFRESH_CHANNEL`, `MUSIC_RESYNC_CHANNEL`, `FILE_VIEWER_CHANNEL`, `chat:active-session` (embedded-chat signal).
|
||||
|
||||
---
|
||||
|
||||
## Appendix A — all `navigate()` call sites (whole frontend, 25)
|
||||
|
||||
"Link?" = a `<Link>`/`<NavLink>` is the right refactor.
|
||||
|
||||
| # | file:line | what | target | Link? | note |
|
||||
|---|-----------|------|--------|-------|------|
|
||||
| 1 | `Jobs/JobsPage.tsx:109` | job list row | `/jobs/:id` | **YES** | H1 |
|
||||
| 2 | `Dashboards/DashboardListApp.tsx:90` | dashboard row (off-page) | `/dashboards/:id` | YES | H2 |
|
||||
| 3 | `Dashboards/DashboardListApp.tsx:114` | inside "New Dashboard" | `/dashboards` | ~ | create action |
|
||||
| 4 | `Dashboards/DashboardPreview.tsx:355` | open dashboard | `/dashboards/:id` | YES | verify vs H2 |
|
||||
| 5 | `Dashboards/DashboardPreview.tsx:377` | open dashboard | `/dashboards/:id` | YES | verify vs H2 |
|
||||
| 6 | `Projects/ProjectListApp.tsx:116` | project row (off-page) | `/projects/:id` | YES | H3 |
|
||||
| 7 | `Projects/ProjectListApp.tsx:142` | inside "New Project" | `/projects` | ~ | create action |
|
||||
| 8 | `Projects/ProjectPreview.tsx:430` | open project | `/projects/:id` | YES | verify vs H3 |
|
||||
| 9 | `Projects/ProjectPreview.tsx:457` | open project | `/projects/:id` | YES | verify vs H3 |
|
||||
| 10 | `Jobs/JobDetail.tsx:709` | back button | `/jobs` | YES | use `BackButton`/`<Link>` |
|
||||
| 11 | `ChatHistory/SessionList.tsx:70` | "New Chat" | `/chat/new` | ~ | borderline (LOW) |
|
||||
| 12 | `components/Combobox.tsx:54` | nav combobox option | `option.href` | YES | M7 |
|
||||
| 13 | `MusicPlayer/MusicPlayerHost.tsx:251` | open current album (setCwd+nav) | `/music` | ~ | has side-effect |
|
||||
| 14 | `FileBrowserWidget/useFileBrowserWidget.ts:53` | open folder in Files | `/files?view=…` | ~ | query nav |
|
||||
| 15 | `FileBrowserApp/useFileBrowserApp.ts:401` | create dashboard from folder | `/dashboards/new?…` | ~ | action-nav |
|
||||
| 16 | `FileBrowserApp/useFileBrowserApp.ts:427` | create dashboard from files | `/dashboards/new?…` | ~ | action-nav |
|
||||
| 17 | `FileBrowserApp/.../TaskRunnerModal.tsx:1168` | after starting job | `/jobs` `/jobs/:id` | NO | post-submit |
|
||||
| 18 | `FileBrowserApp/.../TaskRunnerModal.tsx:1649` | after starting pipeline | `/jobs/:id` | NO | post-submit |
|
||||
| 19 | `Email/EmailScreen.tsx:29` | sync selection → URL (effect) | `/email/:emailId` | NO | delete in H5 |
|
||||
| 20 | `Email/EmailScreen.tsx:31` | clear selection → URL (effect) | `/email` | NO | delete in H5 |
|
||||
| 21 | `ChatHistory/index.tsx:107` | redirect when no id | `/chat` | NO | guard |
|
||||
| 22 | `Projects/ProjectListScreen.tsx:66` | clear/redirect | `/projects` | NO | guard |
|
||||
| 23 | `Authentication/ForgotPassword/useResetPassword.ts:48` | post-reset redirect | `/` | NO | action result |
|
||||
| 24 | `Authentication/LandingPage/Bootstrap.tsx:34` | post-bootstrap redirect | `/` | NO | action result |
|
||||
| 25 | `hooks/useQueryState/useQueryState.ts:39` | generic query-state writer | dynamic | NO | infra hook |
|
||||
|
||||
---
|
||||
|
||||
## ✅ TO-DO — ordered for tomorrow
|
||||
|
||||
### Phase 1 — Quick wins (routes already exist; mechanical, high value)
|
||||
- [ ] **H1** Jobs rows → `<Link to={`/jobs/${job.id}`}>` (`JobsPage.tsx:108`). The clean twin of the /chat fix.
|
||||
- [ ] **H2** Dashboards rows → `<Link>`; retire `SELECTED_DASHBOARD_KEY` as selection source (`DashboardListApp.tsx:133`).
|
||||
- [ ] **H3** Projects rows → `<Link>`; retire `SELECTED_PROJECT` (`ProjectListApp.tsx:161`).
|
||||
- [ ] **H4** Chat detail: read `sessionId` from `useParams`, retire `chat:selected-session` as source of truth (`ChatDetailPanel.tsx:136`) — **finishes the /chat fix**.
|
||||
- [ ] **H5** Email rows → `<Link>` driven by `useParams().emailId`; delete the state↔URL sync effects (`EmailList.tsx:336`, `EmailScreen.tsx:27-33`).
|
||||
- [ ] **M8** Fix or remove the dead `/settings/resources` link (`UserMenu.tsx:51`).
|
||||
- [ ] Verify + convert the 4 preview "open" navigates (`DashboardPreview:355,377`, `ProjectPreview:430,457`); leave genuine post-mutation redirects.
|
||||
|
||||
### Phase 2 — Add a route, then link (per-entity, medium effort)
|
||||
- [ ] **M6** Settings sub-sections → nested route or `?section=`, render `SectionButton` as `<NavLink>` (`SettingsPanel.tsx:39`) — one change covers all settings pages.
|
||||
- [ ] **M1** Capabilities → `/tasks|skills|processes/:dir`, rows → `<Link>` (`CapabilityPage.tsx:431`) — covers 3 screens.
|
||||
- [ ] **M2** TaskLogs → `/task-logs/:id` (`TaskLogs/index.tsx:104`).
|
||||
- [ ] **M3** Activity → `/activity/:id` (`ActivityScreen.tsx:63,73`).
|
||||
- [ ] **M4** FileBrowser folders → `/files?path=`; folder rows + breadcrumbs → `<Link>` (`useFileBrowserApp.ts:269`, `FileItem.tsx`, `Breadcrumb.tsx`).
|
||||
- [ ] **M5** CodeEditor active file → `/code-editor?file=` (`FileTree.tsx:59`, `EditorTabs.tsx:33`).
|
||||
- [ ] **M7** Combobox: render `href` options as `<Link>` (`Combobox.tsx:53`).
|
||||
|
||||
### Phase 3 — Whole-workspace routing decisions (needs a design call first)
|
||||
- [ ] **Music** — decide `/music?path=` (or nested); back `music:cwd` with the URL; rows → `<Link>` (M-music).
|
||||
- [ ] **Soulseek** — decide the nested tree `/soulseek/:section` + `/soulseek/{users,search,rooms,chat}/:x`; **S1 decides S2–S7** (M-slsk).
|
||||
- [ ] **M10** SystemMonitor scope → `/system-monitor/:scope` (`ScopeList.tsx:14`).
|
||||
- [ ] **M9** Plans → `/plans/:name` (`Plans/index.tsx:36`).
|
||||
|
||||
### Phase 4 — Polish + borderline decisions
|
||||
- [ ] Dock + Header + mobile sheet → react-router `<NavLink>` for active state; drop hand-rolled `isActive` (`Dock.tsx`, `Header.tsx`).
|
||||
- [ ] "New Chat" → `<Link to="/chat/new">` (`SessionList.tsx:68`) once H4's channel cleanup lands.
|
||||
- [ ] Jobs back button + any `navigate('/jobs')` → shared `BackButton` (`JobDetail.tsx:709`).
|
||||
- [ ] Decide/skip: Browser tabs, Jobs step deep-link, Preview slug, FileBrowser widget, Music-favorites/Monitor-scope view toggles.
|
||||
|
||||
### Cross-cutting for the refactor itself
|
||||
- [ ] Standardise a URL-as-source-of-truth pattern for panel selection (replace the `usePanelChannel`/`useGlobal`
|
||||
selection channels in the map above with `useParams`/`useSearchParams`, keeping channels only for
|
||||
genuine signals/refresh buses).
|
||||
- [ ] Adopt `<NavLink>` (real react-router) for all nav chrome so active state stops being JS-derived.
|
||||
- [ ] Keep `NavLink.tsx` (query-string wrapper) and `BackButton.tsx` as the standard building blocks.
|
||||
|
||||
---
|
||||
|
||||
## Coverage
|
||||
|
||||
- **Area 1 — Dashboard screens** (`Screens/Dashboard/**`, excl. Settings/Layout): 9 findings (2 HIGH, 4 MEDIUM, 3 LOW) + spillover to `officerdev` apps.
|
||||
- **Area 2 — Workspace panel apps** (`workspaces/officerdev/src/apps/**`): 22 findings (4 HIGH, ~15 MEDIUM, 3 LOW); 11 distinct channel-selection sites.
|
||||
- **Area 3 — Global nav / Layout / Settings / shared components**: 6 findings (0 HIGH — chrome is already links; 3 MEDIUM, 3 LOW) + the 25-site `navigate()` map.
|
||||
|
||||
Raw per-area findings (with full impl / id-in-DOM / URL-change / coverage notes) were produced in the
|
||||
audit run; this document is the consolidated, de-duplicated synthesis. The three HIGH clusters
|
||||
(Jobs · Dashboards · Projects) plus the two chat halves (rows done, detail = H4) are the direct siblings of
|
||||
the `/chat` fix and the highest-value starting point.
|
||||
@@ -57,8 +57,13 @@ Commit messages: simple lowercase, no prefixes, explaining *why*.
|
||||
|
||||
## Running and checking your work
|
||||
|
||||
The server runs under pm2 as `officer`, plus sidecars (`officer-claude`, `officer-opencode`,
|
||||
`officer-email`, `officer-pty`, `officer-vnc`). `pm2 list` shows them; `pm2 logs officer` follows.
|
||||
The server runs under pm2 as `officer`, plus sidecars (`officer-anthropic-proxy`, `officer-agent`,
|
||||
`officer-opencode`, `officer-email`, `officer-pty`, `officer-vnc`, `officer-music`, `officer-vault`,
|
||||
`officer-slskd`). `pm2 list` shows them; `pm2 logs officer` follows.
|
||||
|
||||
Two of those names are worth knowing apart: **`officer-anthropic-proxy` holds the Anthropic credential
|
||||
and proxies API traffic; `officer-agent` is the process that actually runs `claude`.** They used to be
|
||||
one confusingly-named entry (`officer-claude`) that was only the proxy.
|
||||
|
||||
**Don't restart the owner's server to test.** Boot your own on a spare port instead — the running
|
||||
instance holds `PORT` from `.env` (9010):
|
||||
|
||||
+14
-1
@@ -6,12 +6,25 @@ module.exports = {
|
||||
args: 'start',
|
||||
watch: false,
|
||||
},
|
||||
// The Anthropic credential proxy. Despite the old name (`officer-claude`) this process does NOT
|
||||
// run agents — it holds the proxy secret and forwards to api.anthropic.com. The process that runs
|
||||
// agents is `officer-agent` below.
|
||||
{
|
||||
name: 'officer-claude',
|
||||
name: 'officer-anthropic-proxy',
|
||||
script: 'bun',
|
||||
args: 'run src/servers/sidecar/claude/index.ts',
|
||||
watch: false,
|
||||
},
|
||||
// The process that actually runs `claude`. It used to be spawned on demand by the main server,
|
||||
// which made every agent session a grandchild of `officer` and killed it on every restart. As a PM2
|
||||
// peer it survives them. It resolves the owner from the database and the proxy secret from the
|
||||
// proxy's state file, so it needs nothing from `officer` in order to start.
|
||||
{
|
||||
name: 'officer-agent',
|
||||
script: 'bun',
|
||||
args: 'run src/servers/sidecar/claude/user-instance.ts',
|
||||
watch: false,
|
||||
},
|
||||
{
|
||||
name: 'officer-opencode',
|
||||
script: 'bun',
|
||||
|
||||
@@ -2,6 +2,7 @@ export {
|
||||
getUsers,
|
||||
getUserById,
|
||||
getUserByEmail,
|
||||
getOwnerUser,
|
||||
getUserCount,
|
||||
createUser,
|
||||
updateUser,
|
||||
@@ -77,7 +78,12 @@ export {
|
||||
markInterruptedJobs,
|
||||
} from './queries/pipeline-jobs';
|
||||
|
||||
export { appendChatEvent, getChatEventsSince, pruneChatEventsOlderThan } from './queries/chat-events';
|
||||
export {
|
||||
appendChatEvent,
|
||||
getChatEventsSince,
|
||||
getLastChatEventSeq,
|
||||
pruneChatEventsOlderThan,
|
||||
} from './queries/chat-events';
|
||||
|
||||
export {
|
||||
getMusicFavorites,
|
||||
|
||||
@@ -19,6 +19,14 @@ export async function getUserByEmail(email: string): Promise<UserSelect | undefi
|
||||
return user;
|
||||
}
|
||||
|
||||
// Single-user platform: there is exactly one account, created once by POST /auth/bootstrap. Sidecars
|
||||
// that need "who is the owner" (e.g. the agent sidecar, which PM2 starts with no email in its env)
|
||||
// resolve it here rather than being told by the main server.
|
||||
export async function getOwnerUser(): Promise<UserSelect | undefined> {
|
||||
const [user] = await db.select().from(users).orderBy(users.id).limit(1);
|
||||
return user;
|
||||
}
|
||||
|
||||
export async function getUserCount(): Promise<number> {
|
||||
const [result] = await db.select({ count: sql<number>`count(*)::int` }).from(users);
|
||||
return result?.count ?? 0;
|
||||
|
||||
@@ -1,13 +1,10 @@
|
||||
import { eq, and, gt, asc, lt } from 'drizzle-orm';
|
||||
import { eq, and, gt, asc, desc, lt } from 'drizzle-orm';
|
||||
import { db } from '../db';
|
||||
import { chatSessionEvents } from '../schema';
|
||||
|
||||
/** Append one outbound event to a session's durable log; returns its global cursor id. */
|
||||
export async function appendChatEvent(sessionId: string, event: unknown): Promise<number> {
|
||||
const [row] = await db
|
||||
.insert(chatSessionEvents)
|
||||
.values({ sessionId, event })
|
||||
.returning({ id: chatSessionEvents.id });
|
||||
const [row] = await db.insert(chatSessionEvents).values({ sessionId, event }).returning({ id: chatSessionEvents.id });
|
||||
return row!.id;
|
||||
}
|
||||
|
||||
@@ -23,6 +20,21 @@ export async function getChatEventsSince(
|
||||
.orderBy(asc(chatSessionEvents.id));
|
||||
}
|
||||
|
||||
/**
|
||||
* The newest cursor id for one session, or undefined if it has no events yet. The writer uses this to
|
||||
* pick its `prevSeq` chain back up after its own restart, so a client can still tell a contiguous
|
||||
* replay from one with a hole in it.
|
||||
*/
|
||||
export async function getLastChatEventSeq(sessionId: string): Promise<number | undefined> {
|
||||
const [row] = await db
|
||||
.select({ id: chatSessionEvents.id })
|
||||
.from(chatSessionEvents)
|
||||
.where(eq(chatSessionEvents.sessionId, sessionId))
|
||||
.orderBy(desc(chatSessionEvents.id))
|
||||
.limit(1);
|
||||
return row?.id;
|
||||
}
|
||||
|
||||
/** Retention: drop events older than the cutoff (called periodically). */
|
||||
export async function pruneChatEventsOlderThan(cutoff: Date): Promise<void> {
|
||||
await db.delete(chatSessionEvents).where(lt(chatSessionEvents.createdAt, cutoff));
|
||||
|
||||
@@ -138,6 +138,27 @@ export type ServerMessage =
|
||||
| { type: 'task:started'; taskId: string; description: string; taskType?: string }
|
||||
| { type: 'task:notification'; taskId: string; status: 'completed' | 'failed' | 'stopped'; summary: string };
|
||||
|
||||
// The turn-output subset of ServerMessage — everything the agent sidecar produces on its own. The
|
||||
// remaining members (session:init, sync:messages, disconnected, connection-level errors) are officer's:
|
||||
// they describe the browser's connection, not the turn.
|
||||
//
|
||||
// The sidecar builds these, commits them to chat_session_events, and hands officer a finished message
|
||||
// plus its cursor id; officer relays it verbatim. `prevSeq` is the writer's continuity claim — the
|
||||
// cursor of the previous durable message in the same session — which lets a reconnecting client tell a
|
||||
// contiguous replay from one with a hole in it. Absent when the writer cannot vouch for it.
|
||||
export type TurnMessageType =
|
||||
| 'assistant:delta'
|
||||
| 'assistant:text'
|
||||
| 'tool:start'
|
||||
| 'tool:result'
|
||||
| 'result'
|
||||
| 'error'
|
||||
| 'stopped'
|
||||
| 'task:started'
|
||||
| 'task:notification';
|
||||
|
||||
export type TurnMessage = Extract<ServerMessage, { type: TurnMessageType }> & { prevSeq?: number };
|
||||
|
||||
export type ChatEvent =
|
||||
| { type: 'text'; text: string }
|
||||
| { type: 'delta'; text: string }
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import type { ServerWebSocket } from 'bun';
|
||||
import { randomUUID } from 'crypto';
|
||||
import type { ClientMessage, ServerMessage, Message, ChatEvent } from './types';
|
||||
import type { ClientMessage, ServerMessage, Message, ChatEvent, TurnMessage, UserSession } from './types';
|
||||
import { sessionManager } from './session-manager';
|
||||
import { sendClaudeCodeStreaming } from '@@/channels/send-claude-code';
|
||||
import { sendOpenCodeStreaming } from '@@/channels/send-opencode';
|
||||
@@ -165,6 +165,82 @@ export function close(ws: ServerWebSocket<WSData>): void {
|
||||
}
|
||||
}
|
||||
|
||||
// ── Claude Code: relay, don't rebuild ──
|
||||
|
||||
// The agent sidecar owns Claude's turn output end to end — it translates the parser stream, commits each
|
||||
// message to chat_session_events and hands us a finished message plus its cursor id. Officer relays it.
|
||||
// That is what makes a restart survivable: the durable record no longer travels over the socket between
|
||||
// the two processes, so if this one is down the output is already written and the client replays it.
|
||||
//
|
||||
// Officer keeps only the in-memory transcript, which exists to answer a `resume` with sync:messages —
|
||||
// Claude's own transcript is the real record — so it is folded from the same messages, not rebuilt.
|
||||
function foldIntoSession(session: UserSession, msg: TurnMessage, model: string): void {
|
||||
switch (msg.type) {
|
||||
case 'assistant:delta':
|
||||
session.streamBuffer += msg.text;
|
||||
break;
|
||||
|
||||
case 'assistant:text':
|
||||
session.messages.push({ id: randomUUID(), timestamp: Date.now(), role: 'assistant', text: msg.text, model });
|
||||
session.meta.messageCount += 1;
|
||||
session.streamBuffer = '';
|
||||
break;
|
||||
|
||||
case 'tool:start':
|
||||
session.messages.push({
|
||||
id: randomUUID(),
|
||||
timestamp: Date.now(),
|
||||
role: 'tool',
|
||||
toolCallId: msg.toolCallId,
|
||||
toolName: msg.toolName,
|
||||
toolInput: msg.toolInput,
|
||||
});
|
||||
session.meta.messageCount += 1;
|
||||
break;
|
||||
|
||||
case 'tool:result':
|
||||
for (let i = session.messages.length - 1; i >= 0; i--) {
|
||||
const m = session.messages[i]!;
|
||||
if (m.role === 'tool' && m.toolCallId === msg.toolCallId) {
|
||||
m.output = msg.output;
|
||||
m.isError = msg.isError;
|
||||
break;
|
||||
}
|
||||
}
|
||||
break;
|
||||
|
||||
case 'result': {
|
||||
session.isGenerating = false;
|
||||
session.meta.cost.inputTokens += msg.cost.inputTokens;
|
||||
session.meta.cost.outputTokens += msg.cost.outputTokens;
|
||||
session.meta.cost.totalUSD += msg.cost.totalUSD;
|
||||
session.meta.updatedAt = Date.now();
|
||||
// The turn's cost belongs to the assistant message it paid for (as it did when officer built these).
|
||||
const last = session.messages[session.messages.length - 1];
|
||||
if (last?.role === 'assistant' && !last.cost) last.cost = msg.cost;
|
||||
break;
|
||||
}
|
||||
|
||||
case 'error':
|
||||
case 'stopped':
|
||||
session.isGenerating = false;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
function createClaudeMessageHandler(sessionId: string, model: string) {
|
||||
return (msg: TurnMessage, seq?: number): void => {
|
||||
const session = sessionManager.getSession(sessionId);
|
||||
if (!session) return;
|
||||
foldIntoSession(session, msg, model);
|
||||
sendToClient(session.ws as ServerWebSocket<WSData> | null, msg, seq);
|
||||
};
|
||||
}
|
||||
|
||||
// ── OpenCode: officer still translates and persists ──
|
||||
|
||||
// Unchanged from before the split, and still correct for OpenCode: that sidecar reports raw ChatEvents,
|
||||
// so officer does the translation and owns the durable write. Moving it is the `sidecars-opencode` branch.
|
||||
function createEventHandler(sessionId: string, model: string, cwd: string) {
|
||||
return async (event: ChatEvent): Promise<void> => {
|
||||
const session = sessionManager.getSession(sessionId);
|
||||
@@ -298,13 +374,23 @@ function createEventHandler(sessionId: string, model: string, cwd: string) {
|
||||
|
||||
case 'task:started': {
|
||||
// Background task launched (run_in_background / Monitor). Independent of turn state.
|
||||
await emitToSession(sessionId, { type: 'task:started', taskId: event.taskId, description: event.description, taskType: event.taskType });
|
||||
await emitToSession(sessionId, {
|
||||
type: 'task:started',
|
||||
taskId: event.taskId,
|
||||
description: event.description,
|
||||
taskType: event.taskType,
|
||||
});
|
||||
break;
|
||||
}
|
||||
|
||||
case 'task:notification': {
|
||||
// Background task finished — arrives AFTER the turn's 'result'; the whole point of the fix.
|
||||
await emitToSession(sessionId, { type: 'task:notification', taskId: event.taskId, status: event.status, summary: event.summary });
|
||||
await emitToSession(sessionId, {
|
||||
type: 'task:notification',
|
||||
taskId: event.taskId,
|
||||
status: event.status,
|
||||
summary: event.summary,
|
||||
});
|
||||
break;
|
||||
}
|
||||
}
|
||||
@@ -401,7 +487,7 @@ async function handleClaudeCodeChat(
|
||||
|
||||
session.isGenerating = true;
|
||||
|
||||
const onEvent = createEventHandler(sessionId, model, cwd);
|
||||
const onMessage = createClaudeMessageHandler(sessionId, model);
|
||||
|
||||
try {
|
||||
if (!session._claudeKill) {
|
||||
@@ -416,7 +502,7 @@ async function handleClaudeCodeChat(
|
||||
cwd,
|
||||
model,
|
||||
resumeSessionId: msg.resumeSessionId,
|
||||
onEvent,
|
||||
onMessage,
|
||||
});
|
||||
session.piProcess = sessionId as any;
|
||||
session._claudeKill = handle.kill;
|
||||
@@ -572,7 +658,7 @@ async function handleStop(ws: ServerWebSocket<WSData>): Promise<void> {
|
||||
if (isClaudeModel(session.model)) {
|
||||
// Interrupt the current turn but KEEP the persistent session alive (background tasks + the
|
||||
// warm worker survive). Full teardown is 'disconnect' → deleteSession → _claudeKill.
|
||||
void sidecar.interruptClaude(sessionId, session.email);
|
||||
void sidecar.interruptClaude(sessionId);
|
||||
logger.info('Interrupted Claude Code turn via sidecar (session stays warm)', { sessionId });
|
||||
} else {
|
||||
session._claudeKill?.(); // OpenCode: abort the turn via the stored handle
|
||||
|
||||
@@ -8,7 +8,7 @@ import { getTaskByDirName } from './task-files';
|
||||
import { getHomeDir } from '../../data-path';
|
||||
import { resolveBaseCwd } from '../chat/websocket';
|
||||
import { sendClaudeCodeStreaming } from '../../channels/send-claude-code';
|
||||
import type { ChatEvent, MessageCost } from '../chat/types';
|
||||
import type { TurnMessage, MessageCost } from '../chat/types';
|
||||
import * as jobManager from './pipeline-job-manager';
|
||||
|
||||
const DEFAULT_MODEL = 'claude-code';
|
||||
@@ -148,23 +148,26 @@ async function runAgenticStep({
|
||||
fn();
|
||||
};
|
||||
|
||||
const onEvent = (event: ChatEvent) => {
|
||||
// The agent sidecar now hands over finished turn messages rather than raw parser events, so this is a
|
||||
// re-label onto the pipeline's own event stream. The `seq` is ignored: pipeline steps have their own
|
||||
// durable record (the job's events), not the chat cursor.
|
||||
const onMessage = (msg: TurnMessage) => {
|
||||
if (abortSignal.aborted) return;
|
||||
lastActivity = Date.now();
|
||||
|
||||
switch (event.type) {
|
||||
case 'delta':
|
||||
emit({ type: 'assistant:delta', text: event.text, stepIndex, iterationLabel });
|
||||
switch (msg.type) {
|
||||
case 'assistant:delta':
|
||||
emit({ type: 'assistant:delta', text: msg.text, stepIndex, iterationLabel });
|
||||
break;
|
||||
case 'text':
|
||||
emit({ type: 'assistant:text', text: event.text, stepIndex, iterationLabel });
|
||||
case 'assistant:text':
|
||||
emit({ type: 'assistant:text', text: msg.text, stepIndex, iterationLabel });
|
||||
break;
|
||||
case 'tool:start':
|
||||
emit({
|
||||
type: 'tool:start',
|
||||
toolCallId: event.toolCallId,
|
||||
toolName: event.toolName,
|
||||
toolInput: event.toolInput,
|
||||
toolCallId: msg.toolCallId,
|
||||
toolName: msg.toolName,
|
||||
toolInput: msg.toolInput,
|
||||
stepIndex,
|
||||
iterationLabel,
|
||||
});
|
||||
@@ -172,25 +175,29 @@ async function runAgenticStep({
|
||||
case 'tool:result':
|
||||
emit({
|
||||
type: 'tool:result',
|
||||
toolCallId: event.toolCallId,
|
||||
output: event.output,
|
||||
isError: event.isError,
|
||||
toolCallId: msg.toolCallId,
|
||||
output: msg.output,
|
||||
isError: msg.isError,
|
||||
stepIndex,
|
||||
iterationLabel,
|
||||
});
|
||||
break;
|
||||
case 'result':
|
||||
case 'result': {
|
||||
const cost = msg.cost;
|
||||
settle(() => {
|
||||
cleanup?.();
|
||||
resolve(event.cost);
|
||||
resolve(cost);
|
||||
});
|
||||
break;
|
||||
case 'error':
|
||||
}
|
||||
case 'error': {
|
||||
const message = msg.message;
|
||||
settle(() => {
|
||||
cleanup?.();
|
||||
reject(new Error(event.message));
|
||||
reject(new Error(message));
|
||||
});
|
||||
break;
|
||||
}
|
||||
case 'stopped':
|
||||
settle(() => {
|
||||
cleanup?.();
|
||||
@@ -236,7 +243,8 @@ async function runAgenticStep({
|
||||
sessionKey: sessionId,
|
||||
cwd,
|
||||
model,
|
||||
onEvent,
|
||||
durable: false,
|
||||
onMessage,
|
||||
});
|
||||
cleanup = handle.kill;
|
||||
} catch (err) {
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { logger } from '@@/api/chat/logger';
|
||||
import type { MessageCost, ChatEvent } from '@@/api/chat/types';
|
||||
import type { MessageCost, TurnMessage } from '@@/api/chat/types';
|
||||
import * as sidecar from '@@/sidecar-registry';
|
||||
|
||||
type ClaudeCodeParams = {
|
||||
@@ -18,8 +18,8 @@ type ClaudeCodeResult = {
|
||||
cost: MessageCost;
|
||||
};
|
||||
|
||||
export function clearClaudeCodeSession(sessionKey: string, email?: string): void {
|
||||
sidecar.clearClaudeSession(sessionKey, email);
|
||||
export function clearClaudeCodeSession(sessionKey: string): void {
|
||||
sidecar.clearClaudeSession(sessionKey);
|
||||
}
|
||||
|
||||
export async function sendClaudeCode(params: ClaudeCodeParams): Promise<ClaudeCodeResult> {
|
||||
@@ -38,7 +38,9 @@ type ClaudeCodeStreamingParams = {
|
||||
cwd?: string;
|
||||
model?: string;
|
||||
resumeSessionId?: string;
|
||||
onEvent: (event: ChatEvent) => void;
|
||||
durable?: boolean;
|
||||
// Finished turn messages, already committed by the sidecar; `seq` is the cursor id to deliver them under.
|
||||
onMessage: (msg: TurnMessage, seq?: number) => void;
|
||||
};
|
||||
|
||||
type ClaudeCodeStreamingHandle = {
|
||||
@@ -46,22 +48,22 @@ type ClaudeCodeStreamingHandle = {
|
||||
};
|
||||
|
||||
export async function sendClaudeCodeStreaming(params: ClaudeCodeStreamingParams): Promise<ClaudeCodeStreamingHandle> {
|
||||
const { onEvent, ...spawnParams } = params;
|
||||
const { onMessage, ...spawnParams } = params;
|
||||
|
||||
logger.info('Claude Code streaming exec (via sidecar)', { sessionKey: params.sessionKey });
|
||||
|
||||
// Session-scoped subscription. The persistent session outlives the turn, so background task events
|
||||
// (task:notification) arrive AFTER 'result' — do NOT unsubscribe on a terminal turn event; only on
|
||||
// an explicit kill/teardown (the returned handle, called from deleteSession/disconnect).
|
||||
const unsub = sidecar.onClaudeEvent((sessionKey, event) => {
|
||||
if (sessionKey === params.sessionKey) onEvent(event);
|
||||
const unsub = sidecar.onClaudeMessage((sessionKey, msg, seq) => {
|
||||
if (sessionKey === params.sessionKey) onMessage(msg, seq);
|
||||
});
|
||||
|
||||
await sidecar.spawnClaudeStreaming(spawnParams);
|
||||
|
||||
return {
|
||||
kill: () => {
|
||||
sidecar.killClaude(params.sessionKey, params.email);
|
||||
sidecar.killClaude(params.sessionKey);
|
||||
unsub();
|
||||
},
|
||||
};
|
||||
|
||||
+38
-110
@@ -1,6 +1,4 @@
|
||||
import { resolve } from 'node:path';
|
||||
import type { ServerWebSocket } from 'bun';
|
||||
import type { Subprocess } from 'bun';
|
||||
import type {
|
||||
SidecarCommand,
|
||||
SidecarEvent,
|
||||
@@ -15,7 +13,7 @@ import type {
|
||||
VncSessionInfo,
|
||||
} from './sidecar/protocol';
|
||||
import type { SidecarRegistration } from './sidecar/registration-protocol';
|
||||
import type { ChatEvent } from './api/chat/types';
|
||||
import type { ChatEvent, TurnMessage } from './api/chat/types';
|
||||
|
||||
// ── Types ──
|
||||
|
||||
@@ -113,13 +111,6 @@ function findSidecarByCapability(cap: string): RegisteredSidecar | undefined {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
function findSidecarByName(name: string): RegisteredSidecar | undefined {
|
||||
for (const sc of sidecars.values()) {
|
||||
if (sc.name === name) return sc;
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
// ── Event dispatch ──
|
||||
|
||||
function dispatchEvent(msg: SidecarEvent | PtyEvent) {
|
||||
@@ -195,82 +186,27 @@ function sendFireToSidecar(sc: RegisteredSidecar, cmd: SidecarCommand | PtyComma
|
||||
sc.ws.send(JSON.stringify(cmd));
|
||||
}
|
||||
|
||||
// ── On-demand Claude sidecar spawning ──
|
||||
// ── Waiting for a sidecar to appear ──
|
||||
|
||||
const USER_INSTANCE_SCRIPT = resolve(import.meta.dir, 'sidecar/claude/user-instance.ts');
|
||||
const SIDECAR_SPAWN_TIMEOUT_MS = 15_000;
|
||||
// Officer no longer spawns any sidecar; PM2 owns every one of them. The only thing left to handle is
|
||||
// startup order — PM2 brings `officer` and its peers up together, so the first request after a boot can
|
||||
// arrive a beat before the sidecar has finished dialling in. Wait briefly rather than failing the
|
||||
// request. (This replaces ~77 lines of spawn-and-poll: `ensureClaudeSidecar`,
|
||||
// `spawnAndWaitForRegistration`, and the per-email `claudeProcs`/`claudeSpawnWaiters` maps.)
|
||||
const CAPABILITY_WAIT_MS = 15_000;
|
||||
const CAPABILITY_POLL_MS = 100;
|
||||
|
||||
const claudeProcs = new Map<string, Subprocess>();
|
||||
const claudeSpawnWaiters = new Map<string, Promise<RegisteredSidecar>>();
|
||||
|
||||
async function ensureClaudeSidecar(email: string): Promise<RegisteredSidecar> {
|
||||
const name = `claude:${email}`;
|
||||
|
||||
// Already registered?
|
||||
const existing = findSidecarByName(name);
|
||||
async function waitForCapability(cap: string, timeoutMs = CAPABILITY_WAIT_MS): Promise<RegisteredSidecar> {
|
||||
const existing = findSidecarByCapability(cap);
|
||||
if (existing) return existing;
|
||||
|
||||
// Already spawning?
|
||||
const waiter = claudeSpawnWaiters.get(email);
|
||||
if (waiter) return waiter;
|
||||
|
||||
// Spawn and wait for registration
|
||||
const promise = spawnAndWaitForRegistration(email, name);
|
||||
claudeSpawnWaiters.set(email, promise);
|
||||
try {
|
||||
return await promise;
|
||||
} finally {
|
||||
claudeSpawnWaiters.delete(email);
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
await Bun.sleep(CAPABILITY_POLL_MS);
|
||||
const sc = findSidecarByCapability(cap);
|
||||
if (sc) return sc;
|
||||
}
|
||||
}
|
||||
|
||||
async function spawnAndWaitForRegistration(email: string, name: string): Promise<RegisteredSidecar> {
|
||||
const proxySecret = await getProxySecret();
|
||||
const proxyPort = process.env.ANTHROPIC_PROXY_PORT ?? '5051';
|
||||
|
||||
const env: Record<string, string> = {
|
||||
...(process.env as Record<string, string>),
|
||||
CLAUDE_USER_EMAIL: email,
|
||||
ANTHROPIC_BASE_URL: `http://127.0.0.1:${proxyPort}`,
|
||||
ANTHROPIC_API_KEY: proxySecret,
|
||||
};
|
||||
|
||||
const proc = Bun.spawn(['bun', 'run', USER_INSTANCE_SCRIPT], {
|
||||
env,
|
||||
stdout: 'inherit',
|
||||
stderr: 'inherit',
|
||||
});
|
||||
|
||||
claudeProcs.set(email, proc);
|
||||
|
||||
// Clean up on exit
|
||||
proc.exited.then(() => {
|
||||
claudeProcs.delete(email);
|
||||
});
|
||||
|
||||
// Wait for the sidecar to register
|
||||
return new Promise<RegisteredSidecar>((resolve, reject) => {
|
||||
const timeout = setTimeout(() => {
|
||||
unsub();
|
||||
reject(new Error(`Claude sidecar for ${email} failed to register within ${SIDECAR_SPAWN_TIMEOUT_MS}ms`));
|
||||
}, SIDECAR_SPAWN_TIMEOUT_MS);
|
||||
|
||||
// Poll for registration (the sidecar connects via WebSocket and registerSidecar is called)
|
||||
const check = () => {
|
||||
const sc = findSidecarByName(name);
|
||||
if (sc) {
|
||||
clearTimeout(timeout);
|
||||
clearInterval(interval);
|
||||
resolve(sc);
|
||||
}
|
||||
};
|
||||
const interval = setInterval(check, 50);
|
||||
|
||||
const unsub = () => {
|
||||
clearTimeout(timeout);
|
||||
clearInterval(interval);
|
||||
};
|
||||
});
|
||||
throw new Error(`No sidecar with capability "${cap}" registered within ${timeoutMs}ms`);
|
||||
}
|
||||
|
||||
// ── Public API ──
|
||||
@@ -303,10 +239,14 @@ export function getProxySecretSync(): string {
|
||||
return cachedState?.proxySecret ?? '';
|
||||
}
|
||||
|
||||
// ── Claude Code (per-user routing) ──
|
||||
// ── Claude Code (the `officer-agent` sidecar, capability 'claude') ──
|
||||
|
||||
// Single-user platform, so there is exactly one agent sidecar and it is found by capability like every
|
||||
// other one. The `email` on the params is still passed through to the sidecar — it needs it to resolve
|
||||
// paths — but officer no longer uses it to *locate* anything.
|
||||
|
||||
export async function spawnClaude(params: ClaudeSpawnParams): Promise<ClaudeCodeResult> {
|
||||
const sc = await ensureClaudeSidecar(params.email);
|
||||
const sc = await waitForCapability('claude');
|
||||
const res = await sendCommandToSidecar(sc, { type: 'claude:spawn', id: nextId(), params }, LONG_TIMEOUT_MS);
|
||||
if (res.type === 'claude:result') return res.result;
|
||||
if (res.type === 'claude:error') throw new Error(res.error);
|
||||
@@ -314,46 +254,34 @@ export async function spawnClaude(params: ClaudeSpawnParams): Promise<ClaudeCode
|
||||
}
|
||||
|
||||
export async function spawnClaudeStreaming(params: ClaudeSpawnStreamingParams): Promise<void> {
|
||||
const sc = await ensureClaudeSidecar(params.email);
|
||||
const sc = await waitForCapability('claude');
|
||||
const res = await sendCommandToSidecar(sc, { type: 'claude:spawn-streaming', id: nextId(), params });
|
||||
if (res.type === 'claude:spawned') return;
|
||||
if (res.type === 'claude:error') throw new Error(res.error);
|
||||
throw new Error('Unexpected response');
|
||||
}
|
||||
|
||||
export function killClaude(sessionKey: string, email: string): void {
|
||||
const sc = findSidecarByName(`claude:${email}`);
|
||||
if (sc) sendFireToSidecar(sc, { type: 'claude:kill', id: nextId(), sessionKey });
|
||||
export function killClaude(sessionKey: string): void {
|
||||
sendFire('claude', { type: 'claude:kill', id: nextId(), sessionKey });
|
||||
}
|
||||
|
||||
// Interrupt the current turn but keep the persistent session warm (the "stop" button).
|
||||
export function interruptClaude(sessionKey: string, email: string): void {
|
||||
const sc = findSidecarByName(`claude:${email}`);
|
||||
if (sc) sendFireToSidecar(sc, { type: 'claude:interrupt', id: nextId(), sessionKey });
|
||||
export function interruptClaude(sessionKey: string): void {
|
||||
sendFire('claude', { type: 'claude:interrupt', id: nextId(), sessionKey });
|
||||
}
|
||||
|
||||
export function clearClaudeSession(sessionKey: string, email?: string): void {
|
||||
if (email) {
|
||||
const sc = findSidecarByName(`claude:${email}`);
|
||||
if (sc) sendFireToSidecar(sc, { type: 'claude:clear-session', id: nextId(), sessionKey });
|
||||
} else {
|
||||
// Broadcast to all claude sidecars (used when email is not available)
|
||||
for (const sc of sidecars.values()) {
|
||||
if (sc.capabilities.includes('claude')) {
|
||||
sendFireToSidecar(sc, { type: 'claude:clear-session', id: nextId(), sessionKey });
|
||||
}
|
||||
}
|
||||
}
|
||||
export function clearClaudeSession(sessionKey: string): void {
|
||||
sendFire('claude', { type: 'claude:clear-session', id: nextId(), sessionKey });
|
||||
}
|
||||
|
||||
export function onClaudeEvent(handler: (sessionKey: string, event: ChatEvent) => void): () => void {
|
||||
return on('claude:event', (msg) => {
|
||||
if (msg.type === 'claude:event') {
|
||||
handler(
|
||||
(msg as SidecarEvent & { type: 'claude:event' }).sessionKey,
|
||||
(msg as SidecarEvent & { type: 'claude:event' }).event,
|
||||
);
|
||||
}
|
||||
// Turn output arrives finished and already durable: the agent translated it and committed it to
|
||||
// chat_session_events, and `seq` is its cursor id there. Officer relays it — it no longer builds or
|
||||
// persists chat messages for this harness.
|
||||
export function onClaudeMessage(handler: (sessionKey: string, msg: TurnMessage, seq?: number) => void): () => void {
|
||||
return on('claude:message', (ev) => {
|
||||
if (ev.type !== 'claude:message') return;
|
||||
const msg = ev as SidecarEvent & { type: 'claude:message' };
|
||||
handler(msg.sessionKey, msg.msg, msg.seq);
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,182 @@
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { createSessionLogStore, type Delivery, type EventWriter } from './session-log';
|
||||
import type { ChatEvent, MessageCost } from '../../api/chat/types';
|
||||
|
||||
const COST: MessageCost = { inputTokens: 1, outputTokens: 2, totalUSD: 0.0001 };
|
||||
|
||||
/**
|
||||
* A writer that finishes its appends in the WORST possible order — the last one first — so the test
|
||||
* fails unless the store serialises commits itself. `lastSeq` starts empty (a fresh session) unless a
|
||||
* seed is given.
|
||||
*/
|
||||
function reverseOrderWriter(opts: { seed?: number; failOn?: (n: number) => boolean } = {}) {
|
||||
const pending: Array<{ resolve: (seq: number) => void; reject: (err: Error) => void }> = [];
|
||||
let next = 100;
|
||||
const writes: Array<{ sessionId: string; type: string; prevSeq?: number }> = [];
|
||||
|
||||
const writer: EventWriter = {
|
||||
lastSeq: async () => opts.seed,
|
||||
append: (sessionId, msg) => {
|
||||
writes.push({ sessionId, type: msg.type, prevSeq: msg.prevSeq });
|
||||
return new Promise<number>((resolve, reject) => pending.push({ resolve, reject }));
|
||||
},
|
||||
};
|
||||
|
||||
// Settle everything queued so far, newest first.
|
||||
function settleAll() {
|
||||
const batch = pending.splice(0).reverse();
|
||||
for (const p of batch) {
|
||||
const n = next++;
|
||||
if (opts.failOn?.(n)) p.reject(new Error(`write ${n} failed`));
|
||||
else p.resolve(n);
|
||||
}
|
||||
}
|
||||
|
||||
return { writer, settleAll, writes, pendingCount: () => pending.length };
|
||||
}
|
||||
|
||||
function collector() {
|
||||
const got: Delivery[] = [];
|
||||
return { got, deliver: (d: Delivery) => got.push(d) };
|
||||
}
|
||||
|
||||
/** Let the store's promise chain drain, settling writes as they queue up. */
|
||||
async function drain(settleAll: () => void, pendingCount: () => number) {
|
||||
for (let i = 0; i < 50; i++) {
|
||||
await Promise.resolve();
|
||||
if (pendingCount() > 0) settleAll();
|
||||
}
|
||||
}
|
||||
|
||||
const TURN: ChatEvent[] = [
|
||||
{ type: 'delta', text: 'Let me look.' },
|
||||
{ type: 'tool:start', toolCallId: 't1', toolName: 'Read', toolInput: { file: 'a.ts' } },
|
||||
{ type: 'tool:result', toolCallId: 't1', output: 'ok', isError: false },
|
||||
{ type: 'delta', text: 'Found it.' },
|
||||
{ type: 'result', cost: COST },
|
||||
];
|
||||
|
||||
describe('createSessionLogStore', () => {
|
||||
test('deliveries keep push order even when the writes finish backwards', async () => {
|
||||
const { got, deliver } = collector();
|
||||
const { writer, settleAll, pendingCount } = reverseOrderWriter();
|
||||
const store = createSessionLogStore(deliver, writer);
|
||||
|
||||
for (const e of TURN) store.push('s1', e);
|
||||
await drain(settleAll, pendingCount);
|
||||
|
||||
expect(got.map((d) => d.msg.type)).toEqual([
|
||||
'assistant:delta',
|
||||
'assistant:text',
|
||||
'tool:start',
|
||||
'tool:result',
|
||||
'assistant:delta',
|
||||
'assistant:text',
|
||||
'result',
|
||||
]);
|
||||
});
|
||||
|
||||
test('cursors ascend and each durable message chains to the one before it', async () => {
|
||||
const { got, deliver } = collector();
|
||||
const { writer, settleAll, pendingCount } = reverseOrderWriter();
|
||||
const store = createSessionLogStore(deliver, writer);
|
||||
|
||||
for (const e of TURN) store.push('s1', e);
|
||||
await drain(settleAll, pendingCount);
|
||||
|
||||
const durable = got.filter((d) => d.seq !== undefined);
|
||||
expect(durable).toHaveLength(5);
|
||||
|
||||
const seqs = durable.map((d) => d.seq!);
|
||||
expect([...seqs].sort((a, b) => a - b)).toEqual(seqs);
|
||||
|
||||
// First durable message of a fresh session makes no continuity claim; the rest point at their predecessor.
|
||||
expect(durable[0]!.msg.prevSeq).toBeUndefined();
|
||||
for (let i = 1; i < durable.length; i++) {
|
||||
expect(durable[i]!.msg.prevSeq).toBe(durable[i - 1]!.seq);
|
||||
}
|
||||
});
|
||||
|
||||
test('a session that predates this process picks the chain up from the stored cursor', async () => {
|
||||
const { got, deliver } = collector();
|
||||
const { writer, settleAll, pendingCount } = reverseOrderWriter({ seed: 42 });
|
||||
const store = createSessionLogStore(deliver, writer);
|
||||
|
||||
store.push('s1', { type: 'text', text: 'after a restart' });
|
||||
await drain(settleAll, pendingCount);
|
||||
|
||||
expect(got).toHaveLength(1);
|
||||
expect(got[0]!.msg.prevSeq).toBe(42);
|
||||
});
|
||||
|
||||
test('deltas are delivered without a cursor', async () => {
|
||||
const { got, deliver } = collector();
|
||||
const { writer, settleAll, pendingCount } = reverseOrderWriter();
|
||||
const store = createSessionLogStore(deliver, writer);
|
||||
|
||||
store.push('s1', { type: 'delta', text: 'hi' });
|
||||
await drain(settleAll, pendingCount);
|
||||
|
||||
expect(got).toEqual([{ sessionId: 's1', msg: { type: 'assistant:delta', text: 'hi' } }]);
|
||||
});
|
||||
|
||||
test('durable:false skips the writer entirely but still delivers everything', async () => {
|
||||
const { got, deliver } = collector();
|
||||
const { writer, settleAll, writes, pendingCount } = reverseOrderWriter();
|
||||
const store = createSessionLogStore(deliver, writer);
|
||||
|
||||
for (const e of TURN) store.push('job-uuid', e, false);
|
||||
await drain(settleAll, pendingCount);
|
||||
|
||||
expect(writes).toHaveLength(0);
|
||||
expect(got).toHaveLength(7);
|
||||
expect(got.every((d) => d.seq === undefined)).toBe(true);
|
||||
});
|
||||
|
||||
test('a failed write delivers live without a cursor and leaves the chain intact', async () => {
|
||||
const { got, deliver } = collector();
|
||||
// Fail the second successful-cursor slot, i.e. one write in the middle of the turn.
|
||||
const { writer, settleAll, pendingCount } = reverseOrderWriter({ failOn: (n) => n === 101 });
|
||||
const store = createSessionLogStore(deliver, writer);
|
||||
|
||||
store.push('s1', { type: 'text', text: 'one' });
|
||||
await drain(settleAll, pendingCount);
|
||||
store.push('s1', { type: 'text', text: 'two' });
|
||||
await drain(settleAll, pendingCount);
|
||||
store.push('s1', { type: 'text', text: 'three' });
|
||||
await drain(settleAll, pendingCount);
|
||||
|
||||
expect(got.map((d) => d.msg.type)).toEqual(['assistant:text', 'assistant:text', 'assistant:text']);
|
||||
|
||||
const [first, failed, third] = got as [Delivery, Delivery, Delivery];
|
||||
expect(first.seq).toBe(100);
|
||||
// The message the client cannot replay carries neither a cursor nor a continuity claim.
|
||||
expect(failed.seq).toBeUndefined();
|
||||
// ...and the next write chains from the cursor the client actually still holds, not from the hole.
|
||||
expect(third.msg.prevSeq).toBe(first.seq);
|
||||
expect(third.seq).toBe(102);
|
||||
});
|
||||
|
||||
test('sessions are independent, and drop forgets a session', async () => {
|
||||
const { got, deliver } = collector();
|
||||
const { writer, settleAll, pendingCount } = reverseOrderWriter();
|
||||
const store = createSessionLogStore(deliver, writer);
|
||||
|
||||
store.push('a', { type: 'delta', text: 'from-a' });
|
||||
store.push('b', { type: 'delta', text: 'from-b' });
|
||||
store.push('a', { type: 'result', cost: COST });
|
||||
await drain(settleAll, pendingCount);
|
||||
|
||||
// 'a' flushed only its own buffer.
|
||||
const texts = got.filter((d) => d.msg.type === 'assistant:text');
|
||||
expect(texts).toHaveLength(1);
|
||||
expect(texts[0]).toMatchObject({ sessionId: 'a', msg: { text: 'from-a' } });
|
||||
|
||||
// After a drop, a new turn on the same key starts from a clean buffer and re-reads the stored cursor.
|
||||
store.drop('a');
|
||||
got.length = 0;
|
||||
store.push('a', { type: 'result', cost: COST });
|
||||
await drain(settleAll, pendingCount);
|
||||
expect(got.map((d) => d.msg.type)).toEqual(['result']);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,121 @@
|
||||
import type { ChatEvent, TurnMessage } from '../../api/chat/types';
|
||||
import { appendChatEvent, getLastChatEventSeq } from 'officerdb';
|
||||
import { createTurnStream, type TurnOutput } from './turn-stream';
|
||||
|
||||
// The agent sidecar is the writer of record for chat output.
|
||||
//
|
||||
// It used to push raw ChatEvents at officer over the registration socket and let officer translate and
|
||||
// persist them. That socket silently drops when officer is down (`connect.ts:send` — no queue, no
|
||||
// error), so everything the agent produced during a restart was lost: the turn kept running here and
|
||||
// its output went nowhere. Writing to Postgres here instead means an officer restart costs a replay
|
||||
// rather than the output, because the durable record no longer travels over the socket that died.
|
||||
//
|
||||
// Officer still gets every message live — it just gets it already written, with its cursor id attached,
|
||||
// and relays it verbatim.
|
||||
|
||||
export type Delivery = {
|
||||
sessionId: string;
|
||||
msg: TurnMessage;
|
||||
/** The cursor id under which this message is durable. Absent = ephemeral delta, or the write failed. */
|
||||
seq?: number;
|
||||
};
|
||||
|
||||
export type SessionLogStore = {
|
||||
/**
|
||||
* Translate one parser event and commit + deliver whatever it produces, strictly in order.
|
||||
* `durable: false` skips the write (see ClaudeSpawnStreamingParams.durable) — the messages are still
|
||||
* delivered, just without a cursor, because nothing will ever replay them.
|
||||
*/
|
||||
push: (sessionId: string, event: ChatEvent, durable?: boolean) => void;
|
||||
/** Forget a session's buffer and cursor chain (on kill / clear-session). */
|
||||
drop: (sessionId: string) => void;
|
||||
};
|
||||
|
||||
// The durable store, behind an interface so the ordering guarantee below can be tested against a writer
|
||||
// whose writes finish out of order. Defaults to Postgres.
|
||||
export type EventWriter = {
|
||||
append: (sessionId: string, msg: TurnMessage) => Promise<number>;
|
||||
lastSeq: (sessionId: string) => Promise<number | undefined>;
|
||||
};
|
||||
|
||||
const postgresWriter: EventWriter = {
|
||||
append: (sessionId, msg) => appendChatEvent(sessionId, msg),
|
||||
lastSeq: (sessionId) => getLastChatEventSeq(sessionId),
|
||||
};
|
||||
|
||||
type SessionLog = {
|
||||
stream: ReturnType<typeof createTurnStream>;
|
||||
/** Serialises commits so cursor ids are assigned in the order the events actually arrived. */
|
||||
tail: Promise<void>;
|
||||
lastSeq: number | undefined;
|
||||
resolvedLastSeq: boolean;
|
||||
};
|
||||
|
||||
export function createSessionLogStore(
|
||||
deliver: (d: Delivery) => void,
|
||||
writer: EventWriter = postgresWriter,
|
||||
): SessionLogStore {
|
||||
const logs = new Map<string, SessionLog>();
|
||||
|
||||
function logFor(sessionId: string): SessionLog {
|
||||
let log = logs.get(sessionId);
|
||||
if (!log) {
|
||||
log = {
|
||||
stream: createTurnStream(sessionId),
|
||||
tail: Promise.resolve(),
|
||||
lastSeq: undefined,
|
||||
resolvedLastSeq: false,
|
||||
};
|
||||
logs.set(sessionId, log);
|
||||
}
|
||||
return log;
|
||||
}
|
||||
|
||||
async function commit(sessionId: string, log: SessionLog, out: TurnOutput, durable: boolean): Promise<void> {
|
||||
// Deltas are live-only, but still go through the queue: a delta that overtook the `assistant:text`
|
||||
// or `tool:start` in front of it would make the client commit its stream buffer at the wrong point.
|
||||
if (!out.durable || !durable) {
|
||||
deliver({ sessionId, msg: out.msg });
|
||||
return;
|
||||
}
|
||||
|
||||
// Pick the chain back up after a restart of this process, so `prevSeq` stays meaningful for a
|
||||
// session that started before it. Once per session; failure just means no continuity claim.
|
||||
if (!log.resolvedLastSeq) {
|
||||
log.resolvedLastSeq = true;
|
||||
try {
|
||||
log.lastSeq = await writer.lastSeq(sessionId);
|
||||
} catch (err) {
|
||||
console.error(`[agent] could not read last event cursor for ${sessionId}:`, err);
|
||||
}
|
||||
}
|
||||
|
||||
const msg: TurnMessage = log.lastSeq === undefined ? out.msg : { ...out.msg, prevSeq: log.lastSeq };
|
||||
|
||||
try {
|
||||
const seq = await writer.append(sessionId, msg);
|
||||
log.lastSeq = seq;
|
||||
deliver({ sessionId, msg, seq });
|
||||
} catch (err) {
|
||||
// The write IS the durability guarantee, so don't pretend. Deliver live without a cursor: the
|
||||
// client sees the message but won't advance its cursor past something it cannot replay, and the
|
||||
// next successful write chains from the last cursor the client actually holds.
|
||||
console.error(`[agent] failed to persist chat event for ${sessionId}:`, err);
|
||||
deliver({ sessionId, msg: out.msg });
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
push(sessionId, event, durable = true) {
|
||||
const log = logFor(sessionId);
|
||||
// Translation is synchronous and therefore in arrival order; only the commit is queued.
|
||||
for (const out of log.stream.push(event)) {
|
||||
log.tail = log.tail.then(() => commit(sessionId, log, out, durable));
|
||||
}
|
||||
},
|
||||
|
||||
drop(sessionId) {
|
||||
logs.delete(sessionId);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -101,6 +101,31 @@ export async function flushAndSave(): Promise<void> {
|
||||
await saveState();
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the Anthropic proxy secret out of the *proxy* sidecar's state file.
|
||||
*
|
||||
* The proxy (`officer-anthropic-proxy`) and the agent (`officer-agent`) keep separate state — see
|
||||
* `initPaths`: `DATA_PATH/sidecar/` versus `DATA_PATH/<email>/sidecar/` — so the agent cannot reach
|
||||
* the secret through `getState()`. It used to be handed the secret in env by the main server, and
|
||||
* needing that handoff is precisely why the agent had to be spawned by `officer` (and therefore died
|
||||
* with it). Reading it off disk keeps the two processes independent, with the proxy still the only
|
||||
* writer.
|
||||
*
|
||||
* Returns '' when the secret is not on disk yet: `ensureProxySecret` persists through a 30s debounce,
|
||||
* so a brand-new install has a window where the file exists without it. Callers should treat '' as
|
||||
* "retry later" rather than fatal.
|
||||
*/
|
||||
export function readProxySecretFromDisk(): string {
|
||||
try {
|
||||
const proxyStateFile = join(DATA_PATH, 'sidecar', 'claude-state.json');
|
||||
if (!existsSync(proxyStateFile)) return '';
|
||||
const parsed = JSON.parse(readFileSync(proxyStateFile, 'utf-8')) as Partial<PersistedState>;
|
||||
return parsed.proxySecret ?? '';
|
||||
} catch {
|
||||
return '';
|
||||
}
|
||||
}
|
||||
|
||||
// ── Lockfile ──
|
||||
|
||||
export function acquireLock(): boolean {
|
||||
|
||||
@@ -0,0 +1,147 @@
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { createTurnStream } from './turn-stream';
|
||||
import type { ChatEvent, MessageCost } from '../../api/chat/types';
|
||||
|
||||
const SESSION = 'sess-1';
|
||||
const COST: MessageCost = { inputTokens: 10, outputTokens: 20, totalUSD: 0.001 };
|
||||
|
||||
// Feed a whole event sequence through one stream and collect what came out, so the assertions read as
|
||||
// "this turn produced this transcript" rather than per-call plumbing.
|
||||
function run(events: ChatEvent[]) {
|
||||
const stream = createTurnStream(SESSION);
|
||||
const out = events.flatMap((e) => stream.push(e));
|
||||
return {
|
||||
all: out,
|
||||
durable: out.filter((o) => o.durable).map((o) => o.msg),
|
||||
types: out.map((o) => o.msg.type),
|
||||
};
|
||||
}
|
||||
|
||||
describe('createTurnStream', () => {
|
||||
test('deltas are live-only and never durable', () => {
|
||||
const { all, durable } = run([
|
||||
{ type: 'delta', text: 'he' },
|
||||
{ type: 'delta', text: 'llo' },
|
||||
]);
|
||||
expect(all).toHaveLength(2);
|
||||
expect(all.every((o) => o.durable)).toBe(false);
|
||||
expect(durable).toHaveLength(0);
|
||||
});
|
||||
|
||||
test('an explicit text event wins over the deltas that produced it', () => {
|
||||
const { durable } = run([
|
||||
{ type: 'delta', text: 'par' },
|
||||
{ type: 'delta', text: 'tial' },
|
||||
{ type: 'text', text: 'partial and complete' },
|
||||
]);
|
||||
expect(durable).toEqual([{ type: 'assistant:text', text: 'partial and complete' }]);
|
||||
});
|
||||
|
||||
test('a text event with no text falls back to the accumulated buffer', () => {
|
||||
const { durable } = run([
|
||||
{ type: 'delta', text: 'buffered' },
|
||||
{ type: 'text', text: '' },
|
||||
]);
|
||||
expect(durable).toEqual([{ type: 'assistant:text', text: 'buffered' }]);
|
||||
});
|
||||
|
||||
test('an empty text event with an empty buffer produces nothing', () => {
|
||||
expect(run([{ type: 'text', text: '' }]).all).toHaveLength(0);
|
||||
});
|
||||
|
||||
test('the buffer is flushed as one message before a tool call', () => {
|
||||
const { durable } = run([
|
||||
{ type: 'delta', text: 'Let me ' },
|
||||
{ type: 'delta', text: 'check.' },
|
||||
{ type: 'tool:start', toolCallId: 't1', toolName: 'Read', toolInput: { file: 'a.ts' } },
|
||||
]);
|
||||
expect(durable).toEqual([
|
||||
{ type: 'assistant:text', text: 'Let me check.' },
|
||||
{ type: 'tool:start', toolCallId: 't1', toolName: 'Read', toolInput: { file: 'a.ts' } },
|
||||
]);
|
||||
});
|
||||
|
||||
test('the buffer is flushed before the turn result, and result carries the session id', () => {
|
||||
const { durable } = run([
|
||||
{ type: 'delta', text: 'Done.' },
|
||||
{ type: 'result', cost: COST },
|
||||
]);
|
||||
expect(durable).toEqual([
|
||||
{ type: 'assistant:text', text: 'Done.' },
|
||||
{ type: 'result', sessionId: SESSION, cost: COST },
|
||||
]);
|
||||
});
|
||||
|
||||
test('a flushed buffer is not emitted twice', () => {
|
||||
const { durable } = run([
|
||||
{ type: 'delta', text: 'once' },
|
||||
{ type: 'tool:start', toolCallId: 't1', toolName: 'Read', toolInput: {} },
|
||||
{ type: 'tool:result', toolCallId: 't1', output: 'ok', isError: false },
|
||||
{ type: 'result', cost: COST },
|
||||
]);
|
||||
expect(durable.filter((m) => m.type === 'assistant:text')).toEqual([{ type: 'assistant:text', text: 'once' }]);
|
||||
});
|
||||
|
||||
test('deltas after a flush start a fresh block', () => {
|
||||
const { durable } = run([
|
||||
{ type: 'delta', text: 'first' },
|
||||
{ type: 'tool:start', toolCallId: 't1', toolName: 'Read', toolInput: {} },
|
||||
{ type: 'delta', text: 'second' },
|
||||
{ type: 'result', cost: COST },
|
||||
]);
|
||||
expect(durable.filter((m) => m.type === 'assistant:text')).toEqual([
|
||||
{ type: 'assistant:text', text: 'first' },
|
||||
{ type: 'assistant:text', text: 'second' },
|
||||
]);
|
||||
});
|
||||
|
||||
test('a full turn keeps the client-visible order', () => {
|
||||
const { types } = run([
|
||||
{ type: 'delta', text: 'a' },
|
||||
{ type: 'tool:start', toolCallId: 't1', toolName: 'Bash', toolInput: { cmd: 'ls' } },
|
||||
{ type: 'tool:result', toolCallId: 't1', output: 'a.ts', isError: false },
|
||||
{ type: 'delta', text: 'b' },
|
||||
{ type: 'result', cost: COST },
|
||||
]);
|
||||
expect(types).toEqual([
|
||||
'assistant:delta',
|
||||
'assistant:text',
|
||||
'tool:start',
|
||||
'tool:result',
|
||||
'assistant:delta',
|
||||
'assistant:text',
|
||||
'result',
|
||||
]);
|
||||
});
|
||||
|
||||
test('errors and stops are durable, and do not flush a partial answer away', () => {
|
||||
const { durable } = run([
|
||||
{ type: 'delta', text: 'half' },
|
||||
{ type: 'error', message: 'boom' },
|
||||
]);
|
||||
expect(durable).toEqual([{ type: 'error', message: 'boom' }]);
|
||||
expect(run([{ type: 'stopped' }]).durable).toEqual([{ type: 'stopped' }]);
|
||||
});
|
||||
|
||||
test('background task events pass through and are durable — the reason the queue exists', () => {
|
||||
const { durable } = run([
|
||||
{ type: 'result', cost: COST },
|
||||
{ type: 'task:started', taskId: 'bg1', description: 'long job', taskType: 'local_agent' },
|
||||
{ type: 'task:notification', taskId: 'bg1', status: 'completed', summary: 'all good' },
|
||||
]);
|
||||
expect(durable).toEqual([
|
||||
{ type: 'result', sessionId: SESSION, cost: COST },
|
||||
{ type: 'task:started', taskId: 'bg1', description: 'long job', taskType: 'local_agent' },
|
||||
{ type: 'task:notification', taskId: 'bg1', status: 'completed', summary: 'all good' },
|
||||
]);
|
||||
});
|
||||
|
||||
test('streams are independent', () => {
|
||||
const a = createTurnStream('a');
|
||||
const b = createTurnStream('b');
|
||||
a.push({ type: 'delta', text: 'from-a' });
|
||||
b.push({ type: 'delta', text: 'from-b' });
|
||||
expect(a.push({ type: 'result', cost: COST })[0]!.msg).toEqual({ type: 'assistant:text', text: 'from-a' });
|
||||
expect(b.push({ type: 'result', cost: COST })[0]!.msg).toEqual({ type: 'assistant:text', text: 'from-b' });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,101 @@
|
||||
import type { ChatEvent, TurnMessage } from '../../api/chat/types';
|
||||
|
||||
// Translation from the parser's ChatEvent stream to the browser-facing turn messages, moved here from
|
||||
// the main server (`chat/websocket.ts:createEventHandler`). It lives with the process that produces the
|
||||
// stream because it is stateful: `delta` events accumulate into a buffer that has to be flushed as one
|
||||
// `assistant:text` at the next boundary (a tool call, or the end of the turn). A consumer downstream of
|
||||
// a socket that can drop cannot hold that state correctly.
|
||||
//
|
||||
// Pure apart from the buffer, and the buffer is the whole of it — which is what makes it testable.
|
||||
|
||||
export type TurnOutput = {
|
||||
msg: TurnMessage;
|
||||
/** false only for `assistant:delta`: superseded by the `assistant:text` that follows, never persisted. */
|
||||
durable: boolean;
|
||||
};
|
||||
|
||||
export type TurnStream = {
|
||||
push: (event: ChatEvent) => TurnOutput[];
|
||||
};
|
||||
|
||||
export function createTurnStream(sessionId: string): TurnStream {
|
||||
let buffer = '';
|
||||
|
||||
// Emit whatever deltas have accumulated as one complete message. Called at every boundary where the
|
||||
// assistant stops talking, so the transcript holds text blocks rather than a thousand fragments.
|
||||
function flush(): TurnOutput[] {
|
||||
if (!buffer) return [];
|
||||
const text = buffer;
|
||||
buffer = '';
|
||||
return [{ msg: { type: 'assistant:text', text }, durable: true }];
|
||||
}
|
||||
|
||||
function push(event: ChatEvent): TurnOutput[] {
|
||||
switch (event.type) {
|
||||
case 'delta':
|
||||
buffer += event.text;
|
||||
return [{ msg: { type: 'assistant:delta', text: event.text }, durable: false }];
|
||||
|
||||
case 'text': {
|
||||
// An explicit full text block wins over the accumulated deltas that produced it.
|
||||
const text = event.text || buffer;
|
||||
buffer = '';
|
||||
return text ? [{ msg: { type: 'assistant:text', text }, durable: true }] : [];
|
||||
}
|
||||
|
||||
case 'tool:start':
|
||||
return [
|
||||
...flush(),
|
||||
{
|
||||
msg: {
|
||||
type: 'tool:start',
|
||||
toolCallId: event.toolCallId,
|
||||
toolName: event.toolName,
|
||||
toolInput: event.toolInput,
|
||||
},
|
||||
durable: true,
|
||||
},
|
||||
];
|
||||
|
||||
case 'tool:result':
|
||||
return [
|
||||
{
|
||||
msg: { type: 'tool:result', toolCallId: event.toolCallId, output: event.output, isError: event.isError },
|
||||
durable: true,
|
||||
},
|
||||
];
|
||||
|
||||
case 'result':
|
||||
return [...flush(), { msg: { type: 'result', sessionId, cost: event.cost }, durable: true }];
|
||||
|
||||
case 'error':
|
||||
return [{ msg: { type: 'error', message: event.message }, durable: true }];
|
||||
|
||||
case 'stopped':
|
||||
return [{ msg: { type: 'stopped' }, durable: true }];
|
||||
|
||||
case 'task:started':
|
||||
return [
|
||||
{
|
||||
msg: {
|
||||
type: 'task:started',
|
||||
taskId: event.taskId,
|
||||
description: event.description,
|
||||
taskType: event.taskType,
|
||||
},
|
||||
durable: true,
|
||||
},
|
||||
];
|
||||
|
||||
case 'task:notification':
|
||||
return [
|
||||
{
|
||||
msg: { type: 'task:notification', taskId: event.taskId, status: event.status, summary: event.summary },
|
||||
durable: true,
|
||||
},
|
||||
];
|
||||
}
|
||||
}
|
||||
|
||||
return { push };
|
||||
}
|
||||
@@ -2,30 +2,40 @@ import { existsSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs';
|
||||
import { join, resolve } from 'node:path';
|
||||
import { homedir } from 'node:os';
|
||||
import type { SidecarCommand, SidecarEvent } from '../protocol';
|
||||
import { initPaths, loadState, flushAndSave, acquireLock, releaseLock } from './state';
|
||||
import { initPaths, loadState, flushAndSave, acquireLock, releaseLock, readProxySecretFromDisk } from './state';
|
||||
import { createSessionLogStore } from './session-log';
|
||||
import { setMcpConfigPath } from './claude-manager';
|
||||
import * as claudeManager from './claude-manager';
|
||||
import { createSidecarConnector } from '../connect';
|
||||
import { sign } from '../../jwt';
|
||||
import { getUserByEmail, getEmailAccounts } from 'officerdb';
|
||||
import { getUserByEmail, getOwnerUser, getEmailAccounts } from 'officerdb';
|
||||
|
||||
const email = process.env.CLAUDE_USER_EMAIL;
|
||||
if (!email) {
|
||||
console.error('[user-instance] CLAUDE_USER_EMAIL is required');
|
||||
process.exit(1);
|
||||
// PM2 starts this sidecar with no user in its env. Single-user platform, so resolve the owner from the
|
||||
// database rather than being told who to run as by the main server — one less thing that has to come
|
||||
// from `officer` before this process can work. CLAUDE_USER_EMAIL still wins when set, for manual runs.
|
||||
async function resolveOwner() {
|
||||
const explicit = process.env.CLAUDE_USER_EMAIL?.trim();
|
||||
for (;;) {
|
||||
const user = explicit ? await getUserByEmail(explicit) : await getOwnerUser();
|
||||
if (user) return user;
|
||||
// Fresh install: wait for POST /auth/bootstrap instead of exiting into a PM2 restart loop.
|
||||
console.log(`[agent] no ${explicit ? `user "${explicit}"` : 'owner account'} yet — retrying in 5s`);
|
||||
await Bun.sleep(5_000);
|
||||
}
|
||||
}
|
||||
|
||||
const dbUser = await resolveOwner();
|
||||
const email = dbUser.email;
|
||||
|
||||
const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data');
|
||||
const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${process.env.PORT ?? '5000'}`;
|
||||
const OFFICER_API_URL = process.env.OFFICER_API_URL ?? `http://127.0.0.1:${process.env.PORT ?? '9010'}`;
|
||||
// Same officer instance for both, so the fallback port has to agree. It used to default to 5000 for the
|
||||
// WebSocket and 9010 for the REST base, which would have split them apart if PORT were ever unset.
|
||||
const OFFICER_PORT = process.env.PORT ?? '9010';
|
||||
const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${OFFICER_PORT}`;
|
||||
const OFFICER_API_URL = process.env.OFFICER_API_URL ?? `http://127.0.0.1:${OFFICER_PORT}`;
|
||||
const MCP_SERVER_SCRIPT = resolve(import.meta.dir, '../../mcp-tool-server.ts');
|
||||
|
||||
// Mint a long-lived JWT for this user so tools (e.g. gmail) can call back to dev-platform as them
|
||||
const dbUser = await getUserByEmail(email);
|
||||
if (!dbUser) {
|
||||
console.error(`[user-instance] no user found for ${email}`);
|
||||
process.exit(1);
|
||||
}
|
||||
const OFFICER_AUTH_TOKEN = await sign({ id: dbUser.id, email, username: dbUser.username }, '30d');
|
||||
|
||||
// Single-user platform: the owner runs Claude with no isolation — real HOME, real ~/.claude — so
|
||||
@@ -49,7 +59,7 @@ process.env.HOME = homeDir;
|
||||
initPaths(email);
|
||||
|
||||
if (!acquireLock()) {
|
||||
console.error(`[claude:${email}] another instance is already running (lock file exists with live PID)`);
|
||||
console.error(`[agent] another instance is already running for ${email} (lock file exists with live PID)`);
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
@@ -85,6 +95,30 @@ function generateMcpConfig(): string {
|
||||
return join(contextDir, 'mcp-host.json');
|
||||
}
|
||||
|
||||
// ── Anthropic credentials ──
|
||||
|
||||
// The `claude` CLI inherits this process's env (claude-manager spawns with `process.env`), so the proxy
|
||||
// endpoint and secret have to be set here. Officer used to inject both when it spawned this process;
|
||||
// reading them ourselves is what lets this sidecar be a PM2 peer instead of a child of the server.
|
||||
//
|
||||
// Resolved lazily rather than once at boot: PM2 starts the proxy and the agent together, and
|
||||
// `ensureProxySecret` persists on a 30s debounce, so on a first-ever boot the secret can be briefly
|
||||
// absent. Re-checked before every spawn until it lands.
|
||||
const ANTHROPIC_PROXY_PORT = process.env.ANTHROPIC_PROXY_PORT ?? '5051';
|
||||
|
||||
function ensureAnthropicEnv(): void {
|
||||
process.env.ANTHROPIC_BASE_URL ??= `http://127.0.0.1:${ANTHROPIC_PROXY_PORT}`;
|
||||
if (process.env.ANTHROPIC_API_KEY) return;
|
||||
|
||||
const secret = readProxySecretFromDisk();
|
||||
if (secret) {
|
||||
process.env.ANTHROPIC_API_KEY = secret;
|
||||
console.log('[agent] anthropic proxy secret loaded from disk');
|
||||
} else {
|
||||
console.warn('[agent] anthropic proxy secret not on disk yet — retrying before next spawn');
|
||||
}
|
||||
}
|
||||
|
||||
// ── Startup ──
|
||||
|
||||
// The owner runs un-isolated with HOME as their real home, so the generated container CLAUDE.md is
|
||||
@@ -92,8 +126,17 @@ function generateMcpConfig(): string {
|
||||
// terminal `claude` loads too.
|
||||
|
||||
setMcpConfigPath(generateMcpConfig());
|
||||
ensureAnthropicEnv();
|
||||
|
||||
console.log(`[claude:${email}] started (HOME=${homeDir})`);
|
||||
console.log(`[agent] started for ${email} (HOME=${homeDir})`);
|
||||
|
||||
// ── Turn output ──
|
||||
|
||||
// Every message a turn produces is translated, committed to chat_session_events and only then pushed to
|
||||
// officer. `connection` is initialised below, before any command can arrive to invoke this.
|
||||
const sessionLog = createSessionLogStore((d) =>
|
||||
connection.send({ type: 'claude:message', sessionKey: d.sessionId, msg: d.msg, seq: d.seq }),
|
||||
);
|
||||
|
||||
// ── Command handlers ──
|
||||
|
||||
@@ -106,6 +149,7 @@ async function handleCommand(cmd: SidecarCommand, reply: ReplyFn) {
|
||||
break;
|
||||
|
||||
case 'claude:spawn': {
|
||||
ensureAnthropicEnv();
|
||||
try {
|
||||
const result = await claudeManager.spawnClaude(cmd.params);
|
||||
reply({ type: 'claude:result', id: cmd.id, result });
|
||||
@@ -116,24 +160,23 @@ async function handleCommand(cmd: SidecarCommand, reply: ReplyFn) {
|
||||
}
|
||||
|
||||
case 'claude:spawn-streaming': {
|
||||
ensureAnthropicEnv();
|
||||
reply({ type: 'claude:spawned', id: cmd.id, sessionKey: cmd.params.sessionKey });
|
||||
|
||||
const onEvent = (event: import('../../api/chat/types').ChatEvent) => {
|
||||
connection.send({ type: 'claude:event', sessionKey: cmd.params.sessionKey, event });
|
||||
};
|
||||
|
||||
claudeManager.spawnClaudeStreaming(cmd.params, onEvent).catch((err) => {
|
||||
connection.send({
|
||||
type: 'claude:event',
|
||||
sessionKey: cmd.params.sessionKey,
|
||||
event: { type: 'error', message: err instanceof Error ? err.message : String(err) },
|
||||
const { sessionKey, durable = true } = cmd.params;
|
||||
claudeManager
|
||||
.spawnClaudeStreaming(cmd.params, (event) => sessionLog.push(sessionKey, event, durable))
|
||||
.catch((err) => {
|
||||
// Through the log like any other output, so a failure to start is durable and replayable too.
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
sessionLog.push(sessionKey, { type: 'error', message }, durable);
|
||||
});
|
||||
});
|
||||
break;
|
||||
}
|
||||
|
||||
case 'claude:kill':
|
||||
claudeManager.killClaudeSession(cmd.sessionKey);
|
||||
sessionLog.drop(cmd.sessionKey);
|
||||
reply({ type: 'claude:killed', id: cmd.id });
|
||||
break;
|
||||
|
||||
@@ -144,6 +187,7 @@ async function handleCommand(cmd: SidecarCommand, reply: ReplyFn) {
|
||||
|
||||
case 'claude:clear-session':
|
||||
claudeManager.clearSession(cmd.sessionKey);
|
||||
sessionLog.drop(cmd.sessionKey);
|
||||
reply({ type: 'claude:session-cleared', id: cmd.id });
|
||||
break;
|
||||
|
||||
@@ -158,9 +202,12 @@ async function handleCommand(cmd: SidecarCommand, reply: ReplyFn) {
|
||||
|
||||
// ── Connect to API server ──
|
||||
|
||||
// A stable identity, not a per-email one. Officer looks this sidecar up by the 'claude' capability, so
|
||||
// it no longer needs to know which user is running to find it — that was the last thing tying the
|
||||
// registry's claude verbs to an email argument.
|
||||
const connection = createSidecarConnector({
|
||||
apiUrl: `${API_URL}/api/sidecar/register`,
|
||||
name: `claude:${email}`,
|
||||
name: 'agent',
|
||||
capabilities: ['claude'],
|
||||
onCommand(cmd, reply) {
|
||||
handleCommand(cmd as SidecarCommand, reply as ReplyFn);
|
||||
@@ -170,7 +217,7 @@ const connection = createSidecarConnector({
|
||||
// ── Graceful shutdown ──
|
||||
|
||||
async function shutdown(signal: string) {
|
||||
console.log(`[claude:${email}] ${signal} received, saving state...`);
|
||||
console.log(`[agent] ${signal} received, saving state...`);
|
||||
connection.destroy();
|
||||
await flushAndSave();
|
||||
releaseLock();
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { MessageCost, ChatEvent } from '../api/chat/types';
|
||||
import type { MessageCost, ChatEvent, TurnMessage } from '../api/chat/types';
|
||||
|
||||
// ── Envelope ──
|
||||
|
||||
@@ -35,7 +35,10 @@ export type SidecarEvent =
|
||||
| { type: 'proxy:secret'; id: string; secret: string }
|
||||
// Claude Code
|
||||
| { type: 'claude:spawned'; id: string; sessionKey: string }
|
||||
| { type: 'claude:event'; sessionKey: string; event: ChatEvent }
|
||||
// A finished, browser-facing turn message. The agent has already committed it to chat_session_events
|
||||
// and `seq` is its cursor id there; officer relays it verbatim. No `seq` means it is not durable —
|
||||
// an `assistant:delta` (superseded by the text that follows) or a message whose write failed.
|
||||
| { type: 'claude:message'; sessionKey: string; msg: TurnMessage; seq?: number }
|
||||
| { type: 'claude:result'; id: string; result: ClaudeCodeResult }
|
||||
| { type: 'claude:error'; id: string; error: string }
|
||||
| { type: 'claude:killed'; id: string }
|
||||
@@ -93,6 +96,10 @@ export type ClaudeSpawnStreamingParams = {
|
||||
cwd?: string;
|
||||
model?: string;
|
||||
resumeSessionId?: string; // resume this Claude session uuid (from the /chat session list)
|
||||
// Whether turn output should be committed to chat_session_events (default true). A chat session wants
|
||||
// it — that is what survives an officer restart. A pipeline step does not: its sessionKey is a throwaway
|
||||
// uuid no browser will ever replay, and the job's own event log is its record.
|
||||
durable?: boolean;
|
||||
};
|
||||
|
||||
export type ClaudeCodeResult = {
|
||||
|
||||
@@ -141,6 +141,20 @@ export function useChat(initialSessionId?: string, initialModel?: string | null,
|
||||
function handleMessage(data: unknown) {
|
||||
const msg = data as ServerMessage;
|
||||
|
||||
// Continuity check, before the cursor moves. The writer stamps every durable message with the cursor
|
||||
// of the one before it in the same session, so a mismatch against what we last saw means something is
|
||||
// missing — pruned by retention, or a write that failed. Surface it: a conversation that silently
|
||||
// skips a tool call or half an answer reads as the assistant having done something inexplicable.
|
||||
// Only checked once we actually hold a cursor; opening a session from history starts mid-chain by
|
||||
// design (events are swept after 7 days, the transcript itself is not).
|
||||
const prevSeq = (data as { prevSeq?: number }).prevSeq;
|
||||
if (typeof prevSeq === 'number' && cursorRef.current > 0 && prevSeq !== cursorRef.current) {
|
||||
setMessages((prev) => [
|
||||
...prev,
|
||||
{ role: 'error', text: '⚠️ Some output could not be recovered — part of this conversation is missing above.' },
|
||||
]);
|
||||
}
|
||||
|
||||
// Advance the resume cursor for any durable (seq-carrying) event.
|
||||
const seq = (data as { seq?: number }).seq;
|
||||
if (typeof seq === 'number' && seq > cursorRef.current) cursorRef.current = seq;
|
||||
|
||||
Reference in New Issue
Block a user