feat(core): scope-aware registries and session dispatch carriers

dsh-tools and dsh-system-prompt gain a per-scope registration layer over
dsh-scope: a registration through a scoped context files into that scope,
shadows a same-named global contribution for that scope (per-agent persona
and tool variants), and unwinds with the scope. tools.restrict() masks the
global surface per scope (snapshot-at-registration, loud unknown-name
validation, intersection composition; scoped grants bypass). One visibility
function feeds schemas/get/execute, so prompt, presentation, and dispatch
can never disagree; out-of-view executes as UNKNOWN_TOOL.

Prompt tool providers now receive the AssembleContext and return
{schemas, knownNames}: toolOrder validates against the pre-restriction name
universe (a typo fails every assembly loudly) while ordering operates on
the post-restriction schemas (a restricted-away tool is a normal absence).

dsh-session captures each session's dispatch carrier at enter() from the
entering context's scope tag, and the new sessions.flush(session) owns the
awaited session/flush dispatch. tools/pre|post-execute and
system-prompt/assemble dispatch with scope carriers keyed by their subject;
session/created|event|flush by the owning session's scope.

packages/core/session/package.json

@@ -24,11 +24,13 @@
   "peerDependencies": {
     "@deepseek-ai/dsh-brand": "^0.0.1",
     "@deepseek-ai/dsh-llm": "^0.0.1",
+    "@deepseek-ai/dsh-scope": "^0.0.1",
     "cordis": "^4.0.0-rc.6"
   },
   "devDependencies": {
     "@deepseek-ai/dsh-brand": "workspace:^",
     "@deepseek-ai/dsh-llm": "workspace:^",
+    "@deepseek-ai/dsh-scope": "workspace:^",
     "cordis": "^4.0.0-rc.6"
   }
 }

packages/core/session/src/index.ts

@@ -9,6 +9,8 @@
 import { Context, Service } from 'cordis'
 import { isAbsolute } from 'node:path'
 import { deepFreeze } from '@deepseek-ai/dsh-llm'
+import { scopeOf, scopeTarget } from '@deepseek-ai/dsh-scope'
+import type { Scoped } from '@deepseek-ai/dsh-scope'
 import type { ContentBlock, Message, MessageSource } from '@deepseek-ai/dsh-llm'
 import { SESSION_FORMAT_VERSION, SessionId } from './types.ts'
 import type { CreateSessionOptions, EpochHeader, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SurfaceIntent, SurfaceEventType } from './types.ts'
@@ -33,28 +35,48 @@ declare module 'cordis' {
   interface Events {
     /**
      * A session was created in the store.
+     * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): the carrier is the
+     * session's owner scope, captured when the session was ENTERED (an agent's
+     * session is entered through `agent.ctx`, so its events dispatch in that
+     * agent's scope; a bare `sessions.create()` from a plain plugin dispatches
+     * subject-less). A listener registered through `agent.ctx` hears only that
+     * agent's sessions; a plain plugin listener hears every session.
      * @param session - the session just entered and announced.
      * @mode emit
      */
-    'session/created'(session: Session): void
+    'session/created'(this: Scoped<Session>, session: Session): void
     /**
      * An event was appended to a session log (sync, fire-and-forget). This is
      * the per-append feed a UI or invariant plugin tails.
+     * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): the carrier is the
+     * session's owner scope, captured when the session was ENTERED (an agent's
+     * session is entered through `agent.ctx`, so its events dispatch in that
+     * agent's scope; a bare `sessions.create()` from a plain plugin dispatches
+     * subject-less). A listener registered through `agent.ctx` hears only that
+     * agent's sessions; a plain plugin listener hears every session.
      * @param session - the session whose log grew.
      * @param event - the appended event, exactly as recorded.
      * @mode emit
      */
-    'session/event'(session: Session, event: SessionEvent): void
+    'session/event'(this: Scoped<Session>, session: Session, event: SessionEvent): void
     /**
      * Awaited durability checkpoint. The agent loop awaits
-     * `ctx.parallel('session/flush', session)` at every turn end; persistence
+     * `ctx.sessions.flush(session)` at every turn end; persistence
      * plugins (JSONL, SQLite) drain their write-behind buffers here and on
      * fiber dispose. Awaited (parallel), not a waterfall: every listener runs
-     * and the loop waits for all of them, but none can veto.
+     * and the caller waits for all of them, but none can veto. Dispatch it
+     * through {@link SessionStore.flush} — the store owns the carrier — never
+     * via a raw `ctx.parallel`.
+     * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): the carrier is the
+     * session's owner scope, captured when the session was ENTERED (an agent's
+     * session is entered through `agent.ctx`, so its events dispatch in that
+     * agent's scope; a bare `sessions.create()` from a plain plugin dispatches
+     * subject-less). A listener registered through `agent.ctx` hears only that
+     * agent's sessions; a plain plugin listener hears every session.
      * @param session - the session whose buffered events must reach durable storage.
      * @mode parallel
      */
-    'session/flush'(session: Session): Promise<void> | void
+    'session/flush'(this: Scoped<Session>, session: Session): Promise<void> | void
   }
 }
 
