orca/src/shared/agent-session-mutation-envelope.ts

// Apply ledger/lease rules in fixed order with a shared payload fingerprint; agentSessionLeaseAdmitsWriter alone decides writer authority.

import { createHash } from 'node:crypto'
import type {
  AgentSessionOperationDecision,
  AgentSessionOperationRow
} from './agent-session-operation-ledger'
import { agentSessionLeaseAdmitsWriter } from './agent-session-lease-adjudication'
import type { AgentSessionLease } from './agent-session-record'
import { terminalOwnerRefusalMessage } from './agent-session-legacy-handoff-lease'
import type { AgentSessionMutationEnvelope, AgentSessionWireRefusal } from './agent-session-wire'
import { refuse } from './agent-session-wire-refusals'

/** Sort keys recursively and omit undefined fields so both peers hash equivalent request fields identically. */
export function computeAgentSessionPayloadFingerprint(input: {
  method: string
  sessionId: string
  fields: Record<string, unknown>
}): string {
  return canonicalAgentSessionDigest({
    method: input.method,
    sessionId: input.sessionId,
    fields: input.fields
  })
}

/** The same digest for an operation that has no session to name — a launch decides which surface it
 *  gets, so it has no session id until after it runs. */
export function canonicalAgentSessionDigest(value: Record<string, unknown>): string {
  return createHash('sha256').update(canonicalize(value)).digest('hex')
}

function canonicalize(value: unknown): string {
  if (value === null || typeof value !== 'object') {
    return JSON.stringify(value ?? null)
  }
  if (Array.isArray(value)) {
    return `[${value.map(canonicalize).join(',')}]`
  }
  const entries = Object.entries(value as Record<string, unknown>)
    .filter(([, entry]) => entry !== undefined)
    .sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0))
  return `{${entries.map(([key, entry]) => `${JSON.stringify(key)}:${canonicalize(entry)}`).join(',')}}`
}

/** Reject changed-payload retries before consulting the ledger, avoiding an admitted row an honest retry could replay as done. */
export function agentSessionFingerprintConflict(
  envelope: AgentSessionMutationEnvelope,
  hostFingerprint: string
): AgentSessionWireRefusal | null {
  return envelope.payloadFingerprint === hostFingerprint
    ? null
    : refuse(
        'agent_session_operation_conflict',
        { reason: 'fingerprintMismatch' },
        'The payload does not match the fingerprint the client declared for this operation.'
      )
}

export type AgentSessionMutationAdmission =
  | { decision: 'admit'; row: AgentSessionOperationRow }
  /** The recorded outcome answers this call; do not run the effect again. */
  | { decision: 'replay'; row: AgentSessionOperationRow }
  | { decision: 'refused'; refusal: AgentSessionWireRefusal }

/** Check fingerprint, then ledger replay, then lease. Ignore expectedRuntimeFence: writes name their target or are last-writer-wins, so unseen owner restarts do not refuse them; older hosts still check it. */
export function admitAgentSessionMutation(input: {
  envelope: AgentSessionMutationEnvelope
  /** Fingerprint the host computed from the request it actually received. */
  hostFingerprint: string
  /** Decision from the durable ledger, evaluated under `hostFingerprint`. */
  ledger: AgentSessionOperationDecision
  lease: AgentSessionLease | (() => AgentSessionLease)
  /** A write to the conversation, not to the provider child: a send is accepted and a Stop
   *  withdraws queued messages whoever owns the child, so the lease does not admit them. */
  conversationWrite?: true
}): AgentSessionMutationAdmission {
  const { envelope, lease, ledger } = input
  const mismatch = agentSessionFingerprintConflict(envelope, input.hostFingerprint)
  if (mismatch) {
    return { decision: 'refused', refusal: mismatch }
  }
  if (ledger.decision === 'refused') {
    return { decision: 'refused', refusal: agentSessionLedgerRefusal(envelope, ledger) }
  }
  if (ledger.decision === 'replay') {
    return { decision: 'replay', row: ledger.row }
  }
  const currentLease = typeof lease === 'function' ? lease() : lease
  if (input.conversationWrite) {
    return { decision: 'admit', row: ledger.row }
  }
  const leaseRefusal = refuseUnlessWriterAdmitted(currentLease)
  if (leaseRefusal) {
    return { decision: 'refused', refusal: leaseRefusal }
  }
  return { decision: 'admit', row: ledger.row }
}

/** What the ledger's own refusal of an operation id answers. */
export function agentSessionLedgerRefusal(
  envelope: Pick<AgentSessionMutationEnvelope, 'clientOperationId'>,
  ledger: Extract<AgentSessionOperationDecision, { decision: 'refused' }>
): AgentSessionWireRefusal {
  return refuse(
    ledger.code,
    ledger.details,
    `Operation ${envelope.clientOperationId} was refused: ${ledger.code}.`
  )
}

/** Why the single admission oracle said no, mapped to what the client can do
 *  about it. The predicate itself is never re-implemented here. */
export function refuseUnlessWriterAdmitted(
  lease: AgentSessionLease
): AgentSessionWireRefusal | null {
  if (agentSessionLeaseAdmitsWriter(lease)) {
    return null
  }
  if (lease.unreconciled) {
    return refuse(
      'execution_owner_reconciling',
      { reason: 'hostReconciling' },
      'This host has not yet adjudicated the session lease.'
    )
  }
  if (lease.handoffStage !== null) {
    if (lease.claimStatus === 'conflicted') {
      return refuse(
        'agent_session_conflict',
        { reason: 'claimConflicted' },
        terminalOwnerRefusalMessage(lease)
      )
    }
    return lease.handoffStage === 'new-owner-proving'
      ? refuse('agent_session_conflict', { reason: 'chatStarting' }, 'The chat is still starting.')
      : refuse(
          'agent_session_conflict',
          { reason: 'ownerUnproven' },
          "Orca has not yet confirmed that this chat's previous agent process stopped. Reopen the chat to check again."
        )
  }
  return refuse(
    'agent_session_ownership_unknown',
    { reason: 'noLiveOwner' },
    'The session has no live owner to accept writes.'
  )
}