refactor(orchestration): resolve every caller to one orchestration actor

src/main/runtime/orchestration/db/dispatch-context/dispatch-context-store.ts

@@ -9,6 +9,7 @@ import { recordedCreatorIdentity, type DispatchCreator } from '../dispatch-depth
 import type { OrchestrationDb } from '../orchestration-db'
 import { transitionLifecycleWithDb } from '../lifecycle-transition'
 import { taskNotFoundError, taskNotStartableError } from '../../task-dispatch-refusal'
+import { structuredWorkerActorForIncarnation } from '../../../structured-worker-identity'
 
 export function createDispatchContext(
   this: OrchestrationDb,
@@ -65,6 +66,7 @@ export function createDispatchContext(
       launchTokenHash: launchTokenHash ?? null,
       assigneeHandle,
       assigneePaneKey: assigneePaneKey ?? null,
+      assigneeActor: structuredWorkerActorForIncarnation(processIncarnation),
       processIncarnation: processIncarnation ?? null,
       creatorDispatchId,
       ...recordedCreatorIdentity(params.creator),

src/main/runtime/orchestration/db/dispatch-depth.ts

@@ -26,17 +26,29 @@ export type DispatchCreator =
       paneKey?: string
       /** Remote attachment matching requires the exact incarnation; local rows do not. */
       processIncarnation?: string
+      /** A structured worker's `session:<id>`, recorded beside its handle. */
+      actor?: string | null
     }
+  /** A structured session with no terminal handle, identified by its actor alone. */
+  | { kind: 'actor'; actor: string }
 
 /** Creator identity to persist on a new row, so depth can later tell delegation from bookkeeping. */
 export function recordedCreatorIdentity(creator: DispatchCreator): {
   creatorHandle: string | null
   creatorPaneKey: string | null
+  creatorActor: string | null
 } {
   if (creator.kind === 'system') {
-    return { creatorHandle: null, creatorPaneKey: null }
+    return { creatorHandle: null, creatorPaneKey: null, creatorActor: null }
+  }
+  if (creator.kind === 'actor') {
+    return { creatorHandle: null, creatorPaneKey: null, creatorActor: creator.actor }
+  }
+  return {
+    creatorHandle: creator.handle,
+    creatorPaneKey: creator.paneKey ?? null,
+    creatorActor: creator.actor ?? null
   }
-  return { creatorHandle: creator.handle, creatorPaneKey: creator.paneKey ?? null }
 }
 
 /**
@@ -84,14 +96,14 @@ export function resolveCreatorDepth(this: OrchestrationDb, creator: DispatchCrea
   // Local rows match on handle/pane as they always have. process_incarnation is
   // nullable here and context-only dispatch stores null deliberately, so
   // requiring it would drop real parents.
-  const local = this.findActiveDispatchForAssignee(creator.handle, creator.paneKey) as
-    | DispatchContextRow
-    | undefined
+  const local = findActiveDispatchForCreator.call(this, creator)
   if (local && !isSelfCreatedDispatch(local)) {
     depths.push(local.depth)
   }
 
-  for (const attachment of findPotentiallyLiveAttachmentsForCreator.call(this, creator)) {
+  const attachments =
+    creator.kind === 'terminal' ? findPotentiallyLiveAttachmentsForCreator.call(this, creator) : []
+  for (const attachment of attachments) {
     depths.push(attachment.depth)
   }
 
@@ -109,16 +121,36 @@ export function resolveCreatorDispatchId(
   if (creator.kind === 'system') {
     return null
   }
-  const own = this.findActiveDispatchForAssignee(creator.handle, creator.paneKey)
+  const own = findActiveDispatchForCreator.call(this, creator)
   // Why: a self-dispatch is not a parent Attempt, so it must not be stamped as the child's creator.
   const local = own && !isSelfCreatedDispatch(own) ? own : undefined
-  const remote = findPotentiallyLiveAttachmentsForCreator.call(this, creator)
+  const remote =
+    creator.kind === 'terminal' ? findPotentiallyLiveAttachmentsForCreator.call(this, creator) : []
   if ((local ? 1 : 0) + remote.length !== 1) {
     return null
   }
   return local?.id ?? remote[0]?.dispatch_id ?? null
 }
 
+/** The live Dispatch this creator is itself working on, found the way its identity is recorded. */
+function findActiveDispatchForCreator(
+  this: OrchestrationDb,
+  creator: Exclude<DispatchCreator, { kind: 'system' }>
+): DispatchContextRow | undefined {
+  if (creator.kind === 'terminal') {
+    return this.findActiveDispatchForAssignee(creator.handle, creator.paneKey)
+  }
+  const row = this.db
+    .prepare(
+      `SELECT * FROM dispatch_contexts
+       WHERE assignee_actor = ? AND status IN ('pending', 'dispatched')
+       ORDER BY rowid DESC LIMIT 1`
+    )
+    .get(creator.actor)
+  // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: SELECT * over this table returns the row shape its schema and row type define, like every row cast in db/.
+  return row as DispatchContextRow | undefined
+}
+
 /**
  * Remote attachments matching this caller's pane AND exact process incarnation.
  *

src/main/runtime/orchestration/db/dispatch-row-writer.ts

@@ -14,11 +14,11 @@ import { DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL } from './pane-key-match'
 
 export const DISPATCH_CONTEXT_CLAIM_SQL = `INSERT INTO dispatch_contexts (
   id, run_id, task_id, contract_version, launch_token_hash,
-  assignee_handle, assignee_pane_key, process_incarnation,
-  creator_dispatch_id, creator_handle, creator_pane_key,
+  assignee_handle, assignee_pane_key, assignee_actor, process_incarnation,
+  creator_dispatch_id, creator_handle, creator_pane_key, creator_actor,
   status, failure_count, depth, dispatched_at
 )
-SELECT ?, run_id, id, ?, ?, ?, ?, ?, ?, ?, ?, 'dispatched', ?, ?, datetime('now')
+SELECT ?, run_id, id, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'dispatched', ?, ?, datetime('now')
 FROM tasks
 WHERE id = ? AND status = 'ready'
   AND NOT EXISTS (
@@ -45,8 +45,9 @@ WHERE id = ? AND status = 'ready'
 
 const STARTING_DISPATCH_CONTEXT_SQL = `INSERT INTO dispatch_contexts (
    id, run_id, task_id, contract_version, launch_token_hash, retry_of_dispatch_id,
-   creator_dispatch_id, creator_handle, creator_pane_key, depth, status, dispatched_at
- ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', datetime('now'))`
+   creator_dispatch_id, creator_handle, creator_pane_key, creator_actor, depth, status,
+   dispatched_at
+ ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', datetime('now'))`
 
 const REMOTE_DISPATCH_ATTACHMENT_SQL = `INSERT INTO remote_dispatch_attachments (
    dispatch_id, home_run_id, task_id, home_peer_fingerprint, protocol_version, runtime_epoch, depth
@@ -70,10 +71,12 @@ export function claimDispatchContextRow(
     launchTokenHash: string | null
     assigneeHandle: string
     assigneePaneKey: string | null
+    assigneeActor?: string | null
     processIncarnation: string | null
     creatorDispatchId?: string | null
     creatorHandle?: string | null
     creatorPaneKey?: string | null
+    creatorActor?: string | null
     priorFailures: number
     depth: number
     taskId: string
@@ -89,10 +92,12 @@ export function claimDispatchContextRow(
       params.launchTokenHash,
       params.assigneeHandle,
       params.assigneePaneKey,
+      params.assigneeActor ?? null,
       params.processIncarnation,
       params.creatorDispatchId ?? null,
       params.creatorHandle ?? null,
       params.creatorPaneKey ?? null,
+      params.creatorActor ?? null,
       params.priorFailures,
       params.depth,
       params.taskId,
@@ -118,6 +123,7 @@ export function insertStartingDispatchContextRow(
     creatorDispatchId?: string | null
     creatorHandle?: string | null
     creatorPaneKey?: string | null
+    creatorActor?: string | null
   }
 ): void {
   assertStampedDepth(params.depth)
@@ -131,6 +137,7 @@ export function insertStartingDispatchContextRow(
     params.creatorDispatchId ?? null,
     params.creatorHandle ?? null,
     params.creatorPaneKey ?? null,
+    params.creatorActor ?? null,
     params.depth
   )
 }

src/main/runtime/orchestration/db/messages/direct-mailbox-routing.ts

@@ -1,6 +1,7 @@
 import type { MessageType } from '../../types'
 import type { OrchestrationDb } from '../orchestration-db'
 import { ORCHESTRATION_DELIVERY_BATCH_LIMIT, type MailboxRoutingPage } from './mailbox-routing-page'
+import { activeDispatchOwnsAddressSql } from '../runs/run-coordinator-mail-routing'
 
 export function hasUndeliveredDirectMessageForRun(
   this: OrchestrationDb,
@@ -48,12 +49,7 @@ export function routeDirectMessagePage(
   try {
     const throughClause = throughSequence === undefined ? '' : ' AND sequence <= ?'
     const dispatchOwnershipClause = preserveActiveDispatchOwnership
-      ? ` AND NOT EXISTS (
-           SELECT 1 FROM dispatch_contexts
-           WHERE dispatch_contexts.run_id = messages.run_id
-             AND dispatch_contexts.assignee_handle = messages.to_handle
-             AND dispatch_contexts.status IN ('pending', 'dispatched')
-         )`
+      ? ` AND NOT ${activeDispatchOwnsAddressSql('messages.run_id', 'messages.to_handle')}`
       : ''
     const params: (string | number)[] = [runId, directHandle]
     if (throughSequence !== undefined) {

src/main/runtime/orchestration/db/messages/foreign-direct-mailbox-routing.ts

@@ -1,4 +1,5 @@
 import { parsePaneKey } from '../../../../../shared/stable-pane-id'
+import { parseOrchestrationActor } from '../../../../../shared/orchestration-actor'
 import type { DispatchContextRow, MessageType } from '../../types'
 import { DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL, paneKeyMatchSuffix } from '../pane-key-match'
 import type { OrchestrationDb } from '../orchestration-db'
@@ -20,9 +21,26 @@ export function findActiveDispatchForDirectMessageOwner(
        ORDER BY rowid DESC LIMIT 1`
     )
     .get(runId, directHandle) as DispatchContextRow | undefined
-  if (exact || !paneKey || !parsePaneKey(paneKey)) {
+  if (exact) {
     return exact
   }
+  // A session address owns the Dispatch its session is assigned, as a handle owns its own.
+  if (parseOrchestrationActor(directHandle)) {
+    const byActor = this.db
+      .prepare(
+        `SELECT * FROM dispatch_contexts
+         WHERE run_id = ? AND assignee_actor = ? AND status IN ('pending', 'dispatched')
+         ORDER BY rowid DESC LIMIT 1`
+      )
+      .get(runId, directHandle)
+    if (byActor) {
+      // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: SELECT * over this table returns the row shape its schema and row type define, like every row cast in db/.
+      return byActor as DispatchContextRow
+    }
+  }
+  if (!paneKey || !parsePaneKey(paneKey)) {
+    return undefined
+  }
   return this.db
     .prepare(
       `SELECT * FROM dispatch_contexts
@@ -76,6 +94,31 @@ export function routeForeignDirectMessagesToOwnedMailboxes(
       [directHandle, ...exclusionParams, ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1],
       [directHandle, directHandle, ...exclusionParams, ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1]
     ]
+    if (parseOrchestrationActor(directHandle)) {
+      branches.push(
+        `SELECT candidate.id, candidate.run_id, candidate.type, candidate.sequence
+         FROM (
+           SELECT actor_dispatch.run_id
+           FROM dispatch_contexts AS actor_dispatch INDEXED BY idx_dispatch_assignee_actor
+           JOIN runs AS owner_run
+             ON owner_run.id = actor_dispatch.run_id AND owner_run.legacy = 0
+           WHERE actor_dispatch.assignee_actor = ?
+             AND actor_dispatch.status IN ('pending', 'dispatched')
+           GROUP BY actor_dispatch.run_id
+         ) AS actor_owner
+         JOIN messages AS candidate INDEXED BY idx_messages_undelivered_direct_run
+           ON candidate.run_id = actor_owner.run_id AND candidate.to_handle = ?
+         WHERE 1 = 1${runExclusion}
+           AND candidate.read = 0 AND candidate.delivered_at IS NULL
+           AND candidate.delivery_contract = 'current_delivery'`
+      )
+      branchParams.push([
+        directHandle,
+        directHandle,
+        ...exclusionParams,
+        ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1
+      ])
+    }
     if (paneSuffix !== undefined) {
       branches.push(
         `SELECT candidate.id, candidate.run_id, candidate.type, candidate.sequence

src/main/runtime/orchestration/db/runs/run-binding.ts

@@ -3,13 +3,20 @@ import { OrchestrationError } from '../../orchestration-error'
 import { LEGACY_CONTRACT_VERSION } from '../contract-constants'
 import { isEquivalentPaneKey } from '../pane-key-match'
 import type { OrchestrationDb } from '../orchestration-db'
+import {
+  addressSpellingsOf,
+  runBoundToCoordinator,
+  runCoordinatorKey
+} from '../../orchestration-caller-identity'
 
 export function bindRun(
   this: OrchestrationDb,
   params: {
     runId: string
-    coordinatorHandle: string
-    coordinatorPaneKey: string
+    coordinatorHandle: string | null
+    coordinatorPaneKey: string | null
+    /** `session:<id>` when the coordinator is a structured session; see orchestration-actor. */
+    coordinatorActor?: string | null
     takeoverLegacy?: boolean
     legacyCoordinatorAuthority?: {
       runId: string
@@ -20,16 +27,19 @@ export function bindRun(
     }
   }
 ): RunRow | undefined {
+  const coordinator = {
+    terminalHandle: params.coordinatorHandle,
+    paneKey: params.coordinatorPaneKey,
+    actor: params.coordinatorActor ?? null
+  }
   this.db.exec('BEGIN IMMEDIATE')
   try {
     const run = this.getRunRaw(params.runId)
     if (!run || run.legacy === 1) {
       this.db.exec('ROLLBACK')
       return undefined
     }
-    const sameBinding =
-      run.coordinator_pane_key !== null &&
-      isEquivalentPaneKey(run.coordinator_pane_key, params.coordinatorPaneKey)
+    const sameBinding = runBoundToCoordinator(run, coordinator)
     const adoption = this.getLegacyAdoption()
     const adoptedRun = adoption?.adopted_run_id === params.runId
     const legacyAuthority = params.legacyCoordinatorAuthority
@@ -49,6 +59,7 @@ export function bindRun(
       legacyPrincipal.terminal_handle === legacyAuthority.terminalHandle &&
       isEquivalentPaneKey(legacyPrincipal.pane_key, legacyAuthority.paneKey) &&
       params.coordinatorHandle === legacyAuthority.terminalHandle &&
+      params.coordinatorPaneKey !== null &&
       isEquivalentPaneKey(params.coordinatorPaneKey, legacyAuthority.paneKey)
     )
     if (legacyAuthority && !provenLegacyBinding) {
@@ -109,14 +120,14 @@ export function bindRun(
         }
       )
     }
-    this.unbindOtherRunsForPane(params.coordinatorPaneKey, params.runId)
-    for (const handle of new Set(
-      [run.coordinator_handle, params.coordinatorHandle].filter((value): value is string =>
-        Boolean(value)
-      )
-    )) {
-      this.rememberRunCoordinatorHandle(params.runId, handle)
-      this.routeAllUnreadDirectMessagesToRunMailbox(params.runId, handle)
+    this.unbindOtherRunsForCoordinator(coordinator, params.runId)
+    // Every address of the coordinator being replaced and of the one binding now.
+    for (const address of new Set([
+      ...addressSpellingsOf(runCoordinatorKey(run)),
+      ...addressSpellingsOf(coordinator)
+    ])) {
+      this.rememberRunCoordinatorHandle(params.runId, address)
+      this.routeAllUnreadDirectMessagesToRunMailbox(params.runId, address)
     }
     if (
       (params.takeoverLegacy && !takeoverAlreadyApplied) ||
@@ -128,26 +139,34 @@ export function bindRun(
           coordinatorPrincipal?.status === 'committed' &&
           (params.takeoverLegacy ||
             coordinatorPrincipal.terminal_handle !== params.coordinatorHandle ||
+            params.coordinatorPaneKey === null ||
             !isEquivalentPaneKey(coordinatorPrincipal.pane_key, params.coordinatorPaneKey))
         ) {
           this.setLegacyCompatibilityPrincipalStatus(coordinatorPrincipal.id, 'revoked')
         }
       }
-      // The actor belongs to the coordinator being replaced; nothing here resolves the new one's.
       this.db
         .prepare(
           `UPDATE runs
-           SET coordinator_handle = ?, coordinator_pane_key = ?, coordinator_actor = NULL,
-               coordinator_actor_generation = NULL,
+           SET coordinator_handle = ?, coordinator_pane_key = ?, coordinator_actor = ?,
+               coordinator_actor_generation = consumer_generation + 1,
                consumer_generation = consumer_generation + 1,
                updated_at = datetime('now')
            WHERE id = ?`
         )
-        .run(params.coordinatorHandle, params.coordinatorPaneKey, params.runId)
+        .run(coordinator.terminalHandle, coordinator.paneKey, coordinator.actor, params.runId)
       this.fenceUnacknowledgedMailboxDeliveries(`run:${params.runId}`)
       if (params.takeoverLegacy || replacesLegacyCoordinator) {
         this.promoteLegacyCoordinatorMailForTakeover(params.runId, retainedCoordinatorHandle)
       }
+    } else if (runCoordinatorKey(run).actor !== coordinator.actor) {
+      // Same coordinator, so no new consumer: correct an actor a writer without the column left.
+      this.db
+        .prepare(
+          `UPDATE runs SET coordinator_actor = ?, coordinator_actor_generation = consumer_generation
+           WHERE id = ?`
+        )
+        .run(coordinator.actor, params.runId)
     }
     this.db.exec('COMMIT')
   } catch (error) {