fix(agent-core-v2): roll back failed session initialization

.changeset/clean-session-init-failures.md

@@ -0,0 +1,6 @@
+---
+"@moonshot-ai/agent-core-v2": patch
+"@moonshot-ai/kimi-code": patch
+---
+
+Release failed session initializations, hide half-initialized handles, and block overlapping same-ID attempts from writing the same session files.

packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts

@@ -21,8 +21,10 @@
  * summary is resolved through the read model — falling back to a disk read +
  * backfill on a cold miss. Writes (create / archive / metadata update) keep the
  * read model warm via `SessionMetadata`; new sessions that have not been
- * mirrored yet are simply a cold miss and backfilled on first read. The legacy
- * N+1 path remains as the flag-off fallback — and as the runtime fallback when
+ * mirrored yet are simply a cold miss and backfilled on first read. Persisted
+ * metadata remains the membership source, so a cached row is ignored after its
+ * state document disappears. The legacy N+1 path remains as the flag-off
+ * fallback — and as the runtime fallback when
  * the query store reports `storage.locked` (another process holds the writer
  * lock): the first lock warns once and disables the read model for the rest of
  * the process lifetime.
@@ -193,7 +195,14 @@ export class FileSessionIndex implements ISessionIndex {
 
   private async getFromReadModel(id: string): Promise<SessionSummary | undefined> {
     const cached = await this.queryStore.get<SessionSummary>(SESSION_COLLECTION, id);
-    if (cached !== undefined) return cached;
+    if (cached !== undefined && typeof cached.workspaceId === 'string') {
+      const summary = await this.readSummary(cached.workspaceId, id);
+      if (summary !== undefined) {
+        if (cached.id === id && cached.createdAt === summary.createdAt) return cached;
+        await this.queryStore.put(SESSION_COLLECTION, id, summary);
+        return summary;
+      }
+    }
     // Cold miss: locate the session on disk, then read + backfill.
     for (const workspaceId of await this.listWorkspaceIds()) {
       if (!(await this.hasSession(workspaceId, id))) continue;
@@ -240,11 +249,17 @@ export class FileSessionIndex implements ISessionIndex {
     sessionId: string,
   ): Promise<SessionSummary | undefined> {
     const cached = await this.queryStore.get<SessionSummary>(SESSION_COLLECTION, sessionId);
-    if (cached !== undefined) return cached;
     const summary = await this.readSummary(workspaceId, sessionId);
-    if (summary !== undefined) {
-      await this.queryStore.put(SESSION_COLLECTION, sessionId, summary);
+    if (summary === undefined) return undefined;
+    if (
+      cached !== undefined &&
+      cached.id === sessionId &&
+      cached.workspaceId === workspaceId &&
+      cached.createdAt === summary.createdAt
+    ) {
+      return cached;
     }
+    await this.queryStore.put(SESSION_COLLECTION, sessionId, summary);
     return summary;
   }
 

packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycle.ts

@@ -107,26 +107,29 @@ export interface ISessionLifecycleService {
   create(opts: CreateSessionOptions): Promise<ISessionScopeHandle>;
   /**
    * Return the live handle for `sessionId`, or `undefined` when it is not open.
-   * A session whose cold {@link resume} is still in flight is intentionally NOT
-   * returned — its main agent has not finished restore + replay, so the handle
-   * is half-initialized. Callers that must obtain the handle should
-   * `await resume(sessionId)` instead. This invisibility is a service
-   * invariant, not caller discipline: every read path (`get` / {@link list} /
-   * {@link resume}) agrees a resuming session is not yet observable.
+   * A session whose create, fork, or cold {@link resume} initialization has not
+   * published its metadata, MCP readiness, and any required main-agent restore
+   * is intentionally NOT returned. Once that core state is ready, the handle
+   * is visible while creation hooks finish, unless an explicit close/archive
+   * has taken ownership of teardown. Callers that must wait for publication
+   * should `await resume(sessionId)` instead.
    */
   get(sessionId: string): ISessionScopeHandle | undefined;
   /**
-   * Snapshot of every fully-initialized live session. Excludes sessions still
-   * mid-{@link resume} for the same reason as {@link get}.
+   * Snapshot of every published live session. Excludes sessions whose core
+   * initialization has not reached the publication point described by
+   * {@link get}.
    */
   list(): readonly ISessionScopeHandle[];
   /**
    * Load a persisted session into the live scope tree and restore its main
-   * agent from the persisted wire log. Returns the existing handle when the
-   * session is already live (a no-op in that case — live agents are never
-   * re-restored). Returns `undefined` when the session is unknown to the index
-   * or neither the persisted session summary nor the workspace registry can
-   * provide a workdir (mirrors the cold-source limitation of `fork`).
+   * agent from the persisted wire log. Returns the existing published handle
+   * when the session is already live, and waits for an unpublished same-id
+   * initialization to publish or fail before retrying the lookup. An
+   * initializing session already claimed by close/archive is unavailable.
+   * Returns `undefined` when the session is unknown to the index or neither the
+   * persisted session summary nor the workspace registry can provide a workdir
+   * (mirrors the cold-source limitation of `fork`).
    *
    * Lets the read edges (snapshot / messages) serve cold sessions — created by
    * a previous process or by v1 — without requiring a prior `create` in this