Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions src/main/runtime/orchestration/db/contract-constants.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ export function federatedStubHomeRunId(dispatchId: string): string {
export const LEGACY_CONTRACT_VERSION = 0
export const CURRENT_CONTRACT_VERSION = ORCHESTRATION_CONTRACT_VERSION

// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments, v28 durable local mutation caller identity, v31 dispatch/resource identity links, v32 bounded worker-terminal recovery metadata, v33 durable mailbox pointer Enter state, v34 role-addressed mailbox deliveries, v35 mailbox delivery default and index-predicate repair, v36 dispatch mailbox consumer generation, v37 recorded dispatch creator identity, v39 structured session journal archives.
// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments, v28 durable local mutation caller identity, v31 dispatch/resource identity links, v32 bounded worker-terminal recovery metadata, v33 durable mailbox pointer Enter state, v34 role-addressed mailbox deliveries, v35 mailbox delivery default and index-predicate repair, v36 dispatch mailbox consumer generation, v37 recorded dispatch creator identity, v39 structured session journal archives, v40 federated stub home Runs.
// v41: derive outstanding deliveries from unread messages.
export const SCHEMA_VERSION = 41
// v42: actor principal columns.
export const SCHEMA_VERSION = 42
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import { randomBytes, timingSafeEqual } from 'node:crypto'
import { principalFromPaneKey } from '../../../../../shared/orchestration-principal'
import { OrchestrationError } from '../../orchestration-error'
import { hashDispatchCapability } from '../dispatch-capability-hash'
import { isEquivalentPaneKey } from '../pane-key-match'
Expand Down Expand Up @@ -27,14 +28,16 @@ export function mintDispatchCapability(
this.db
.prepare(
`UPDATE dispatch_contexts
SET capability_hash = ?, assignee_pane_key = ?, process_incarnation = ?,
SET capability_hash = ?, assignee_pane_key = ?, assignee_principal = ?,
process_incarnation = ?,
capability_revoked_at = NULL,
consumer_generation = consumer_generation + 1
WHERE id = ?`
)
.run(
hashDispatchCapability(capability),
params.paneKey,
principalFromPaneKey(params.paneKey),
params.processIncarnation,
params.dispatchId
)
Expand Down
38 changes: 38 additions & 0 deletions src/main/runtime/orchestration/db/dispatch-depth.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -283,4 +283,42 @@ describe('nested worker depth', () => {
expect(row.process_incarnation).toBeNull()
expect(db.resolveCreatorDepth({ kind: 'terminal', handle: 'term_ctx' })).toBe(1)
})

it('pins the invariant "equal principals imply self-created": no nesting parent', () => {
// In PR1 this passes through the pane-key path; the resolver PR's read-switch to principal
// equality must keep it true, so it inherits this red test if it breaks the invariant.
db = new OrchestrationDb(':memory:')
const SELF_PANE = 'tab_self:11111111-1111-4111-8111-111111111111'
const task = db.createTask({ runId: 'run_legacy_local', spec: 'own bookkeeping' })
const row = db.createDispatchContext({
taskId: task.id,
assigneeHandle: 'term_self',
assigneePaneKey: SELF_PANE,
creator: { kind: 'terminal', handle: 'term_self', paneKey: SELF_PANE },
maxDepth: UNCAPPED
})
expect(row.creator_principal).toBe(`pane:${SELF_PANE}`)
expect(row.creator_principal).toBe(row.assignee_principal)
expect(
db.resolveCreatorDepth({ kind: 'terminal', handle: 'term_self', paneKey: SELF_PANE })
).toBe(0)

// A genuinely delegated row (different panes, different principals) still counts.
const delegated = db.createTask({ runId: 'run_legacy_local', spec: 'delegated' })
const worker = db.createDispatchContext({
taskId: delegated.id,
assigneeHandle: 'term_delegate',
assigneePaneKey: 'tab_delegate:22222222-2222-4222-8222-222222222222',
creator: { kind: 'terminal', handle: 'term_self', paneKey: SELF_PANE },
maxDepth: UNCAPPED
})
expect(worker.creator_principal).not.toBe(worker.assignee_principal)
expect(
db.resolveCreatorDepth({
kind: 'terminal',
handle: 'term_delegate',
paneKey: 'tab_delegate:22222222-2222-4222-8222-222222222222'
})
).toBe(1)
})
})
11 changes: 9 additions & 2 deletions src/main/runtime/orchestration/db/dispatch-depth.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import {
ROOT_DISPATCH_DEPTH,
nestedWorkerDepthExceededMessage
} from '../../../../shared/nested-worker-depth'
import { principalFromPaneKey } from '../../../../shared/orchestration-principal'
import { OrchestrationError } from '../orchestration-error'
import { isEquivalentPaneKey } from './pane-key-match'
import type { OrchestrationDb } from './orchestration-db'
Expand Down Expand Up @@ -32,11 +33,17 @@ export type DispatchCreator =
export function recordedCreatorIdentity(creator: DispatchCreator): {
creatorHandle: string | null
creatorPaneKey: string | null
creatorPrincipal: string | null
} {
if (creator.kind === 'system') {
return { creatorHandle: null, creatorPaneKey: null }
// Orca's in-process loop is not an actor row; a handle-only creator also stays NULL.
return { creatorHandle: null, creatorPaneKey: null, creatorPrincipal: null }
}
return {
creatorHandle: creator.handle,
creatorPaneKey: creator.paneKey ?? null,
creatorPrincipal: principalFromPaneKey(creator.paneKey)
}
return { creatorHandle: creator.handle, creatorPaneKey: creator.paneKey ?? null }
}

