Fix #2327: memmy-agent 第二条消息固定报错 session not found

apps/memos-local-plugin/core/pipeline/ALGORITHMS.md

@@ -204,7 +204,7 @@ Mapping conventions:
 | `onTurnStart`             | Retrieval throws                  | Log error, return empty `InjectionPacket`.           |
 | `onTurnEnd`               | No open episode for session       | Throw `Error("no open episode for session ...")`.    |
 | `onTurnEnd`               | Current open episode is closed    | Throw `Error("episode ... is not open")`.            |
-| `closeSession`            | Session unknown                   | `MemosError("session_not_found")`.                   |
+| `closeSession`            | Session unknown                   | No-op (idempotent); mirrors closeEpisode semantics.  |
 | `closeEpisode`            | Episode unknown                   | `MemosError("episode_not_found")`.                   |
 | `retireSkill`             | Skill unknown                     | `MemosError("skill_not_found")`.                     |
 | Post-shutdown calls       | Any method                        | `MemosError("already_shut_down")`.                   |

apps/memos-local-plugin/core/pipeline/README.md

@@ -165,7 +165,7 @@ Implements the adapter contract 1:1. A few translation details:
 | `init()`               | No-op flag flip. Stays idempotent for adapters that re-arm the core.   |
 | `shutdown()`           | Drains the pipeline, closes the DB (if `onShutdown` provided).         |
 | `openSession`          | Delegates to `sessionManager.openSession`.                             |
-| `closeSession`         | Throws `session_not_found` when missing.                               |
+| `closeSession`         | Idempotent; unknown session ids are a no-op (mirrors closeEpisode).    |
 | `openEpisode`          | Calls `startEpisode(sessionId, userMessage="")`.                       |
 | `closeEpisode`         | Idempotent; returns silently if already closed.                        |
 | `onTurnStart`          | Wraps the orchestrator's `InjectionPacket` into a `RetrievalResultDTO`.|

apps/memos-local-plugin/core/pipeline/memory-core.ts

