fix: six confirmed audit findings (token lifecycle, msg-drop, recall scaling)

src/daemon/auth.ts

@@ -52,6 +52,17 @@ export async function verifyToken(
     throw new Error(`Token audience mismatch: expected "${config.audience}"`);
   }
 
+  // Defend locally: the SDK only rejects expiry when `exp` is present, so a
+  // signed token that omits `exp` would never expire. Require a numeric,
+  // not-yet-expired exp. (A 60s skew allowance covers minor clock drift.)
+  const nowSec = Math.floor(Date.now() / 1000);
+  if (typeof identity.exp !== "number" || identity.exp <= 0) {
+    throw new Error("Token missing exp claim");
+  }
+  if (identity.exp <= nowSec - 60) {
+    throw new Error("Token expired");
+  }
+
   return identityToAuthContext(identity);
 }
 
@@ -67,6 +78,7 @@ function identityToAuthContext(identity: VerifiedIdentity): AuthContext {
     delegatedBy: identity.act?.sub,
     accountId: identity.account_id,
     projectId: identity.project_id,
+    exp: identity.exp,
   };
 }
 

src/daemon/memory/engine.ts

@@ -67,6 +67,11 @@ export class MemoryEngine {
   /** FIFO queue of episode IDs awaiting embedding. */
   #embedQueue: string[] = [];
   #embedRunning = false;
+  /** False until the embedder model loads. When the model fails to init
+   * (offline, download hiccup) the engine stays alive in FTS-only mode —
+   * keyword recall, episode persistence and usage tracking all keep working;
+   * only the vector signal is disabled. */
+  #embedderReady = false;
 
   constructor(opts: MemoryEngineOptions) {
     this.#store = opts.store;
@@ -77,7 +82,20 @@ export class MemoryEngine {
   }
 
   async init(): Promise<void> {
-    await this.#embedder.init();
+    // Degrade, don't die: an embedder init failure must NOT take down the
+    // whole engine (which would also disable FTS recall + usage persistence).
+    // Run FTS-only and let recall's vector branch no-op on a null queryVector.
+    try {
+      await this.#embedder.init();
+      this.#embedderReady = true;
+    } catch (err) {
+      this.#embedderReady = false;
+      console.error(
+        `[codeoid/memory] embedder init failed — running in FTS-only mode (no semantic recall): ${
+          err instanceof Error ? err.message : String(err)
+        }`,
+      );
+    }
   }
 
   /** Expose the underlying store so callers that need raw aggregate queries
@@ -111,10 +129,13 @@ export class MemoryEngine {
     const limit = q.limit ?? 8;
     const now = Date.now();
 
-    // Embed the query (blocking — the user's waiting).
-    const queryVector = q.query.trim()
-      ? normalize((await this.#embedder.embed([q.query]))[0]!)
-      : null;
+    // Embed the query (blocking — the user's waiting). Skipped in FTS-only
+    // mode; the vector branch below then no-ops and recall falls back to
+    // keyword + recency + path signals.
+    const queryVector =
+      this.#embedderReady && q.query.trim()
+        ? normalize((await this.#embedder.embed([q.query]))[0]!)
+        : null;
 
     // FTS candidates.
     const ftsRows = this.#store.ftsSearch(q.workspaceId, q.query, this.#ftsK);
@@ -282,6 +303,9 @@ export class MemoryEngine {
   // ── Background embedding worker ───────────────────────────────────────
 
   async #pumpEmbedQueue(): Promise<void> {
+    // In FTS-only mode there's no embedder — leave episodes unembedded
+    // (still persisted + FTS-indexed) rather than throwing per batch.
+    if (!this.#embedderReady) return;
     if (this.#embedRunning) return;
     this.#embedRunning = true;
     try {

src/daemon/memory/store.ts

@@ -70,6 +70,10 @@ export function workspaceIdFromPath(workdir: string): string {
 
 export class SqliteEpisodeStore {
   #db: Database;
+  /** Decoded embedding matrix per workspace, memoized so recall() doesn't
+   * re-read + re-decode every embedding BLOB on each query. Invalidated when
+   * embeddings change (insert-with-embedding / setEmbedding). */
+  #vectorCache = new Map<string, { ids: string[]; vectors: Float32Array[] }>();
 
   constructor(dbPath: string) {
     this.#db = new Database(dbPath, { create: true });
@@ -215,6 +219,10 @@ export class SqliteEpisodeStore {
         episode.createdBy,
       );
 
+    // An insert that already carries an embedding changes the matrix for
+    // its workspace; drop the cached copy so the next recall rebuilds.
+    if (embeddingBuf) this.#vectorCache.delete(episode.workspaceId);
+
     return { ...episode, id };
   }
 
@@ -223,6 +231,10 @@ export class SqliteEpisodeStore {
     this.#db
       .prepare("UPDATE episodes SET embedding = ?, embedding_model = ? WHERE id = ?")
       .run(buf, model, episodeId);
+    // We don't have the workspaceId here; clearing all is correct and cheap
+    // (rebuild happens lazily on the next recall, not here). setEmbedding runs
+    // in background batches, so this doesn't touch the recall hot path.
+    this.#vectorCache.clear();
   }
 
   // ── Turn usage (persistent token/cost tracking) ─────────────────────────
@@ -601,8 +613,13 @@ export class SqliteEpisodeStore {
     return rows;
   }
 
-  /** Build or return the cached (rowid → embedding) matrix for a workspace. */
+  /** Build or return the cached (rowid → embedding) matrix for a workspace.
+   * Memoized: rebuilt only when embeddings change (see #vectorCache
+   * invalidation in insert/setEmbedding), not on every recall. */
   loadVectorMatrix(workspaceId: string): { ids: string[]; vectors: Float32Array[] } {
+    const cached = this.#vectorCache.get(workspaceId);
+    if (cached) return cached;
+
     const rows = this.#db
       .prepare(
         `SELECT id, embedding FROM episodes
@@ -616,7 +633,9 @@ export class SqliteEpisodeStore {
       ids.push(row.id);
       vectors.push(uint8ToFloat32(row.embedding));
     }
-    return { ids, vectors };
+    const matrix = { ids, vectors };
+    this.#vectorCache.set(workspaceId, matrix);
+    return matrix;
   }
 
   /** Look up a recent file read by path + content hash (dedup hit). */

src/daemon/server.ts

@@ -320,6 +320,23 @@ export class DaemonServer {
             return;
           }
 
+          // Enforce token expiry on every message. Without this a
+          // continuously-open socket would honor an expired token forever
+          // (revocation/expiry only took effect on reconnect). On expiry we
+          // close 4003; the client reconnects and re-exchanges its key for a
+          // fresh JWT. (Instant revocation of a still-valid token would need
+          // a periodic re-verify / revocation check — tracked separately.)
+          if (
+            typeof data.auth?.exp === "number" &&
+            data.auth.exp > 0 &&
+            data.auth.exp <= Math.floor(Date.now() / 1000)
+          ) {
+            self.#sockets.delete(data.clientId);
+            self.#manager.disconnectClient(data.clientId);
+            ws.close(4003, "Token expired");
+            return;
+          }
+
           // Authenticated — route through session manager
           const msg = parsed as unknown as ClientMessage;
           const client: AttachedClient = {

src/daemon/session-manager.ts

@@ -62,6 +62,21 @@ function normalizeWorkdir(input: string): string | null {
   }
 }
 
+/** Eager-resume bounds. Resume runs before the daemon listens, and each
+ * session's full transcript is read into memory — so an unbounded resume can
+ * block startup or OOM. Cap to the newest-N sessions and stop past a deadline;
+ * the rest stay on disk (loadable on a future restart with a higher cap). */
+const RESUME_MAX_SESSIONS = 50;
+const RESUME_DEADLINE_MS = 20_000;
+
+/** Sort key for resume ordering: most-recently-active first. Falls back to
+ * createdAt, then 0, so a malformed timestamp never throws. */
+function resumeSortKey(m: { lastActivityAt?: string; createdAt?: string }): number {
+  const t = m.lastActivityAt ?? m.createdAt ?? "";
+  const n = Date.parse(t);
+  return Number.isFinite(n) ? n : 0;
+}
+
 export class SessionManager {
   #sessions = new Map<string, Session>();
   #store: Store;
@@ -97,10 +112,25 @@ export class SessionManager {
    * Rebuilds in-memory session objects and scrollback buffers.
    */
   async resumeSessions(): Promise<number> {
-    const metas = await this.#transcriptStore.loadAllMeta();
+    const allMetas = await this.#transcriptStore.loadAllMeta();
+    // Newest-first by last activity so the cap keeps the most relevant
+    // sessions when there are more than RESUME_MAX_SESSIONS on disk.
+    const sorted = [...allMetas].sort(
+      (a, b) => resumeSortKey(b) - resumeSortKey(a),
+    );
+    const capped = sorted.slice(0, RESUME_MAX_SESSIONS);
+    const deadline = Date.now() + RESUME_DEADLINE_MS;
     let resumed = 0;
-
-    for (const meta of metas) {
+    let skippedDeadline = 0;
+
+    for (let i = 0; i < capped.length; i++) {
+      // Time-box: a few huge transcripts shouldn't wedge startup. Stop and
+      // leave the remainder on disk rather than blocking the daemon listen.
+      if (Date.now() > deadline) {
+        skippedDeadline = capped.length - i;
+        break;
+      }
+      const meta = capped[i]!;
       try {
         const session = new Session({
           name: meta.sessionName,
@@ -144,6 +174,16 @@ export class SessionManager {
       }
     }
 
+    const droppedCap = sorted.length - capped.length;
+    if (droppedCap > 0 || skippedDeadline > 0) {
+      console.warn(
+        `[codeoid] resume: restored ${resumed} of ${sorted.length} session(s); ` +
+          `${droppedCap} left over the ${RESUME_MAX_SESSIONS}-session cap, ` +
+          `${skippedDeadline} skipped past the ${RESUME_DEADLINE_MS}ms deadline ` +
+          `(still on disk; loadable on a future restart).`,
+      );
+    }
+
     return resumed;
   }
 

src/daemon/session.ts

@@ -1036,10 +1036,18 @@ export class Session {
     // emits events at its own pace.
     const query$ = this.#query;
     const ac = this.#abortController;
+    // Loop-local snapshot of the input queue. The teardown `finally` below
+    // must only clear the queue/task slots if a *newer* loop hasn't already
+    // replaced them: an un-awaited interrupt() followed by a fast send() can
+    // build a fresh loop before this finally runs, and without these identity
+    // guards we'd null the new loop's queue/task and silently drop the next
+    // message (the next #pushUserMessage would throw into a swallowed catch).
+    const queue$ = this.#inputQueue;
     // Set when a resume fails because the backing Claude Code conversation
     // doesn't exist — drives the post-teardown replay below.
     let recoverContent: string | null = null;
-    this.#consumerTask = (async () => {
+    let selfTask: Promise<void> | null = null;
+    selfTask = this.#consumerTask = (async () => {
       try {
         for await (const msg of query$) {
           this.#handleAgentMessage(msg);
@@ -1094,9 +1102,12 @@ export class Session {
         this.#chunker?.onTurnEnd();
         if (this.#query === query$) this.#query = null;
         if (this.#abortController === ac) this.#abortController = null;
-        this.#inputQueue?.close();
-        this.#inputQueue = null;
-        this.#consumerTask = null;
+        // Always close OUR queue (idempotent if interrupt already did), but
+        // only clear the slot/task if a newer loop hasn't taken over — see
+        // the queue$ snapshot comment above.
+        queue$?.close();
+        if (this.#inputQueue === queue$) this.#inputQueue = null;
+        if (this.#consumerTask === selfTask) this.#consumerTask = null;
         // Resolve any pending tool approvals — they're awaiting
         // canUseTool callbacks that will never fire on a torn-down
         // SDK loop. Without this, the client-side `await

src/protocol/types.ts

@@ -1202,6 +1202,10 @@ export interface AuthContext {
   accountId: string;
   /** Project ID */
   projectId: string;
+  /** Token expiry (Unix seconds). Carried so the daemon can reject an
+   * expired token on a long-lived connection instead of trusting the
+   * handshake forever. 0/undefined means the token carried no exp. */
+  exp?: number;
 }
 
 /**

web/src/lib/ws.test.ts

@@ -168,6 +168,42 @@ describe("CodeoidClient", () => {
     expect(MockWS.instances.length).toBeGreaterThan(afterFirstRetry);
   });
 
+  it("re-exchanges the token via getToken on each (re)connect", async () => {
+    // Regression: a reconnect after the initial JWT expired used to replay
+    // the dead token forever. With a getToken supplier each connect mints a
+    // fresh token.
+    let n = 0;
+    const c = new CodeoidClient({
+      url: "ws://x",
+      token: "static-fallback",
+      getToken: async () => `fresh-${++n}`,
+    });
+    const ws1 = await connectClient(c);
+    expect(ws1.parsed[0]).toMatchObject({ token: "fresh-1" });
+
+    ws1.drop();
+    await flush();
+    const ws2 = MockWS.last();
+    expect(ws2).not.toBe(ws1);
+    ws2.open();
+    await flush();
+    // The reconnect carried a freshly-minted token, not the stale one.
+    expect(ws2.parsed[0]).toMatchObject({ token: "fresh-2" });
+    ws2.recv(AUTH_OK);
+  });
+
+  it("falls back to the static token when getToken throws", async () => {
+    const c = new CodeoidClient({
+      url: "ws://x",
+      token: "static-fallback",
+      getToken: async () => {
+        throw new Error("no stored key");
+      },
+    });
+    const ws = await connectClient(c);
+    expect(ws.parsed[0]).toMatchObject({ token: "static-fallback" });
+  });
+
   it("does not spawn duplicate reconnect loops on force-reconnect", async () => {
     const c = new CodeoidClient({ url: "ws://x", token: "t" });
     const ws1 = await connectClient(c);

web/src/lib/ws.ts

@@ -28,7 +28,13 @@ export type ClientStatus =
 
 export interface ConnectOptions {
   url: string; // ws://host:port — daemon's WS endpoint
+  /** Initial access token used for the first handshake (and the fallback
+   * if `getToken` is absent or fails). */
   token: string;
+  /** Optional fresh-token supplier, called on every (re)connect's open so a
+   * reconnect after the initial JWT expired re-exchanges the stored key for
+   * a new JWT instead of replaying the dead token forever. */
+  getToken?: () => Promise<string>;
   /** Bounded reconnect attempts; once exhausted we land in `failed`. */
   maxAttempts?: number;
   /** Logger for transport-level diagnostics. */
@@ -58,6 +64,8 @@ const HEARTBEAT_TIMEOUT_MS = 8_000;
 
 export class CodeoidClient {
   #opts: ConnectOptions;
+  /** Current access token; refreshed via opts.getToken on each connect. */
+  #token: string;
   #ws: WebSocket | null = null;
   #pending = new Map<string, PendingRequest>();
   #statusHandlers = new Set<StatusHandler>();
@@ -75,6 +83,7 @@ export class CodeoidClient {
 
   constructor(opts: ConnectOptions) {
     this.#opts = opts;
+    this.#token = opts.token;
   }
 
   /** Subscribe to status changes. Returns an unsubscribe fn. */
@@ -351,11 +360,28 @@ export class CodeoidClient {
       this.#ws = ws;
 
       let authResolved = false;
-      const onOpen = () => {
+      const onOpen = async () => {
+        // Refresh the token before the handshake so a reconnect after the
+        // prior JWT expired mints a fresh one (the daemon now closes 4003 on
+        // an expired token; without this we'd replay the dead token forever).
+        if (this.#opts.getToken) {
+          try {
+            this.#token = await this.#opts.getToken();
+          } catch (err) {
+            this.#log("warn", "getToken failed; using last token", { err });
+          }
+        }
+        // We may have awaited above — bail if a newer connect superseded
+        // this socket meanwhile.
+        if (this.#ws !== ws) return;
         // Daemon requires the first frame to carry an auth token. Shape
         // matches the existing Rust client: `{ type: "auth", token }`.
         // (Daemon only checks the `token` field; `type` is ignored.)
-        ws.send(JSON.stringify({ type: "auth", token: this.#opts.token }));
+        try {
+          ws.send(JSON.stringify({ type: "auth", token: this.#token }));
+        } catch (err) {
+          this.#log("warn", "auth send failed (socket closed during refresh)", { err });
+        }
       };
       const onMessage = (ev: MessageEvent<string>) => {
         let msg: DaemonMessage;