/**
Expand Down
17 changes: 12 additions & 5 deletions src/main/runtime/orchestration/db/dispatch-row-writer.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import type Database from '../../../sqlite/sync-database'
import { principalFromPaneKey } from '../../../../shared/orchestration-principal'
import { DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL } from './pane-key-match'

/**
Expand All @@ -14,11 +15,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_principal, process_incarnation,
creator_dispatch_id, creator_handle, creator_pane_key, creator_principal,
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 (
Expand All @@ -45,8 +46,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_principal,
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
Expand Down Expand Up @@ -74,6 +76,7 @@ export function claimDispatchContextRow(
creatorDispatchId?: string | null
creatorHandle?: string | null
creatorPaneKey?: string | null
creatorPrincipal?: string | null
priorFailures: number
depth: number
taskId: string
Expand All @@ -89,10 +92,12 @@ export function claimDispatchContextRow(
params.launchTokenHash,
params.assigneeHandle,
params.assigneePaneKey,
principalFromPaneKey(params.assigneePaneKey),
params.processIncarnation,
params.creatorDispatchId ?? null,
params.creatorHandle ?? null,
params.creatorPaneKey ?? null,
params.creatorPrincipal ?? null,
params.priorFailures,
params.depth,
params.taskId,
Expand All @@ -118,6 +123,7 @@ export function insertStartingDispatchContextRow(
creatorDispatchId?: string | null
creatorHandle?: string | null
creatorPaneKey?: string | null
creatorPrincipal?: string | null
}
): void {
assertStampedDepth(params.depth)
Expand All @@ -131,6 +137,7 @@ export function insertStartingDispatchContextRow(
params.creatorDispatchId ?? null,
params.creatorHandle ?? null,
params.creatorPaneKey ?? null,
params.creatorPrincipal ?? null,
params.depth
)
}
Expand Down
2 changes: 2 additions & 0 deletions src/main/runtime/orchestration/db/orchestration-db.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import {
} from './runs/run-coordinator-mail-routing'
import { createTables } from './schema/create-tables'
import { migrate } from './schema/migrate'
import { backfillPrincipalColumns } from './schema/principal-column-backfill'

class OrchestrationDbCore {
db: Database.Database
Expand All @@ -30,6 +31,7 @@ class OrchestrationDbCore {
createTables.call(this as unknown as OrchestrationDb)
migrate.call(this as unknown as OrchestrationDb)
backfillFederatedStubHomeRuns(this.db)
backfillPrincipalColumns(this.db)
createCoordinatorMailRoutingTrigger.call(this as unknown as OrchestrationDb)
rememberCurrentRunCoordinatorHandles.call(this as unknown as OrchestrationDb)
hardenOrchestrationDatabaseFiles(dbPath)
Expand Down
203 changes: 203 additions & 0 deletions src/main/runtime/orchestration/db/principal-dual-write.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,203 @@
import { afterEach, describe, expect, it } from 'vitest'
import { OrchestrationDb } from './orchestration-db'
import { structuredAgentSessionTabId } from '../../../../shared/structured-agent-session-tab-id'

const LEAF1 = '11111111-1111-4111-8111-111111111111'
const LEAF2 = '22222222-2222-4222-8222-222222222222'
const PANE_A = `tab_a:${LEAF1}`
const PANE_B = `tab_b:${LEAF2}`

describe('principal dual-write', () => {
let db: OrchestrationDb

afterEach(() => db?.close())

function readDispatch(id: string) {
return db.db

Check failure on line 16 in src/main/runtime/orchestration/db/principal-dual-write.test.ts

View workflow job for this annotation

GitHub Actions / static analysis

typescript(consistent-type-assertions)

Do not use any type assertions.
.prepare('SELECT assignee_principal, creator_principal FROM dispatch_contexts WHERE id = ?')
.get(id) as { assignee_principal: string | null; creator_principal: string | null }
}

it('createRun, bindRun, and unbind track the coordinator principal', () => {
db = new OrchestrationDb(':memory:')
const run = db.createRun({
objective: 'principals',
coordinatorHandle: 'term_a',
coordinatorPaneKey: PANE_A
})
expect(run.coordinator_principal).toBe(`pane:${PANE_A}`)

const rebound = db.bindRun({
runId: run.id,
coordinatorHandle: 'term_b',
coordinatorPaneKey: PANE_B
})
expect(rebound?.coordinator_principal).toBe(`pane:${PANE_B}`)

// A second Run claiming the same pane unbinds the first; a stale principal would misattribute
// the Run once reads flip.
db.createRun({ objective: 'usurper', coordinatorHandle: 'term_c', coordinatorPaneKey: PANE_B })
expect(db.getRunRaw(run.id)?.coordinator_principal).toBeNull()
expect(db.getRunRaw(run.id)?.coordinator_pane_key).toBeNull()
})

it('still caches both handles across a same-pane rebind', () => {
db = new OrchestrationDb(':memory:')
const run = db.createRun({
objective: 'remint',
coordinatorHandle: 'term_a',
coordinatorPaneKey: PANE_A
})
db.bindRun({ runId: run.id, coordinatorHandle: 'term_reminted', coordinatorPaneKey: PANE_A })
expect(
db.db
.prepare(
'SELECT terminal_handle FROM run_coordinator_handles WHERE run_id = ? ORDER BY terminal_handle'
)
.all(run.id)
).toEqual([{ terminal_handle: 'term_a' }, { terminal_handle: 'term_reminted' }])
})

it('classifies a structured coordinator pane key to its session principal at createRun', () => {
db = new OrchestrationDb(':memory:')
const paneKey = `${structuredAgentSessionTabId('sess_run')}:${LEAF1}`
const run = db.createRun({
objective: 'structured',
coordinatorHandle: 'structworker_a',
coordinatorPaneKey: paneKey
})
expect(run.coordinator_principal).toBe('session:sess_run')
// COALESCE: the handle wins the cache address while one exists; the principal takes over only
// for a handle-less session coordinator (a later PR's writer).
expect(
db.db
.prepare('SELECT terminal_handle FROM run_coordinator_handles WHERE run_id = ?')
.get(run.id)
).toEqual({ terminal_handle: 'structworker_a' })
})

it('createDispatchContext writes assignee and creator principals by classification', () => {
db = new OrchestrationDb(':memory:')
const task = db.createTask({ runId: 'run_legacy_local', spec: 'assigned' })
const withPane = db.createDispatchContext({
taskId: task.id,
assigneeHandle: 'term_w',
assigneePaneKey: PANE_A,
creator: { kind: 'terminal', handle: 'term_creator', paneKey: PANE_B },
maxDepth: Number.MAX_SAFE_INTEGER
})
expect(readDispatch(withPane.id)).toEqual({
assignee_principal: `pane:${PANE_A}`,
creator_principal: `pane:${PANE_B}`
})

const task2 = db.createTask({ runId: 'run_legacy_local', spec: 'bare' })
const bare = db.createDispatchContext({
taskId: task2.id,
assigneeHandle: 'term_bare',
creator: { kind: 'system' },
maxDepth: Number.MAX_SAFE_INTEGER
})
expect(readDispatch(bare.id)).toEqual({ assignee_principal: null, creator_principal: null })
})

it('classifies a structured assignee pane key at the writer chokepoint', () => {
db = new OrchestrationDb(':memory:')
const task = db.createTask({ runId: 'run_legacy_local', spec: 'structured worker' })
const paneKey = `${structuredAgentSessionTabId('sess_w')}:${LEAF1}`
const dispatch = db.createDispatchContext({
taskId: task.id,
assigneeHandle: 'structworker_w',
assigneePaneKey: paneKey,
creator: { kind: 'system' },
maxDepth: Number.MAX_SAFE_INTEGER
})
expect(readDispatch(dispatch.id).assignee_principal).toBe('session:sess_w')
})

it('a terminal creator with a handle but no pane key records a NULL creator principal', () => {
db = new OrchestrationDb(':memory:')
const task = db.createTask({ runId: 'run_legacy_local', spec: 'handle-only creator' })
const dispatch = db.createDispatchContext({
taskId: task.id,
assigneeHandle: 'term_w',
creator: { kind: 'terminal', handle: 'term_creator' },
maxDepth: Number.MAX_SAFE_INTEGER
})
expect(readDispatch(dispatch.id).creator_principal).toBeNull()
})

it('worker-start leaves assignee NULL until authority attaches, then tracks re-pointing', () => {
db = new OrchestrationDb(':memory:')
const { dispatch } = db.createStartingWorkerDispatch({
taskSpec: 'supervised',
taskRunId: 'run_legacy_local',
startOptions: {},
creator: { kind: 'terminal', handle: 'term_creator', paneKey: PANE_B },
maxDepth: Number.MAX_SAFE_INTEGER
})
expect(readDispatch(dispatch.id)).toEqual({
assignee_principal: null,
creator_principal: `pane:${PANE_B}`
})

db.prepareStartingWorkerAuthority({
dispatchId: dispatch.id,
handle: 'term_worker',
paneKey: PANE_A,
processIncarnation: 'inc-1',
worktreeId: 'wt-1',
effects: [],
setupState: 'not_applicable',
terminalOwnership: 'created'
})
expect(readDispatch(dispatch.id)).toEqual({
assignee_principal: `pane:${PANE_A}`,
creator_principal: `pane:${PANE_B}`
})

// Re-pointing the Dispatch moves the assignee principal and leaves the creator alone.
db.mintDispatchCapability({
dispatchId: dispatch.id,
paneKey: PANE_B,
processIncarnation: 'inc-2'
})
expect(readDispatch(dispatch.id)).toEqual({
assignee_principal: `pane:${PANE_B}`,
creator_principal: `pane:${PANE_B}`
})
})

it('worker terminal resources track pane identity through create and transfer', () => {
db = new OrchestrationDb(':memory:')
const owned = db.createWorkerTerminalResourceStatement({
dispatchId: 'ctx_owner',
worktreeId: 'wt-1',
terminalHandle: 'term_w',
paneKey: PANE_A,
processIncarnation: 'inc-1',
ownership: 'owned'
})
expect(owned.principal).toBe(`pane:${PANE_A}`)

const external = db.createWorkerTerminalResourceStatement({
dispatchId: 'ctx_external',
worktreeId: 'wt-1',
terminalHandle: 'term_e',
paneKey: null,
processIncarnation: null,
ownership: 'external'
})
expect(external.principal).toBeNull()

const transferred = db.transferWorkerTerminalResourceStatement({
resourceId: owned.id,
toDispatchId: 'ctx_next',
terminalHandle: 'term_w2',
paneKey: PANE_B,
processIncarnation: 'inc-2',
hostScope: null
})
expect(transferred.principal).toBe(`pane:${PANE_B}`)
})
})
Loading
Loading