@@ -404,6 +426,14 @@ export class SessionForkError extends Error {
  */
 export class SessionStore extends Service {
   private store = new Map<SessionId, Session>()
+  /**
+   * Each live session's dispatch carrier, captured at {@link enter} from the
+   * ENTERING context's scope tag (an agent session is entered through
+   * `agent.ctx` ⇒ its events dispatch in that agent's scope; a bare session ⇒
+   * subject-less carrier). WeakMap so a detached session drops its carrier
+   * with the entry.
+   */
+  private carriers = new WeakMap<Session, Scoped<Session>>()
   private counter = 0
 
   constructor(ctx: Context) {
@@ -498,7 +528,15 @@ export class SessionStore extends Service {
    */
   enter(session: Session): () => void {
     if (this.store.has(session.id)) throw new Error(`session "${session.id}" already exists`)
-    session.onAppend = (event) => { this.ctx.emit('session/event', session, event) }
+    // The carrier is decided HERE, once, from the ENTERING context's scope tag
+    // (`this.ctx` is the caller's context — the tracker mechanism): every
+    // session/created|event|flush dispatch for this session uses it, so the
+    // session's whole event feed is scope-filtered consistently. The base is
+    // the session itself (scoped listeners' `this` is the session).
+    const carrier = scopeTarget(session, scopeOf(this.ctx))
+    this.carriers.set(session, carrier)
+    const emitCtx = this.ctx
+    session.onAppend = (event) => { emitCtx.emit(carrier, 'session/event', session, event) }
     this.store.set(session.id, session)
     return () => {
       session.onAppend = undefined
@@ -506,12 +544,32 @@ export class SessionStore extends Service {
     }
   }
 
-  /** Emit `session/created` for an {@link enter}ed session. Separate from
-   * {@link enter} so the caller can yield the detach disposer first (rollback
-   * safety — see {@link enter}).
+  /** Emit `session/created` for an {@link enter}ed session (with the carrier
+   * {@link enter} captured). Separate from {@link enter} so the caller can
+   * yield the detach disposer first (rollback safety — see {@link enter}).
    * @param session - the entered session to announce to listeners. */
   announce(session: Session): void {
-    this.ctx.emit('session/created', session)
+    this.ctx.emit(this.carrierFor(session), 'session/created', session)
+  }
+
+  /**
+   * Dispatch the awaited `session/flush` durability checkpoint for `session`,
+   * with the carrier captured at {@link enter}. THE flush entry point: the
+   * store owns the carrier, so callers (the loop's turn-end checkpoint, idle
+   * injection, teardown drains) must come through here rather than dispatch a
+   * raw `ctx.parallel('session/flush', …)` — one owner, one spelling, and the
+   * scoped-dispatch invariant can pin it.
+   * @param session - the session whose buffered events must reach durable storage.
+   * @returns resolves when every flush listener has settled; rejects if one rejects.
+   */
+  async flush(session: Session): Promise<void> {
+    await this.ctx.parallel(this.carrierFor(session), 'session/flush', session)
+  }
+
+  /** The carrier {@link enter} captured, or a subject-less one for a session
+   * never entered (defensive: dispatch stays filtered either way). */
+  private carrierFor(session: Session): Scoped<Session> {
+    return this.carriers.get(session) ?? scopeTarget(session, undefined)
   }
 
   /**

packages/core/session/tests/scoped.spec.ts

@@ -0,0 +1,112 @@
+import { describe, expect, it } from 'vitest'
+import { Context } from 'cordis'
+import { createScope, scopeOf } from '@deepseek-ai/dsh-scope'
+import type { Scope, ScopeKey } from '@deepseek-ai/dsh-scope'
+import SessionStore from '@deepseek-ai/dsh-session'
+import type { Session } from '@deepseek-ai/dsh-session'
+
+async function mount(): Promise<Context> {
+  const ctx = new Context()
+  await ctx.plugin(SessionStore)
+  return ctx
+}
+
+async function mintScope(ctx: Context, name: string): Promise<Scope> {
+  let scope!: Scope
+  // The scoped context resolves services through the MINTING plugin's
+  // dependency chain — the minter must inject what scope holders will reach.
+  await ctx.plugin(Object.assign((inner: Context) => { scope = createScope(inner, { name }) },
+    { inject: ['sessions'] }))
+  return scope
+}
+
+/** The key a test scope was minted with. */
+function keyOf(scope: Scope): ScopeKey {
+
+  return scopeOf(scope.ctx)!
+}
+
+describe('session dispatch carriers', () => {
+  it('a session entered through a scoped context dispatches its events in that scope', async () => {
+    const ctx = await mount()
+    const scope = await mintScope(ctx, 'owner')
+    const otherScope = await mintScope(ctx, 'other')
+
+    const heard: string[] = []
+    ctx.on('session/event', (_session, event) => void heard.push(`global:${event.type}`))
+    scope.ctx.on('session/event', (_session, event) => void heard.push(`owner:${event.type}`))
+    otherScope.ctx.on('session/event', (_session, event) => void heard.push(`other:${event.type}`))
+    scope.ctx.on('session/created', session => void heard.push(`owner-created:${session.id}`))
+    otherScope.ctx.on('session/created', session => void heard.push(`other-created:${session.id}`))
+
+    const session = scope.ctx.sessions.create()
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+
+    expect(heard).toEqual([
+      `owner-created:${session.id}`,
+      'global:turn/start',
+      'owner:turn/start',
+    ])
+  })
+
+  it('a bare session dispatches subject-less: scoped listeners never hear it', async () => {
+    const ctx = await mount()
+    const scope = await mintScope(ctx, 'owner')
+    const heard: string[] = []
+    ctx.on('session/event', (_s, event) => void heard.push(`global:${event.type}`))
+    scope.ctx.on('session/event', (_s, event) => void heard.push(`owner:${event.type}`))
+
+    const bare = ctx.sessions.create()
+    bare.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    expect(heard).toEqual(['global:turn/start'])
+  })
+})
+
+describe('sessions.flush()', () => {
+  it('dispatches session/flush with the owning carrier and awaits all listeners', async () => {
+    const ctx = await mount()
+    const scope = await mintScope(ctx, 'owner')
+    const flushed: string[] = []
+    ctx.on('session/flush', async (session: Session) => {
+      await Promise.resolve()
+      flushed.push(`global:${session.id}`)
+    })
+    scope.ctx.on('session/flush', (session: Session) => void flushed.push(`owner:${session.id}`))
+
+    const owned = scope.ctx.sessions.create()
+    const bare = ctx.sessions.create()
+    await ctx.sessions.flush(owned)
+    await ctx.sessions.flush(bare)
+
+    // Parallel dispatch: listener completion order is unspecified (the global
+    // listener awaits a microtask) — assert set membership per flush instead.
+    expect(flushed.slice(0, 2).sort()).toEqual([`global:${owned.id}`, `owner:${owned.id}`])
+    expect(flushed.slice(2)).toEqual([`global:${bare.id}`])
+  })
+
+  it('propagates a rejecting flush listener (the caller owns the failure policy)', async () => {
+    const ctx = await mount()
+    ctx.on('session/flush', () => Promise.reject(new Error('disk full')))
+    const session = ctx.sessions.create()
+    await expect(ctx.sessions.flush(session)).rejects.toThrow('disk full')
+  })
+
+  it('flushes a never-entered session with a subject-less carrier (defensive path)', async () => {
+    const ctx = await mount()
+    const scope = await mintScope(ctx, 'owner')
+    const flushed: string[] = []
+    ctx.on('session/flush', (session: Session) => void flushed.push(`global:${session.id}`))
+    scope.ctx.on('session/flush', (session: Session) => void flushed.push(`owner:${session.id}`))
+
+    const detached = ctx.sessions.prepare()
+    await ctx.sessions.flush(detached)
+    expect(flushed).toEqual([`global:${detached.id}`])
+  })
+
+  it('keyOf sanity: distinct scopes carry distinct keys', async () => {
+    const ctx = await mount()
+    const a = await mintScope(ctx, 'a')
+    const b = await mintScope(ctx, 'b')
+    expect(keyOf(a)).not.toBe(keyOf(b))
+  })
+})

packages/core/session/tsconfig.json

@@ -19,6 +19,9 @@
     },
     {
       "path": "../../llm/llm"
+    },
+    {
+      "path": "../../core/scope"
     }
   ]
 }