Merge branch 'worktree/process-plugin-streaming-72f9b3' into worktree/job-kill-delivery

# Conflicts:
#	packages/shell/bash-local/README.i18n.yaml
#	packages/shell/bash-local/README.md
#	packages/shell/bash-local/README.zh.md
#	packages/shell/bash-local/src/index.ts
#	packages/shell/bash-local/tests/executor.spec.ts
#	packages/shell/pwsh-local/README.i18n.yaml
#	packages/shell/pwsh-local/README.md
#	packages/shell/pwsh-local/README.zh.md
#	packages/shell/pwsh-local/src/index.ts
#	packages/shell/shell/README.i18n.yaml
#	packages/shell/shell/README.md
#	packages/shell/shell/README.zh.md
#	snapshots/session/pwsh-tool-turn/tool-schemas.expected.json

packages/api/session-controller/src/client/contract/session.ts

@@ -98,7 +98,7 @@ export interface ISession {
     attachmentId: AttachmentIdType,
   ): Promise<RemoteResult<{ attachment: ImageAttachmentRef; data: Uint8Array }>>
   /**
-   * Apply one edit, remove, or strict steer action to a still-pending queue occurrence.
+   * Apply one edit, remove, or Steer action to a still-pending queue occurrence.
    * @param itemId - agent-owned inbox occurrence identity.
    * @param action - requested queue operation.
    * @returns acceptance, or a business/transport error.

packages/api/session-controller/src/client/sessions/session.ts

@@ -271,6 +271,7 @@ export class Session implements SessionFace {
         parentSessionId: this.address.parentSessionId,
         childSessionId: this.address.childSessionId,
         mode: 'continuable',
+        delivery: mode,
         content: routedContent,
         clientTimeZone: resolvedClientTimeZone(),
       }, signal)

packages/api/session-controller/src/commands.ts

@@ -20,6 +20,7 @@ import type { SessionEvent, SessionHeader, SessionId, UserMessage } from '@deeps
 import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
 import { SessionTitleInvalidError } from '@deepseek-ai/dsh-session-title'
 import { canonicalClientTimeZone } from '@deepseek-ai/dsh-util-time'
+import { assertNever } from '@deepseek-ai/dsh-util-values'
 import { RemoteError, remoteErrorOf } from '@deepseek-ai/dsh-typert-protocol'
 import type { Workspace } from '@deepseek-ai/dsh-workspace'
 import {
@@ -58,6 +59,14 @@ interface SessionReadState {
   readonly events: readonly SessionEvent[]
 }
 
+type PromptContentCandidate =
+  | SessionPromptRequest['content'][number]
+  | Extract<SessionUpdateQueueRequest['action'], { readonly kind: 'edit' }>['content'][number]
+
+function hasPromptContent(content: readonly PromptContentCandidate[]): boolean {
+  return content.some(part => part.type !== 'text' || part.text.trim().length > 0)
+}
+
 /** Implements Session business commands delegated by the Session Controller Remote service. */
 export class SessionCommandController {
   /**
@@ -287,11 +296,18 @@ export class SessionCommandController {
   }
 
   /**
-   * Admit one browser prompt after explicit Agent resume and image validation.
+   * Reject empty content, then admit one prompt after Agent and attachment validation.
    * @param request - Session identity, prompt content, source metadata, and delivery mode.
    * @returns acknowledgement that the Agent accepted the prompt.
    */
   async prompt(request: SessionPromptRequest): Promise<SessionPromptValue> {
+    if (!hasPromptContent(request.content)) {
+      throw new RemoteError(
+        'gateway/bad-request',
+        'prompt content must include non-whitespace text or an attachment',
+        {},
+      )
+    }
     const clientTimeZone = request.clientTimeZone === undefined
       ? undefined
       : canonicalClientTimeZone(request.clientTimeZone)
@@ -407,21 +423,35 @@ export class SessionCommandController {
    * @returns acknowledgement that the queue mutation was applied.
    */
   updateQueue(request: SessionUpdateQueueRequest): SessionUpdateQueueValue {
-    if (request.action.kind === 'edit'
-      && request.action.content.some(block => block.type !== 'text')) {
-      throw new RemoteError(
-        'session/attachment-invalid',
-        'queue edits accept text content only',
-        { reason: 'QUEUE_EDIT_NON_TEXT' },
-      )
+    if (request.action.kind === 'edit') {
+      if (request.action.content.some(block => block.type !== 'text')) {
+        throw new RemoteError(
+          'session/attachment-invalid',
+          'queue edits accept text content only',
+          { reason: 'QUEUE_EDIT_NON_TEXT' },
+        )
+      }
+      if (!hasPromptContent(request.action.content)) {
+        throw new RemoteError(
+          'gateway/bad-request',
+          'queue edit content must include non-whitespace text',
+          {},
+        )
+      }
     }
     const agent = this.ctx.agents.get(request.sessionId)
-    if (agent !== undefined && hasApiSessionSubagentOwner(this.ctx, agent.session, agent)) {
-      throw apiSessionSubagentOwnershipError(request.sessionId)
-    }
     if (agent === undefined) {
       throw new RemoteError('session/queue-item-not-found', 'queued item is no longer pending', { itemId: request.itemId })
     }
+    if (hasApiSessionSubagentOwner(this.ctx, agent.session, agent)) {
+      const identity = this.ctx.sessionProjections
+        .snapshot(agent.session, ['subagent'])
+        .values.subagent
+      if (identity?.mode !== 'continuable'
+        || !agent.session.isOwnSeq(identity.seq)) {
+        throw apiSessionSubagentOwnershipError(request.sessionId)
+      }
+    }
     const nextTurn = agent.inbox.nextTurn.find(message => message.id === request.itemId)
     const nextStep = agent.inbox.nextStep.find(message => message.id === request.itemId)
     const located = nextTurn === undefined
@@ -434,20 +464,28 @@ export class SessionCommandController {
     if (request.action.kind === 'steer' && (target !== 'next-turn' || agent.status !== 'running')) {
       throw new RemoteError('session/steer-unavailable', 'current turn no longer accepts steering', { itemId: request.itemId })
     }
-    if (request.action.kind === 'edit') {
-      agent.inbox.replace(request.itemId, freezeMessage<UserMessage>({
-        ...message,
-        content: [...request.action.content],
-      }))
-    } else {
-      agent.inbox.remove(request.itemId)
-      if (request.action.kind === 'remove') {
+    switch (request.action.kind) {
+      case 'edit':
+        agent.inbox.replace(request.itemId, freezeMessage<UserMessage>({
+          ...message,
+          content: [...request.action.content],
+        }))
+        break
+      case 'remove': {
+        agent.inbox.remove(request.itemId)
         const source = message.source
         if (source.kind === 'user' && 'rpcId' in source) {
           this.ctx.fileUploads.retirePrompt(agent, source.rpcId)
         }
+        break
       }
-      if (request.action.kind === 'steer') agent.steer(message)
+      case 'steer':
+        agent.inbox.remove(request.itemId)
+        agent.steer(message)
+        break
+      /* v8 ignore next 2 -- closed-union exhaustiveness guard */
+      default:
+        assertNever(request.action, 'queue action')
     }
     return { accepted: true }
   }

packages/api/session-controller/src/control.ts

@@ -1,10 +1,10 @@
 /** Live Session queue and projection state with reconnect baselines. */
 
 import type { Context } from '@deepseek-ai/cordis'
-import type { Agent } from '@deepseek-ai/dsh-agent'
+import type { Agent, InboxState } from '@deepseek-ai/dsh-agent'
 import { Deque } from '@deepseek-ai/dsh-deque'
 import type {
-  Session, SessionEvent, SessionEventMap, SessionId, UserMessage,
+  Session, SessionId, UserMessage,
 } from '@deepseek-ai/dsh-session'
 import type { JsonValue } from '@deepseek-ai/dsh-util-values'
 import type {
@@ -21,7 +21,6 @@ export class SessionControlController {
 
   /** @param ctx - Host context carrying live Agent and projection services. */
   constructor(private readonly ctx: Context) {
-    ctx.on('session/event', (session, event) => { this.onSessionEvent(session, event) })
     ctx.sessionProjections.onChanged((session, key, value, seq) => {
       this.broadcast({
         type: 'projection',
@@ -30,6 +29,14 @@ export class SessionControlController {
         value: value as JsonValue,
         seq,
       })
+      if (key !== 'inbox') return
+      const agent = this.ctx.agents.get(session.id)
+      if (agent?.session !== session) return
+      this.broadcast({
+        type: 'queue',
+        sessionId: session.id,
+        items: queueItemsFromInbox(value as InboxState),
+      })
     })
     ctx.effect(() => () => {
       for (const stream of this.streams) stream.end()
@@ -83,17 +90,6 @@ export class SessionControlController {
     return blocks
   }
 
-  private onSessionEvent(session: Session, event: SessionEvent): void {
-    if (event.type !== 'agent/inbox/spliced') return
-    const agent = this.ctx.agents.get(session.id)
-    if (agent?.session !== session) return
-    this.broadcast({
-      type: 'queue',
-      sessionId: session.id,
-      items: queueItems(agent, event.data),
-    })
-  }
-
   private broadcast(frame: SessionControlFrame): void {
     for (const stream of this.streams) stream.push(frame)
   }
@@ -140,24 +136,22 @@ class ControlQueue {
   }
 }
 
-function queueItems(
-  agent: Agent,
-  splice?: SessionEventMap['agent/inbox/spliced'],
-): SessionQueuedItem[] {
-  const project = (target: 'next-turn' | 'next-step'): readonly UserMessage[] => {
-    const messages = target === 'next-turn' ? agent.inbox.nextTurn : agent.inbox.nextStep
-    return splice?.target === target
-      ? messages.toSpliced(splice.start, splice.removedCount ?? 0, ...splice.inserted)
-      : messages
-  }
+function queueItems(agent: Agent): SessionQueuedItem[] {
+  return queueItemsFromInbox({
+    'next-turn': agent.inbox.nextTurn,
+    'next-step': agent.inbox.nextStep,
+  })
+}
+
+function queueItemsFromInbox(inbox: InboxState): SessionQueuedItem[] {
   return [
-    ...project('next-turn').map(message => ({
+    ...inbox['next-turn'].map(message => ({
       id: message.id,
       placement: 'queued' as const,
       ...promptRpcId(message),
       message: { id: message.id, content: message.content as unknown as JsonValue[] },
     })),
-    ...project('next-step').map(message => ({
+    ...inbox['next-step'].map(message => ({
       id: message.id,
       placement: message.source.kind === 'user' ? 'steering' as const : 'context' as const,
       ...promptRpcId(message),

packages/api/session-controller/src/types.ts

@@ -150,7 +150,11 @@ export interface ModelCatalog {
 
 /** One client-requested mutation of a still-pending queue item. */
 export type QueueAction =
-  | { readonly kind: 'edit'; readonly content: readonly ContentBlock[] }
+  | {
+    readonly kind: 'edit'
+    /** Non-empty text-only replacement content. */
+    readonly content: readonly ContentBlock[]
+  }
   | { readonly kind: 'remove' }
   | { readonly kind: 'steer' }
 
@@ -307,6 +311,7 @@ export interface SessionPromptRequest {
   readonly requestId: SessionRequestId
   readonly sessionId: SessionId
   readonly mode: 'queue' | 'steer'
+  /** At least one non-whitespace text part or attachment. */
   readonly content: readonly PromptContentPart[]
   readonly clientTimeZone?: string
 }