@@ -2135,10 +2135,16 @@ export function createMemoryCore(
     ensureLive();
     const existing = handle.sessionManager.getSession(sessionId);
     if (!existing) {
-      throw new MemosError(
-        "session_not_found",
-        `session not found: ${sessionId}`,
-      );
+      // Idempotent: closing an already-unknown session is a no-op. Adapters
+      // and the memmy-agent bridge can race the lifecycle (host `/new`
+      // followed by a fresh user message, or a duplicate `session.close`
+      // fire from `on_session_end`) — a hard throw here surfaced to the
+      // user as "session not found" on the second message even though the
+      // session was correctly re-opened by the next turn. Mirror the
+      // `session.closed` bus event which is emit-and-forget.
+      log.debug("closeSession.unknown_session_ignored", { sessionId });
+      turnStartApiLogBySession.delete(sessionId);
+      return;
     }
     handle.sessionManager.closeSession(sessionId, "client");
     turnStartApiLogBySession.delete(sessionId);

apps/memos-local-plugin/tests/unit/pipeline/regression-2327-second-message.test.ts

@@ -0,0 +1,161 @@
+/**
+ * Regression test for #2327 — memmy-agent "second message" session_not_found.
+ *
+ * Pattern: the host fires `session.close` (e.g. /new or an explicit session
+ * end) and then immediately starts a new turn. Because of clock skew, a race,
+ * or an eager `on_session_end`, `closeSession` can be called on an id the
+ * orchestrator has already evicted from its live map. Previously that threw
+ * `MemosError("session_not_found")` which surfaced to the user.  The fix makes
+ * `closeSession` idempotent — unknown sessions are silently ignored — so the
+ * subsequent `onTurnStart` or `openEpisode` can reopen the session normally.
+ */
+
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import type { AgentKind, SessionId } from "../../../agent-contract/dto.js";
+import {
+  createMemoryCore,
+  createPipeline,
+} from "../../../core/pipeline/index.js";
+import { rootLogger } from "../../../core/logger/index.js";
+import { DEFAULT_CONFIG } from "../../../core/config/defaults.js";
+import { resolveHome } from "../../../core/config/paths.js";
+import { makeTmpDb, type TmpDbHandle } from "../../helpers/tmp-db.js";
+import { fakeEmbedder } from "../../helpers/fake-embedder.js";
+
+const AGENT: AgentKind = "openclaw";
+
+let db: TmpDbHandle | null = null;
+
+describe("memory-core / regression #2327: second-message session_not_found", () => {
+  beforeEach(async () => {
+    db = await makeTmpDb();
+  });
+
+  afterEach(async () => {
+    if (db) {
+      await db.cleanup();
+      db = null;
+    }
+  });
+
+  function buildCore() {
+    const pipeline = createPipeline({
+      agent: AGENT,
+      home: resolveHome(AGENT, "/tmp/memos-mc-test"),
+      config: DEFAULT_CONFIG,
+      db: db!.db,
+      repos: db!.repos,
+      llm: null,
+      reflectLlm: null,
+      embedder: fakeEmbedder({ dimensions: 384 }),
+      log: rootLogger.child({ channel: "test.regress-2327" }),
+      namespace: { agentKind: AGENT, profileId: "main" },
+      now: () => Date.now(),
+    });
+    return createMemoryCore(
+      pipeline,
+      resolveHome(AGENT, "/tmp/memos-mc-test"),
+      "test",
+    );
+  }
+
+  it("closeSession on an already-closed session does NOT throw (idempotent)", async () => {
+    const core = buildCore();
+    await core.init();
+
+    const sid = await core.openSession({ agent: AGENT, sessionId: "se_regress_2327" as SessionId });
+
+    const ep = await core.openEpisode({ sessionId: sid, userMessage: "first message" });
+    await core.closeEpisode(ep);
+    await core.closeSession(sid);
+
+    // Second call — session no longer in the live map; must not throw.
+    await expect(core.closeSession(sid)).resolves.toBeUndefined();
+
+    await core.shutdown();
+  });
+
+  it("openEpisode succeeds on second message after closeSession was called (fix for #2327)", async () => {
+    const core = buildCore();
+    await core.init();
+
+    const sid = await core.openSession({ agent: AGENT, sessionId: "se_regress_2327" as SessionId });
+
+    // First message — normal path.
+    const ep1 = await core.openEpisode({ sessionId: sid, userMessage: "first message" });
+    await core.closeEpisode(ep1);
+
+    // Host fires session.close (e.g. /new command or adapter lifecycle).
+    await core.closeSession(sid);
+
+    // Second message arrives.  Adapter re-opens the session first (normal
+    // path) — this must succeed even though closeSession already ran.
+    const sid2 = await core.openSession({ agent: AGENT, sessionId: sid });
+    expect(sid2).toBe(sid);
+
+    const ep2 = await core.openEpisode({ sessionId: sid, userMessage: "second message" });
+    expect(ep2).toBeTruthy();
+    expect(ep2).not.toBe(ep1);
+
+    await core.shutdown();
+  });
+
+  it("openEpisode succeeds even when adapter skips re-openSession after close (defensive path)", async () => {
+    // Some adapter implementations (or races) may call openEpisode without an
+    // intervening openSession.  ensureSession inside the orchestrator handles
+    // this by reopening from the DB row; the bug was that closeSession threw
+    // before this path was reached, making the second turn fatal.
+    const core = buildCore();
+    await core.init();
+
+    const sid = await core.openSession({ agent: AGENT, sessionId: "se_regress_2327b" as SessionId });
+
+    const ep1 = await core.openEpisode({ sessionId: sid, userMessage: "first" });
+    await core.closeEpisode(ep1);
+    await core.closeSession(sid);
+
+    // Deliberately skip openSession — openEpisode must still succeed because
+    // onTurnStart / startEpisode re-opens via ensureSession internally.
+    const ep2 = await core.openEpisode({ sessionId: sid, userMessage: "second" });
+    expect(ep2).toBeTruthy();
+
+    await core.shutdown();
+  });
+
+  it("onTurnStart on a closed session does NOT throw (ensureSession reopens)", async () => {
+    const core = buildCore();
+    await core.init();
+
+    const sid = await core.openSession({ agent: AGENT, sessionId: "se_regress_2327c" as SessionId });
+
+    // First turn — normal.
+    const first = await core.onTurnStart({
+      agent: AGENT,
+      sessionId: sid,
+      userText: "first message",
+      ts: Date.now(),
+    });
+    expect(first.query.sessionId).toBe(sid);
+
+    await core.onTurnEnd({
+      agent: AGENT,
+      sessionId: first.query.sessionId ?? sid,
+      episodeId: first.query.episodeId!,
+      agentText: "ok",
+      toolCalls: [],
+      ts: Date.now(),
+    });
+    await core.closeSession(sid);
+
+    // Second turn — ensureSession must reopen; must not throw.
+    const second = await core.onTurnStart({
+      agent: AGENT,
+      sessionId: sid,
+      userText: "second message",
+      ts: Date.now(),
+    });
+    expect(second.query.sessionId).toBe(sid);
+
+    await core.shutdown();
+  });
+});