fix(devframe): subscribe implicit state resources

Author: dvcolomban

packages/devframe/src/adapters/initiate.ts

@@ -321,7 +321,7 @@ export function initDevframe(
         const mounted = mountMcpHttp(app, context, mcpPath, {
           serverName: `${def.id} (devframe)`,
           serverVersion: def.version ?? '0.0.0',
-          exposeSharedState: true,
+          exposeSharedState: mcpConfig.exposeSharedState ?? true,
           allowedOrigins: mcpConfig.allowedOrigins,
         })
         mcpDispose = mounted.dispose

packages/devframe/src/adapters/mcp/__tests__/fixtures/resource-stdio-server.ts

@@ -8,7 +8,15 @@ const definition: DevframeDefinition = {
   packageName: '@devframe/resource-stdio-test',
   homepage: 'https://example.com',
   description: 'Stdio resource test fixture.',
-  setup(ctx) {
+  async setup(ctx) {
+    const state = await ctx.rpc.sharedState.get('stdio:counter', {
+      initialValue: { count: 0 },
+    })
+    ctx.agent.registerTool({
+      id: 'increment-state',
+      description: 'Increment the fixture state.',
+      handler: () => state.mutate(value => void (value.count += 1)),
+    })
     const fixed = ctx.agent.registerResource({
       id: 'status',
       uri: 'https://example.com/status',

packages/devframe/src/adapters/mcp/__tests__/mcp-http.test.ts

@@ -126,6 +126,58 @@ describe('mcp adapter (streamable http route)', () => {
     await client.close()
   })
 
+  it('pushes subscribed shared-state updates and cleans up on disconnect', async () => {
+    let updateState: (() => void) | undefined
+    const started = await boot(defineTestDef({
+      async setup(ctx) {
+        const state = await ctx.rpc.sharedState.get('build:status', {
+          initialValue: { revision: 0 },
+        })
+        updateState = () => state.mutate(value => void (value.revision += 1))
+      },
+    }))
+    const transport = originTransport(started)
+    const client = new Client({ name: 'test-client', version: '0.0.0' })
+    const notifications: string[] = []
+    client.setNotificationHandler('notifications/resources/updated', (notification) => {
+      notifications.push(notification.params.uri)
+    })
+
+    await client.connect(transport)
+    const uri = 'devframe://state/build%3Astatus'
+    await client.subscribeResource({ uri })
+    updateState!()
+    await vi.waitFor(() => expect(notifications).toEqual([uri]))
+
+    await transport.terminateSession()
+    updateState!()
+    expect(notifications).toEqual([uri])
+    await client.close()
+  })
+
+  it('can disable implicit shared-state MCP exposure for the HTTP route', async () => {
+    server = await createDevServer(defineTestDef({
+      async setup(ctx) {
+        await ctx.rpc.sharedState.get('hidden:state', { initialValue: { value: true } })
+      },
+    }), {
+      host: '127.0.0.1',
+      port: 0,
+      mcp: { exposeSharedState: false },
+    })
+    const client = new Client({ name: 'test-client', version: '0.0.0' })
+    try {
+      await client.connect(originTransport(server))
+      const resources = await client.listResources()
+      const tools = await client.listTools()
+      expect(resources.resources).toEqual([])
+      expect(tools.tools.map(tool => tool.name)).not.toContain('devframe_state_read')
+    }
+    finally {
+      await client.close()
+    }
+  })
+
   it('tears the session down on DELETE and rejects reuse of the id', async () => {
     const started = await boot()
     const url = `${started.origin}/__mcp`

packages/devframe/src/adapters/mcp/__tests__/mcp-server.test.ts

@@ -471,6 +471,60 @@ describe('mcp adapter (in-memory)', () => {
     }
   })
 
+  it('subscribes to shared-state updates and removes the listener on unsubscribe', async () => {
+    const { ctx, client, cleanup } = await bootPair()
+    const notifications: string[] = []
+    client.setNotificationHandler('notifications/resources/updated', (notification) => {
+      notifications.push(notification.params.uri)
+    })
+    try {
+      const state = await ctx.rpc.sharedState.get('my-plugin:counter', {
+        initialValue: { count: 0 },
+      })
+      const uri = `devframe://state/${encodeURIComponent('my-plugin:counter')}`
+
+      await Promise.all([
+        client.subscribeResource({ uri }),
+        client.subscribeResource({ uri }),
+      ])
+      state.mutate(value => void (value.count += 1))
+      await vi.waitFor(() => expect(notifications).toEqual([uri]))
+
+      await client.unsubscribeResource({ uri })
+      state.mutate(value => void (value.count += 1))
+      await client.listResources()
+      expect(notifications).toEqual([uri])
+    }
+    finally {
+      await cleanup()
+    }
+  })
+
+  it('rejects hidden, missing, and malformed shared-state subscriptions', async () => {
+    const ctx = await createHostContext({ cwd: process.cwd(), mode: 'dev', host: nullHost() })
+    await ctx.rpc.sharedState.get('visible:key', { initialValue: { value: true } })
+    await ctx.rpc.sharedState.get('hidden:key', { initialValue: { value: false } })
+    const { server, dispose } = buildMcpServerFromContext(ctx, {
+      serverName: 'test',
+      serverVersion: '0.0.0-test',
+      exposeSharedState: key => key.startsWith('visible:'),
+    })
+    const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair()
+    await server.connect(serverTransport)
+    const client = new Client({ name: 'test-client', version: '0.0.0' })
+    await client.connect(clientTransport)
+    try {
+      await expect(client.subscribeResource({ uri: 'devframe://state/hidden%3Akey' })).rejects.toThrow('unknown resource URI')
+      await expect(client.subscribeResource({ uri: 'devframe://state/missing' })).rejects.toThrow('unknown resource URI')
+      await expect(client.subscribeResource({ uri: 'devframe://state/%E0%A4%A' })).rejects.toThrow('unknown resource URI')
+    }
+    finally {
+      await client.close()
+      await server.close()
+      await dispose()
+    }
+  })
+
   it('omits non-object output schemas (MCP requires type: "object")', async () => {
     const { ctx, client, cleanup } = await bootPair()
     try {
@@ -598,6 +652,7 @@ describe('mcp adapter (stdio)', () => {
       expect(resources.resources.map(resource => resource.uri)).toEqual(expect.arrayContaining([
         'https://example.com/status',
         'devframe://logs/app',
+        'devframe://state/stdio%3Acounter',
       ]))
       const templates = await client.listResourceTemplates()
       expect(templates.resourceTemplates.map(template => template.uriTemplate)).toEqual(['devframe://logs/{name}'])
@@ -611,7 +666,15 @@ describe('mcp adapter (stdio)', () => {
       expect(JSON.parse((template.contents[0] as { text: string }).text)).toEqual({ process: 'worker' })
 
       await client.subscribeResource({ uri: 'https://example.com/status' })
-      await vi.waitFor(() => expect(updates).toEqual(['https://example.com/status']))
+      await client.subscribeResource({ uri: 'devframe://state/stdio%3Acounter' })
+      const increment = await client.callTool({ name: 'increment-state', arguments: {} })
+      expect(increment.isError).toBeFalsy()
+      const updatedState = await client.readResource({ uri: 'devframe://state/stdio%3Acounter' })
+      expect(JSON.parse((updatedState.contents[0] as { text: string }).text)).toEqual({ count: 1 })
+      await vi.waitFor(() => expect(updates).toEqual(expect.arrayContaining([
+        'https://example.com/status',
+        'devframe://state/stdio%3Acounter',
+      ])))
     }
     finally {
       await client.close()

packages/devframe/src/adapters/mcp/build-server.ts

@@ -303,7 +303,11 @@ function registerResourceHandlers(
   ctx: DevframeNodeContext,
   exposeSharedState: boolean | ((key: string) => boolean),
 ): () => Promise<void> {
-  const subscriptions = new Map<string, () => void | Promise<void>>()
+  const stateFilter = sharedStateFilter(exposeSharedState)
+  const subscriptions = new Map<string, {
+    kind: 'agent' | 'state'
+    cleanup: () => void | Promise<void>
+  }>()
   let subscriptionOperations = Promise.resolve()
   const runSubscriptionOperation = <Result>(operation: () => Promise<Result>): Promise<Result> => {
     const result = subscriptionOperations.then(operation)
@@ -325,10 +329,9 @@ function registerResourceHandlers(
       resources.push(...listed.resources)
     }
 
-    if (exposeSharedState !== false) {
-      const filter = typeof exposeSharedState === 'function' ? exposeSharedState : () => true
+    if (stateFilter) {
       for (const key of ctx.rpc.sharedState.keys()) {
-        if (!filter(key))
+        if (!stateFilter(key))
           continue
         resources.push({
           uri: `devframe://state/${encodeURIComponent(key)}`,
@@ -370,7 +373,7 @@ function registerResourceHandlers(
     }
 
     const parsed = parseResourceUri(uri)
-    if (parsed.kind === 'state') {
+    if (parsed.kind === 'state' && stateFilter?.(parsed.key) && ctx.rpc.sharedState.keys().includes(parsed.key)) {
       const state = await ctx.rpc.sharedState.get(parsed.key)
       return {
         contents: [
@@ -393,24 +396,36 @@ function registerResourceHandlers(
         return {}
 
       const resource = resolveAgentResource(ctx, uri)
-      if (!resource)
+      if (resource) {
+        const cleanup = await ctx.agent.subscribeResource(resource.id, uri)
+        subscriptions.set(uri, { kind: 'agent', cleanup })
+        return {}
+      }
+
+      const parsed = parseResourceUri(uri)
+      if (parsed.kind !== 'state' || !stateFilter?.(parsed.key) || !ctx.rpc.sharedState.keys().includes(parsed.key))
         throw new Error(`[devframe/mcp] unknown resource URI "${uri}"`)
 
-      const cleanup = await ctx.agent.subscribeResource(resource.id, uri)
-      subscriptions.set(uri, cleanup)
+      const state = await ctx.rpc.sharedState.get(parsed.key)
+      const cleanup = state.on('updated', () => {
+        if (!subscriptions.has(uri))
+          return
+        void server.sendResourceUpdated({ uri }).catch(() => { /* ignore transport errors */ })
+      })
+      subscriptions.set(uri, { kind: 'state', cleanup })
       return {}
     })
   })
 
   server.setRequestHandler('resources/unsubscribe', async (request) => {
     const { uri } = request.params
     return await runSubscriptionOperation(async () => {
-      const cleanup = subscriptions.get(uri)
-      if (!cleanup)
+      const subscription = subscriptions.get(uri)
+      if (!subscription)
         return {}
 
       subscriptions.delete(uri)
-      await cleanup()
+      await subscription.cleanup()
       return {}
     })
   })
@@ -423,13 +438,23 @@ function registerResourceHandlers(
 
   const offManifest = ctx.agent.events.on(DEVFRAME_EVENTS.bus.agentManifestChanged, () => {
     void runSubscriptionOperation(async () => {
-      for (const [uri, cleanup] of [...subscriptions]) {
+      for (const [uri, subscription] of [...subscriptions]) {
+        if (subscription.kind === 'state')
+          continue
+        try {
+          await subscription.cleanup()
+        }
+        catch {
+          continue
+        }
         subscriptions.delete(uri)
-        await cleanup()
         const resource = resolveAgentResource(ctx, uri)
         if (!resource)
           continue
-        subscriptions.set(uri, await ctx.agent.subscribeResource(resource.id, uri))
+        subscriptions.set(uri, {
+          kind: 'agent',
+          cleanup: await ctx.agent.subscribeResource(resource.id, uri),
+        })
       }
     }).catch(() => { /* ignore subscription cleanup errors during reconciliation */ })
   })
@@ -440,7 +465,7 @@ function registerResourceHandlers(
     await runSubscriptionOperation(async () => {
       const active = [...subscriptions.values()]
       subscriptions.clear()
-      await Promise.all(active.map(cleanup => cleanup()))
+      await Promise.all(active.map(subscription => subscription.cleanup()))
     })
   }
 }
@@ -520,7 +545,13 @@ function parseResourceUri(uri: string): { kind: 'resource', id: string } | { kin
   if (!match)
     return { kind: 'unknown' }
   const [, kind, rest] = match
-  const decoded = decodeURIComponent(rest!)
+  let decoded: string
+  try {
+    decoded = decodeURIComponent(rest!)
+  }
+  catch {
+    return { kind: 'unknown' }
+  }
   if (kind === 'resource')
     return { kind: 'resource', id: decoded }
   return { kind: 'state', key: decoded }

packages/devframe/src/types/devframe.ts

@@ -110,6 +110,12 @@ export interface McpRouteOptions {
    * deprecated `allowedHosts`/`allowedOrigins` transport flags).
    */
   allowedOrigins?: readonly string[] | false
+  /**
+   * Expose shared-state keys as MCP resources and through the built-in
+   * `devframe_state_read` tool. Defaults to `true`; pass a predicate to
+   * expose selected keys.
+   */
+  exposeSharedState?: boolean | ((key: string) => boolean)
 }
 
 export interface DevframeCliOptions {

packages/hub/src/node/initiate.ts

@@ -575,7 +575,7 @@ export function initHub(options: InitHubOptions): HubInstance {
       const mounted = mountMcpHttp(app, ctx, joinURL(base, mcpRoute), {
         serverName: options.name ?? 'devframes-hub',
         serverVersion: options.version ?? '0.0.0',
-        exposeSharedState: true,
+        exposeSharedState: mcpConfig.exposeSharedState ?? true,
         allowedOrigins: mcpConfig.allowedOrigins,
       })
       return { context: ctx, mcp: { path: mcpRoute }, dispose: mounted.dispose }

tests/__snapshots__/tsnapi/devframe/index.snapshot.d.ts

@@ -428,6 +428,7 @@ export interface EventUnsubscribe {
 export interface McpRouteOptions {
   path?: string;
   allowedOrigins?: readonly string[] | false;
+  exposeSharedState?: boolean | ((_: string) => boolean);
 }
 export interface RemoteAssets {
   package: string;