diff --git a/.changeset/release-session-event-stream-resources.md b/.changeset/release-session-event-stream-resources.md
new file mode 100644
index 0000000000..45599ca3fb
--- /dev/null
+++ b/.changeset/release-session-event-stream-resources.md
@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Release event-stream resources when sessions close or are archived.
diff --git a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts
index 14e44cdfb5..672ad95ba2 100644
--- a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts
+++ b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts
@@ -25,8 +25,10 @@
  *      an atomic `getSnapshotState` for the snapshot route.
  *
  * A session is activated (journaling starts) on first `subscribe` /
- * `getSnapshotState` / `getCursor` and stays active for the process lifetime so
- * the journal is continuous from first activation onward.
+ * `getSnapshotState` / `getCursor`. Concurrent activations share one owner;
+ * closing or archiving the session releases its listeners and journal handle.
+ * A later resume reopens the same journal so its `{seq, epoch}` stays
+ * continuous across live-owner lifetimes.
  */
 
 import type {
@@ -133,14 +135,20 @@ const GLOBAL_SESSION_ID = '__global__';
 async function disposeSessionState(state: SessionState): Promise<void> {
   for (const d of state.lifecycleDisposables) d.dispose();
   for (const d of state.agentDisposables.values()) d.dispose();
+  await state.queue;
+  state.targets.clear();
   await state.journal.close();
 }
 
 export class SessionEventBroadcaster {
   private readonly sessions = new Map<string, SessionState>();
+  private readonly initializations = new Map<string, Promise<SessionState | undefined>>();
+  private readonly retirements = new Map<string, Promise<void>>();
   private readonly maxBufferSize: number;
   private readonly coreEventSubscription: IDisposable;
+  private readonly sessionLifecycleSubscriptions: readonly IDisposable[];
   private closed = false;
+  private closePromise: Promise<void> | undefined;
 
   constructor(
     private readonly opts: {
@@ -154,6 +162,15 @@ export class SessionEventBroadcaster {
     this.coreEventSubscription = opts.core.accessor
       .get(IEventService)
       .subscribe((event) => this.onCoreEvent(event));
+    const lifecycle = opts.core.accessor.get(ISessionLifecycleService);
+    this.sessionLifecycleSubscriptions = [
+      lifecycle.onDidCloseSession(({ sessionId }) => {
+        this.releaseSessionState(sessionId);
+      }),
+      lifecycle.onDidArchiveSession(({ sessionId }) => {
+        this.releaseSessionState(sessionId);
+      }),
+    ];
   }
 
   /** Subscribe a connection to a session's stream (activates the session). */
@@ -270,10 +287,16 @@ export class SessionEventBroadcaster {
     return watermark;
   }
 
-  async close(): Promise<void> {
-    if (this.closed) return;
+  close(): Promise<void> {
+    this.closePromise ??= this.closeOnce();
+    return this.closePromise;
+  }
+
+  private async closeOnce(): Promise<void> {
     this.closed = true;
     this.coreEventSubscription.dispose();
+    for (const subscription of this.sessionLifecycleSubscriptions) subscription.dispose();
+    await Promise.allSettled([...this.initializations.values(), ...this.retirements.values()]);
     for (const state of this.sessions.values()) {
       await disposeSessionState(state);
     }
@@ -281,58 +304,152 @@ export class SessionEventBroadcaster {
   }
 
   private async ensureState(sessionId: string): Promise<SessionState | undefined> {
-    if (this.closed) return undefined;
-    let state = this.sessions.get(sessionId);
-    if (state !== undefined) return state;
+    while (!this.closed) {
+      const retirement = this.retirements.get(sessionId);
+      if (retirement !== undefined) {
+        await retirement;
+        continue;
+      }
+      const state = this.sessions.get(sessionId);
+      if (state !== undefined) return state;
+      let initialization = this.initializations.get(sessionId);
+      if (initialization === undefined) {
+        initialization = this.initializeSessionState(sessionId);
+        this.initializations.set(sessionId, initialization);
+      }
+      let initialized: SessionState | undefined;
+      try {
+        initialized = await initialization;
+      } finally {
+        if (this.initializations.get(sessionId) === initialization) {
+          this.initializations.delete(sessionId);
+        }
+      }
+      const release = this.retirements.get(sessionId);
+      if (release === undefined) return initialized;
+      await release;
+      if (this.closed) return undefined;
+      if (this.opts.core.accessor.get(ISessionLifecycleService).get(sessionId) === undefined) {
+        return undefined;
+      }
+    }
+    return undefined;
+  }
 
-    const session = this.opts.core.accessor.get(ISessionLifecycleService).get(sessionId);
+  private async initializeSessionState(sessionId: string): Promise<SessionState | undefined> {
+    const lifecycle = this.opts.core.accessor.get(ISessionLifecycleService);
+    const session = lifecycle.get(sessionId);
     if (session === undefined) return undefined;
 
     const journal = await SessionEventJournal.open(
       sessionJournalPath(this.opts.eventsDir, sessionId),
       this.opts.logger,
     );
-    if (this.closed) {
+    if (
+      this.closed ||
+      this.retirements.has(sessionId) ||
+      lifecycle.get(sessionId) !== session
+    ) {
       await journal.close();
       return undefined;
     }
-    const activity = session.accessor.get(ISessionActivityService);
-    state = {
-      sessionId,
-      journal,
-      tracker: new InFlightTurnTracker(),
-      roster: new SubagentRosterTracker(),
-      activity,
-      lastStatus: activity.status(),
-      tail: [],
-      targets: new Map(),
-      queue: Promise.resolve(),
-      agentDisposables: new Map(),
-      lifecycleDisposables: [],
-      knownInteractions: new Map(),
-    };
-    this.sessions.set(sessionId, state);
+    let state: SessionState | undefined;
     try {
+      const activity = session.accessor.get(ISessionActivityService);
+      state = {
+        sessionId,
+        journal,
+        tracker: new InFlightTurnTracker(),
+        roster: new SubagentRosterTracker(),
+        activity,
+        lastStatus: activity.status(),
+        tail: [],
+        targets: new Map(),
+        queue: Promise.resolve(),
+        agentDisposables: new Map(),
+        lifecycleDisposables: [],
+        knownInteractions: new Map(),
+      };
       this.attachAgents(sessionId, session, state);
       this.attachInteractions(sessionId, session, state);
     } catch (error) {
-      this.sessions.delete(sessionId);
-      await disposeSessionState(state);
+      if (state === undefined) await journal.close();
+      else await disposeSessionState(state);
       if (error instanceof Error && error.message === 'InstantiationService has been disposed') return undefined;
       throw error;
     }
+    if (state === undefined) return undefined;
+    if (
+      this.closed ||
+      this.retirements.has(sessionId) ||
+      lifecycle.get(sessionId) !== session
+    ) {
+      await disposeSessionState(state);
+      return undefined;
+    }
+    this.sessions.set(sessionId, state);
     return state;
   }
 
-  private async ensureGlobalState(): Promise<SessionState> {
-    let state = this.sessions.get(GLOBAL_SESSION_ID);
+  private releaseSessionState(sessionId: string): void {
+    const previousRetirement = this.retirements.get(sessionId);
+    const initialization = this.initializations.get(sessionId);
+    const activeState = this.sessions.get(sessionId);
+    const retirement = (async () => {
+      if (previousRetirement !== undefined) {
+        await previousRetirement;
+        return;
+      }
+      let state = activeState;
+      if (state === undefined && initialization !== undefined) {
+        try {
+          state = await initialization;
+        } catch {
+          return;
+        }
+      }
+      if (state === undefined) return;
+      if (activeState !== undefined && this.sessions.get(sessionId) !== state) return;
+      await disposeSessionState(state);
+      if (this.sessions.get(sessionId) === state) this.sessions.delete(sessionId);
+    })();
+    this.retirements.set(sessionId, retirement);
+    const finishRetirement = (): void => {
+      if (this.retirements.get(sessionId) === retirement) {
+        this.retirements.delete(sessionId);
+      }
+    };
+    void retirement.then(finishRetirement, finishRetirement);
+  }
+
+  private async ensureGlobalState(): Promise<SessionState | undefined> {
+    if (this.closed) return undefined;
+    const state = this.sessions.get(GLOBAL_SESSION_ID);
     if (state !== undefined) return state;
+    let initialization = this.initializations.get(GLOBAL_SESSION_ID);
+    if (initialization === undefined) {
+      initialization = this.initializeGlobalState();
+      this.initializations.set(GLOBAL_SESSION_ID, initialization);
+    }
+    try {
+      return await initialization;
+    } finally {
+      if (this.initializations.get(GLOBAL_SESSION_ID) === initialization) {
+        this.initializations.delete(GLOBAL_SESSION_ID);
+      }
+    }
+  }
 
+  private async initializeGlobalState(): Promise<SessionState | undefined> {
     const journal = await SessionEventJournal.open(
       sessionJournalPath(this.opts.eventsDir, GLOBAL_SESSION_ID),
       this.opts.logger,
     );
-    state = {
+    if (this.closed) {
+      await journal.close();
+      return undefined;
+    }
+    const state: SessionState = {
       sessionId: GLOBAL_SESSION_ID,
       journal,
       tracker: new InFlightTurnTracker(),
@@ -410,6 +527,13 @@ export class SessionEventBroadcaster {
 
   private async dispatchGlobal(event: Event): Promise<void> {
     const state = await this.ensureGlobalState();
+    if (
+      state === undefined ||
+      this.closed ||
+      this.sessions.get(GLOBAL_SESSION_ID) !== state
+    ) {
+      return;
+    }
     state.queue = state.queue
       .then(() => this.dispatch(state, event, isVolatileEventType(event.type)))
       .catch((error: unknown) => this.logDispatchDropped(state.sessionId, event.type, error));
@@ -435,6 +559,13 @@ export class SessionEventBroadcaster {
       throw error;
     }
     if (state === undefined) return;
+    if (
+      this.closed ||
+      this.retirements.has(sessionId) ||
+      this.sessions.get(sessionId) !== state
+    ) {
+      return;
+    }
     state.queue = state.queue
       .then(() => this.dispatch(state, event, isVolatileEventType(event.type)))
       .catch((error: unknown) => this.logDispatchDropped(state.sessionId, event.type, error));
@@ -447,8 +578,8 @@ export class SessionEventBroadcaster {
       state.agentDisposables.set(handle.id, this.attachAgent(sessionId, handle));
     };
     for (const handle of agents.list()) subscribeAgent(handle);
+    state.lifecycleDisposables.push(agents.onDidCreate((handle) => subscribeAgent(handle)));
     state.lifecycleDisposables.push(
-      agents.onDidCreate((handle) => subscribeAgent(handle)),
       agents.onDidDispose((agentId) => {
         const d = state.agentDisposables.get(agentId);
         if (d !== undefined) {
@@ -602,6 +733,8 @@ export class SessionEventBroadcaster {
           }
         }
       }),
+    );
+    state.lifecycleDisposables.push(
       interactions.onDidResolve(({ id, response }) => {
         const kind = state.knownInteractions.get(id);
         if (kind === undefined) return;
diff --git a/packages/kap-server/test/sessionEventBroadcaster.test.ts b/packages/kap-server/test/sessionEventBroadcaster.test.ts
index 3d1cc822fe..76af7340ad 100644
--- a/packages/kap-server/test/sessionEventBroadcaster.test.ts
+++ b/packages/kap-server/test/sessionEventBroadcaster.test.ts
@@ -16,6 +16,7 @@ import {
   IEventBus,
   IEventService,
   ISessionActivity,
+  ISessionIndex,
   ISessionInteractionService,
   ISessionLifecycleService,
   IWireService,
@@ -28,7 +29,10 @@ import {
   type BroadcastTarget,
   SessionEventBroadcaster,
 } from '../src/transport/ws/v1/sessionEventBroadcaster';
-import type { EventEnvelope } from '../src/transport/ws/v1/sessionEventJournal';
+import {
+  type EventEnvelope,
+  SessionEventJournal,
+} from '../src/transport/ws/v1/sessionEventJournal';
 
 // ---------------------------------------------------------------------------
 // Fakes
@@ -36,10 +40,15 @@ import type { EventEnvelope } from '../src/transport/ws/v1/sessionEventJournal';
 
 class FakeAgentBus {
   private handlers: Array<(e: AgentEvent) => void> = [];
+  disposeCount = 0;
+  get subscriberCount(): number {
+    return this.handlers.length;
+  }
   subscribe(handler: (e: AgentEvent) => void) {
     this.handlers.push(handler);
     return {
       dispose: () => {
+        this.disposeCount += 1;
         const i = this.handlers.indexOf(handler);
         if (i >= 0) this.handlers.splice(i, 1);
       },
@@ -71,10 +80,17 @@ class FakeAgentHandle {
   readonly bus = new FakeAgentBus();
   readonly accessor;
   private readonly services = new Map<unknown, unknown>();
+  failNextAccess = false;
   constructor(readonly id: string) {
     this.services.set(IEventBus, this.bus);
     this.accessor = {
-      get: (token: unknown) => this.services.get(token),
+      get: (token: unknown) => {
+        if (this.failNextAccess) {
+          this.failNextAccess = false;
+          throw new Error('agent bus unavailable');
+        }
+        return this.services.get(token);
+      },
     };
   }
   set(token: unknown, service: unknown): void {
@@ -88,8 +104,12 @@ class FakeLifecycle {
   /** Real interaction kernel — served at the session accessor. */
   readonly interactions = new SessionInteractionService();
   baseStatus: 'idle' | 'running' = 'idle';
+  onStatus: (() => void) | undefined;
   readonly activity = {
     status: () => {
+      const onStatus = this.onStatus;
+      this.onStatus = undefined;
+      onStatus?.();
       if (this.interactions.listPending('approval').length > 0) return 'awaiting_approval';
       if (this.interactions.listPending('question').length > 0) return 'awaiting_question';
       return this.baseStatus;
@@ -106,11 +126,21 @@ class FakeLifecycle {
   }
   onDidCreate(h: (h: IScopeHandle) => void) {
     this.createHandlers.push(h);
-    return { dispose: () => {} };
+    return {
+      dispose: () => {
+        const index = this.createHandlers.indexOf(h);
+        if (index >= 0) this.createHandlers.splice(index, 1);
+      },
+    };
   }
   onDidDispose(h: (id: string) => void) {
     this.disposeHandlers.push(h);
-    return { dispose: () => {} };
+    return {
+      dispose: () => {
+        const index = this.disposeHandlers.indexOf(h);
+        if (index >= 0) this.disposeHandlers.splice(index, 1);
+      },
+    };
   }
   addAgent(id: string): FakeAgentHandle {
     const handle = new FakeAgentHandle(id);
@@ -125,27 +155,84 @@ class FakeLifecycle {
   }
 }
 
-function makeCore(sessions: Map<string, FakeLifecycle>, eventBus = new FakeEventBus()): Scope {
+class FakeSessionLifecycleEvents {
+  private readonly closeHandlers: Array<(event: { sessionId: string }) => void> = [];
+  private readonly archiveHandlers: Array<(event: { sessionId: string }) => void> = [];
+  failGet = false;
+
+  readonly onDidCloseSession = (handler: (event: { sessionId: string }) => void) => {
+    this.closeHandlers.push(handler);
+    return {
+      dispose: () => {
+        const index = this.closeHandlers.indexOf(handler);
+        if (index >= 0) this.closeHandlers.splice(index, 1);
+      },
+    };
+  };
+
+  readonly onDidArchiveSession = (handler: (event: { sessionId: string }) => void) => {
+    this.archiveHandlers.push(handler);
+    return {
+      dispose: () => {
+        const index = this.archiveHandlers.indexOf(handler);
+        if (index >= 0) this.archiveHandlers.splice(index, 1);
+      },
+    };
+  };
+
+  close(sessionId: string): void {
+    for (const handler of [...this.closeHandlers]) handler({ sessionId });
+  }
+
+  archive(sessionId: string): void {
+    for (const handler of [...this.archiveHandlers]) handler({ sessionId });
+  }
+}
+
+function makeCore(
+  sessions: Map<string, FakeLifecycle>,
+  eventBus = new FakeEventBus(),
+  lifecycleEvents = new FakeSessionLifecycleEvents(),
+): Scope {
+  const scopeHandles = new Map<
+    string,
+    { lifecycle: FakeLifecycle; handle: IScopeHandle }
+  >();
+  const sessionLifecycle = {
+    get: (sid: string) => {
+      if (lifecycleEvents.failGet) throw new Error('InstantiationService has been disposed');
+      const lifecycle = sessions.get(sid);
+      if (lifecycle === undefined) {
+        scopeHandles.delete(sid);
+        return undefined;
+      }
+      const cached = scopeHandles.get(sid);
+      if (cached?.lifecycle === lifecycle) return cached.handle;
+      const sessionAccessor = {
+        get: (t: unknown) => {
+          if (t === IAgentLifecycleService) return lifecycle;
+          if (t === ISessionActivity) return lifecycle.activity;
+          if (t === ISessionInteractionService) return lifecycle.interactions;
+          return undefined;
+        },
+      };
+      const handle = {
+        id: sid,
+        kind: 1,
+        accessor: sessionAccessor,
+        dispose: () => {},
+      } as unknown as IScopeHandle;
+      scopeHandles.set(sid, { lifecycle, handle });
+      return handle;
+    },
+    onDidCloseSession: lifecycleEvents.onDidCloseSession,
+    onDidArchiveSession: lifecycleEvents.onDidArchiveSession,
+  };
   const accessor = {
     get(token: unknown): unknown {
       if (token === IEventService) return eventBus;
-      if (token === ISessionLifecycleService) {
-        return {
-          get: (sid: string) => {
-            const lifecycle = sessions.get(sid);
-            if (lifecycle === undefined) return undefined;
-            const sessionAccessor = {
-              get: (t: unknown) => {
-                if (t === IAgentLifecycleService) return lifecycle;
-                if (t === ISessionActivity) return lifecycle.activity;
-                if (t === ISessionInteractionService) return lifecycle.interactions;
-                return undefined;
-              },
-            };
-            return { id: sid, kind: 1, accessor: sessionAccessor, dispose: () => {} };
-          },
-        };
-      }
+      if (token === ISessionLifecycleService) return sessionLifecycle;
+      if (token === ISessionIndex) return { get: async () => undefined };
       return undefined;
     },
   };
@@ -169,15 +256,17 @@ describe('SessionEventBroadcaster', () => {
   let dir: string;
   let sessions: Map<string, FakeLifecycle>;
   let eventBus: FakeEventBus;
+  let lifecycleEvents: FakeSessionLifecycleEvents;
   let bc: SessionEventBroadcaster;
 
   beforeEach(async () => {
     dir = await mkdtemp(join(tmpdir(), 'kimi-broadcaster-test-'));
     sessions = new Map();
     eventBus = new FakeEventBus();
+    lifecycleEvents = new FakeSessionLifecycleEvents();
     bc = new SessionEventBroadcaster({
       eventsDir: dir,
-      core: makeCore(sessions, eventBus),
+      core: makeCore(sessions, eventBus, lifecycleEvents),
       maxBufferSize: 3,
     });
   });
@@ -208,6 +297,309 @@ describe('SessionEventBroadcaster', () => {
     expect(envelopes[0]!.volatile).toBeUndefined();
   });
 
+  it('shares one owner when concurrent subscribers first activate a session', async () => {
+    const lc = new FakeLifecycle();
+    const main = lc.addAgent('main');
+    sessions.set('s1', lc);
+    const first = collectingTarget();
+    const second = collectingTarget();
+
+    expect(
+      await Promise.all([bc.subscribe('s1', first.target), bc.subscribe('s1', second.target)]),
+    ).toEqual([true, true]);
+    expect(main.bus.subscriberCount).toBe(1);
+
+    main.bus.emit(agentEvent('turn.started', { turnId: 1 }));
+    expect((await bc.getCursor('s1')).seq).toBe(2);
+    expect(first.envelopes.map((envelope) => envelope.type)).toEqual([
+      'event.session.status_changed',
+      'turn.started',
+    ]);
+    expect(second.envelopes.map((envelope) => envelope.type)).toEqual([
+      'event.session.status_changed',
+      'turn.started',
+    ]);
+  });
+
+  it('allows activation to retry after a shared concurrent initialization fails', async () => {
+    const lc = new FakeLifecycle();
+    const main = lc.addAgent('main');
+    main.failNextAccess = true;
+    sessions.set('s1', lc);
+
+    const failed = await Promise.allSettled([
+      bc.subscribe('s1', collectingTarget().target),
+      bc.subscribe('s1', collectingTarget().target),
+    ]);
+    expect(
+      failed.map((result) =>
+        result.status === 'rejected' ? (result.reason as Error).message : result.status,
+      ),
+    ).toEqual(['agent bus unavailable', 'agent bus unavailable']);
+    expect(main.bus.subscriberCount).toBe(0);
+
+    await expect(bc.subscribe('s1', collectingTarget().target)).resolves.toBe(true);
+    expect(main.bus.subscriberCount).toBe(1);
+  });
+
+  it('releases partial agent subscriptions when activation fails before publish', async () => {
+    const lc = new FakeLifecycle();
+    const first = lc.addAgent('agent-a');
+    const second = lc.addAgent('agent-b');
+    second.failNextAccess = true;
+    sessions.set('s1', lc);
+
+    await expect(bc.subscribe('s1', collectingTarget().target)).rejects.toThrow(
+      'agent bus unavailable',
+    );
+    expect([first.bus.subscriberCount, second.bus.subscriberCount]).toEqual([0, 0]);
+
+    await expect(bc.subscribe('s1', collectingTarget().target)).resolves.toBe(true);
+    expect([first.bus.subscriberCount, second.bus.subscriberCount]).toEqual([1, 1]);
+  });
+
+  it('retires an initializing owner once when close and archive fire together', async () => {
+    const lc = new FakeLifecycle();
+    const main = lc.addAgent('main');
+    sessions.set('s1', lc);
+    lc.onStatus = () => {
+      sessions.delete('s1');
+      lifecycleEvents.close('s1');
+      lifecycleEvents.archive('s1');
+    };
+    const closeJournal = vi.spyOn(SessionEventJournal.prototype, 'close');
+
+    try {
+      await expect(bc.subscribe('s1', collectingTarget().target)).resolves.toBe(false);
+      expect(main.bus.subscriberCount).toBe(0);
+      expect(main.bus.disposeCount).toBe(1);
+      expect(closeJournal).toHaveBeenCalledTimes(1);
+    } finally {
+      closeJournal.mockRestore();
+    }
+  });
+
+  it.each(['close', 'archive'] as const)(
+    'returns cold results when lifecycle %s happens during journal activation',
+    async (transition) => {
+      const lc = new FakeLifecycle();
+      const main = lc.addAgent('main');
+      sessions.set('s1', lc);
+      const originalOpen = SessionEventJournal.open;
+      let signalOpenStarted!: () => void;
+      let releaseOpen!: () => void;
+      const openStarted = new Promise<void>((resolve) => {
+        signalOpenStarted = resolve;
+      });
+      const openGate = new Promise<void>((resolve) => {
+        releaseOpen = resolve;
+      });
+      const openJournal = vi
+        .spyOn(SessionEventJournal, 'open')
+        .mockImplementation(async (filePath, logger) => {
+          if (filePath.endsWith('s1.jsonl')) {
+            signalOpenStarted();
+            await openGate;
+          }
+          return originalOpen(filePath, logger);
+        });
+      let subscribing: Promise<boolean> | undefined;
+
+      try {
+        subscribing = bc.subscribe('s1', collectingTarget().target);
+        await openStarted;
+        sessions.delete('s1');
+        lifecycleEvents[transition]('s1');
+        releaseOpen();
+
+        await expect(subscribing).resolves.toBe(false);
+        expect(main.bus.subscriberCount).toBe(0);
+        await expect(bc.getCursor('s1')).resolves.toEqual({ seq: 0, epoch: '' });
+        await expect(bc.getSnapshotState('s1')).resolves.toEqual({
+          seq: 0,
+          epoch: '',
+          inFlightTurn: null,
+          subagents: [],
+        });
+      } finally {
+        releaseOpen();
+        if (subscribing !== undefined) await Promise.allSettled([subscribing]);
+        openJournal.mockRestore();
+      }
+    },
+  );
+
+  it.each(['close', 'archive'] as const)(
+    'moves ownership to a fresh session after lifecycle %s',
+    async (transition) => {
+      const firstLifecycle = new FakeLifecycle();
+      const firstMain = firstLifecycle.addAgent('main');
+      sessions.set('s1', firstLifecycle);
+      await bc.subscribe('s1', collectingTarget().target);
+      expect(firstMain.bus.subscriberCount).toBe(1);
+
+      sessions.delete('s1');
+      lifecycleEvents[transition]('s1');
+      expect(firstMain.bus.subscriberCount).toBe(0);
+
+      const secondLifecycle = new FakeLifecycle();
+      const secondMain = secondLifecycle.addAgent('main');
+      sessions.set('s1', secondLifecycle);
+      const secondView = collectingTarget();
+      await bc.subscribe('s1', secondView.target);
+      expect(secondMain.bus.subscriberCount).toBe(1);
+
+      secondMain.bus.emit(agentEvent('turn.started', { turnId: 1 }));
+      await bc.getCursor('s1');
+      expect(secondView.envelopes.map((envelope) => envelope.type)).toEqual([
+        'event.session.status_changed',
+        'turn.started',
+      ]);
+    },
+  );
+
+  it('delivers durable events already queued when lifecycle close retires the owner', async () => {
+    const lc = new FakeLifecycle();
+    const main = lc.addAgent('main');
+    sessions.set('s1', lc);
+    const view = collectingTarget();
+    await bc.subscribe('s1', view.target);
+
+    main.bus.emit(agentEvent('turn.started', { turnId: 1 }));
+    sessions.delete('s1');
+    lifecycleEvents.close('s1');
+    await expect(bc.subscribe('s1', collectingTarget().target)).resolves.toBe(false);
+
+    expect(view.envelopes.map((envelope) => envelope.type)).toEqual([
+      'event.session.status_changed',
+      'turn.started',
+    ]);
+  });
+
+  it('returns false when subscribe waits on retirement during broadcaster close', async () => {
+    const lc = new FakeLifecycle();
+    lc.addAgent('main');
+    sessions.set('s1', lc);
+    await bc.subscribe('s1', collectingTarget().target);
+    const originalClose = SessionEventJournal.prototype.close;
+    let signalCloseStarted!: () => void;
+    let releaseClose!: () => void;
+    const closeStarted = new Promise<void>((resolve) => {
+      signalCloseStarted = resolve;
+    });
+    const closeGate = new Promise<void>((resolve) => {
+      releaseClose = resolve;
+    });
+    const closeJournal = vi
+      .spyOn(SessionEventJournal.prototype, 'close')
+      .mockImplementation(async function (this: SessionEventJournal) {
+        signalCloseStarted();
+        await closeGate;
+        return originalClose.call(this);
+      });
+    let subscribing: Promise<boolean> | undefined;
+    let closing: Promise<void> | undefined;
+
+    try {
+      sessions.delete('s1');
+      lifecycleEvents.close('s1');
+      await closeStarted;
+      subscribing = bc.subscribe('s1', collectingTarget().target);
+      closing = bc.close();
+      lifecycleEvents.failGet = true;
+      releaseClose();
+
+      await expect(subscribing).resolves.toBe(false);
+      await closing;
+      expect(closeJournal).toHaveBeenCalledTimes(1);
+    } finally {
+      lifecycleEvents.failGet = false;
+      releaseClose();
+      await Promise.allSettled([subscribing, closing].filter((value) => value !== undefined));
+      closeJournal.mockRestore();
+    }
+  });
+
+  it('does not append a core session event after lifecycle release wins the enqueue race', async () => {
+    const firstLifecycle = new FakeLifecycle();
+    const firstMain = firstLifecycle.addAgent('main');
+    sessions.set('s1', firstLifecycle);
+    const firstView = collectingTarget();
+    await bc.subscribe('s1', firstView.target);
+    firstMain.bus.emit(agentEvent('turn.started', { turnId: 1 }));
+    const beforeRelease = await bc.getCursor('s1');
+
+    eventBus.emit({
+      type: 'session.meta.updated',
+      payload: { sessionId: 's1', title: 'stale' },
+    });
+    sessions.delete('s1');
+    lifecycleEvents.close('s1');
+    await expect(bc.subscribe('s1', collectingTarget().target)).resolves.toBe(false);
+    expect(firstView.envelopes.map((envelope) => envelope.type)).not.toContain(
+      'session.meta.updated',
+    );
+
+    const secondLifecycle = new FakeLifecycle();
+    const secondMain = secondLifecycle.addAgent('main');
+    sessions.set('s1', secondLifecycle);
+    await bc.subscribe('s1', collectingTarget().target);
+    await expect(bc.getCursor('s1')).resolves.toEqual(beforeRelease);
+    secondMain.bus.emit(agentEvent('turn.ended', { turnId: 2 }));
+    expect((await bc.getCursor('s1')).seq).toBe(beforeRelease.seq + 1);
+    const replay = await bc.getBufferedSince('s1', beforeRelease);
+    expect(replay.events.map(({ envelope }) => envelope.type)).toEqual(['turn.ended']);
+  });
+
+  it('returns session-owned subscriptions to baseline across a 3 by 8 lifecycle soak', async () => {
+    const sessionIds = Array.from({ length: 8 }, (_, index) => `soak-${index}`);
+    const allLifecycles: FakeLifecycle[] = [];
+    const resourceSamples: Array<{
+      round: number;
+      activeOwners: number;
+      heapUsed: number;
+      rss: number;
+    }> = [];
+
+    for (let round = 1; round <= 3; round += 1) {
+      const lifecycles = sessionIds.map(() => {
+        const lifecycle = new FakeLifecycle();
+        lifecycle.addAgent('main');
+        return lifecycle;
+      });
+      allLifecycles.push(...lifecycles);
+      for (const [index, sessionId] of sessionIds.entries()) {
+        sessions.set(sessionId, lifecycles[index]!);
+        expect(await bc.subscribe(sessionId, collectingTarget().target)).toBe(true);
+      }
+      expect(
+        lifecycles.reduce(
+          (total, lifecycle) => total + lifecycle.handles[0]!.bus.subscriberCount,
+          0,
+        ),
+      ).toBe(8);
+
+      for (const [index, sessionId] of sessionIds.entries()) {
+        sessions.delete(sessionId);
+        if (index % 2 === 0) lifecycleEvents.close(sessionId);
+        else lifecycleEvents.archive(sessionId);
+      }
+      for (const sessionId of sessionIds) {
+        expect(await bc.subscribe(sessionId, collectingTarget().target)).toBe(false);
+      }
+      const activeOwners = allLifecycles.reduce(
+        (total, lifecycle) => total + lifecycle.handles[0]!.bus.subscriberCount,
+        0,
+      );
+      const memory = process.memoryUsage();
+      resourceSamples.push({ round, activeOwners, heapUsed: memory.heapUsed, rss: memory.rss });
+      expect(activeOwners).toBe(0);
+    }
+
+    console.info('SessionEventBroadcaster 3x8 resource samples', resourceSamples);
+    expect(resourceSamples.map((sample) => sample.activeOwners)).toEqual([0, 0, 0]);
+  });
+
   it('fans out volatile events with the current watermark + offset, not journaled', async () => {
     const lc = new FakeLifecycle();
     const main = lc.addAgent('main');
@@ -470,6 +862,93 @@ describe('SessionEventBroadcaster', () => {
     });
   });
 
+  it('shares one global journal owner when concurrent core events first activate it', async () => {
+    const lc = new FakeLifecycle();
+    lc.addAgent('main');
+    sessions.set('s1', lc);
+    const { target, envelopes } = collectingTarget();
+    await bc.subscribe('s1', target);
+    const openJournal = vi.spyOn(SessionEventJournal, 'open');
+
+    try {
+      const event = {
+        type: 'event.model_catalog.changed',
+        payload: {
+          changed: [{ provider_id: 'provider:test', provider_name: 'Example', added: 1, removed: 0 }],
+          unchanged: [],
+          failed: [],
+        },
+      };
+      eventBus.emit(event);
+      eventBus.emit(event);
+
+      await vi.waitFor(() => {
+        expect(envelopes).toHaveLength(2);
+      });
+      expect(openJournal).toHaveBeenCalledTimes(1);
+      expect(envelopes.map((envelope) => envelope.seq)).toEqual([1, 2]);
+    } finally {
+      openJournal.mockRestore();
+    }
+  });
+
+  it('shares the in-flight global activation barrier across concurrent close calls', async () => {
+    const originalOpen = SessionEventJournal.open;
+    let signalOpenStarted!: () => void;
+    let releaseOpen!: () => void;
+    const openStarted = new Promise<void>((resolve) => {
+      signalOpenStarted = resolve;
+    });
+    const openGate = new Promise<void>((resolve) => {
+      releaseOpen = resolve;
+    });
+    const openJournal = vi
+      .spyOn(SessionEventJournal, 'open')
+      .mockImplementation(async (filePath, logger) => {
+        if (filePath.endsWith('__global__.jsonl')) {
+          signalOpenStarted();
+          await openGate;
+        }
+        return originalOpen(filePath, logger);
+      });
+    let closing: Promise<void> | undefined;
+    let observedClosing: Promise<void> | undefined;
+
+    try {
+      eventBus.emit({
+        type: 'event.model_catalog.changed',
+        payload: { changed: [], unchanged: [], failed: [] },
+      });
+      await openStarted;
+      let closeSettled = false;
+      closing = bc.close();
+      const duplicateClosing = bc.close();
+      expect(duplicateClosing).toBe(closing);
+      observedClosing = duplicateClosing.then(() => {
+        closeSettled = true;
+      });
+      await new Promise<void>((resolve) => {
+        setImmediate(resolve);
+      });
+      expect(closeSettled).toBe(false);
+
+      releaseOpen();
+      await observedClosing;
+      eventBus.emit({
+        type: 'event.model_catalog.changed',
+        payload: { changed: [], unchanged: [], failed: [] },
+      });
+      await Promise.resolve();
+      expect(openJournal).toHaveBeenCalledTimes(1);
+    } finally {
+      releaseOpen();
+      await Promise.allSettled(
+        [closing, observedClosing].filter((value) => value !== undefined),
+      );
+      openJournal.mockRestore();
+    }
+  });
+
   it('subscribe returns false for an unknown session', async () => {
     const { target } = collectingTarget();
     expect(await bc.subscribe('nope', target)).toBe(false);
diff --git a/packages/kap-server/test/wsV1Resync.test.ts b/packages/kap-server/test/wsV1Resync.test.ts
index ff01d33851..ff0b05df88 100644
--- a/packages/kap-server/test/wsV1Resync.test.ts
+++ b/packages/kap-server/test/wsV1Resync.test.ts
@@ -24,6 +24,7 @@ interface Frame {
   type: string;
   id?: string;
   seq?: number;
+  epoch?: string;
   session_id?: string;
   payload?: Record<string, unknown>;
   volatile?: boolean;
@@ -235,6 +236,60 @@ describe('server-v2 /api/v1/ws resync', () => {
     await c2.closed;
   });
 
+  it('keeps the cursor epoch when archive and restore replace the live session owner', async () => {
+    const sid = await createSession();
+    await ensureMainAgent(sid);
+    const first = await openConn(wsUrl, server!.authTokenService.getToken());
+    await first.next((frame) => frame.type === 'server_hello');
+    first.send({
+      type: 'client_hello',
+      id: 'h1',
+      payload: withToken({ client_id: 'cli', subscriptions: [sid] }),
+    });
+    await first.next((frame) => frame.type === 'ack' && frame.id === 'h1');
+    emitAgentEvent(sid, { type: 'turn.started', turnId: 1 } as unknown as DomainEvent);
+    const beforeArchive = await first.next((frame) => frame.type === 'turn.started');
+    expect(beforeArchive.epoch).toMatch(/^ep_/);
+    first.ws.close();
+    await first.closed;
+
+    const archiveResponse = await fetch(`${base}/api/v1/sessions/${sid}:archive`, {
+      method: 'POST',
+      headers: authHeaders(server as RunningServer),
+    } as never);
+    expect(((await archiveResponse.json()) as { code: number }).code).toBe(0);
+    const restoreResponse = await fetch(`${base}/api/v1/sessions/${sid}:restore`, {
+      method: 'POST',
+      headers: authHeaders(server as RunningServer),
+    } as never);
+    expect(((await restoreResponse.json()) as { code: number }).code).toBe(0);
+    await ensureMainAgent(sid);
+
+    const second = await openConn(wsUrl, server!.authTokenService.getToken());
+    await second.next((frame) => frame.type === 'server_hello');
+    second.send({
+      type: 'client_hello',
+      id: 'h2',
+      payload: withToken({
+        client_id: 'cli',
+        subscriptions: [sid],
+        cursors: {
+          [sid]: { seq: beforeArchive.seq, epoch: beforeArchive.epoch },
+        },
+      }),
+    });
+    await second.next((frame) => frame.type === 'ack' && frame.id === 'h2');
+    expect(second.frames.some((frame) => frame.type === 'resync_required')).toBe(false);
+
+    emitAgentEvent(sid, { type: 'turn.ended', turnId: 2 } as unknown as DomainEvent);
+    const afterRestore = await second.next((frame) => frame.type === 'turn.ended');
+    expect(afterRestore.epoch).toBe(beforeArchive.epoch);
+    expect(afterRestore.seq).toBeGreaterThan(beforeArchive.seq!);
+
+    second.ws.close();
+    await second.closed;
+  });
+
   it('sends resync_required on epoch mismatch', async () => {
     const sid = await createSession();
     const c = await openConn(wsUrl, server!.authTokenService.getToken());