diff --git a/apps/memos-local-plugin/core/pipeline/ALGORITHMS.md b/apps/memos-local-plugin/core/pipeline/ALGORITHMS.md index a639b9480..386f23398 100644 --- a/apps/memos-local-plugin/core/pipeline/ALGORITHMS.md +++ b/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")`. | diff --git a/apps/memos-local-plugin/core/pipeline/README.md b/apps/memos-local-plugin/core/pipeline/README.md index 57faf5a82..eecfaa31a 100644 --- a/apps/memos-local-plugin/core/pipeline/README.md +++ b/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`.| diff --git a/apps/memos-local-plugin/core/pipeline/memory-core.ts b/apps/memos-local-plugin/core/pipeline/memory-core.ts index 660ea737c..08d6dc5ff 100644 --- a/apps/memos-local-plugin/core/pipeline/memory-core.ts +++ b/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); diff --git a/apps/memos-local-plugin/tests/unit/pipeline/regression-2327-second-message.test.ts b/apps/memos-local-plugin/tests/unit/pipeline/regression-2327-second-message.test.ts new file mode 100644 index 000000000..18bfb9fc9 --- /dev/null +++ b/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(); + }); +});