mirror of
https://github.com/stablyai/orca.git
synced 2026-11-01 09:37:33 +00:00
* feat(native-chat): add a chat-owned command receipt store Adds the agent_session_command_receipts table and an insert-if-absent receipt write that runs inside a command's own journal transaction. A receipt records the command's identity, canonical hash, status and a pointer to the row holding its result; it dies with its chat record. Nothing uses it yet: the existing operation ledger stays authoritative. * fix(native-chat): key command receipts like today's ledger ids Rows are keyed by caller and operation id; a send looks across callers and other methods look up only their caller's row, matching the existing ledger. Bad rows and newer table shapes read as unreadable instead of throwing. * fix(native-chat): order receipt lookups by insertion and key by operation id Global lookups now take rows in insertion order, so a read and an insert agree on which receipt answers an id regardless of the host clock. The primary key leads with the operation id, so its own index serves both lookups. * fix(native-chat): accept sends through their own receipt Send (direct, queued, steer) and /compact no longer commit a separate ledger reservation before the message is saved. Their command receipt is inserted-if-absent in the same transaction as the submission or queued draft; a repeated id replays from it, a changed request conflicts. Operation ids are checked for syntax only, so age, clock skew, pruning and the reservation's own storage can no longer refuse a send. The ledger success row they already co-committed stays as a temporary insert-if-absent write so families not yet switched keep seeing these ids; it is never read on this path. * fix(native-chat): answer a send's receipt one way, share /compact's id space with Send, wake delivery once - After its commit, a send whose publication fails answers through the same receipt replay a retry takes, so one id gets one answer; the in-memory copy and journal reload are gone. - /compact ids are global like Send's: a retry under a new connection replays instead of needing a second acknowledgment write. - The receipt route calls its runner directly; the runner wakes delivery when a new message row commits. - The compatibility ledger row expires from host time; each temporary site is labelled. - Receipt read errors and post-commit failures are logged. * test(native-chat): expect receipt answers where tests assumed the send ledger Restart-continuation tests now model the case Send cannot vouch for (saved but not loaded into memory) and keep it unconfirmed; a later throw after the message is in the chat answers continued, as a resend would. Drop /compact's unreachable legacy replay branch and fix stale comments. * test: run the queued tests that open the journal database in the Node runtime project * fix(native-chat): queued-message actions and prompt answers accept through their own receipt Send Now, Delete, Resume and prompt answers move onto the command-receipt route Send uses: the receipt is inserted with the command's own write. A no-op answers only once its receipt commits (otherwise outcome unknown), and a prompt answer's replay reads the exact revision it was accepted at. * fix(native-chat): replay a queued action from what its receipt recorded Send Now's acknowledgement points at the submission it handed off; a question answer's receipt keeps the accepted resolution; a no-op that cannot save its receipt answers outcome unknown and holds nothing in memory; classified refusals pass through. * docs(native-chat): receipt comments say what a replay answers from * fix(native-chat): refuse a prompt answer on a newer Orca's journal with the update words A prompt answer now runs through the same newer-Orca refusal the queued-message actions use, so a read-only journal answers agent_session_journal_unreadable instead of throwing journal_read_only.
170 lines
6.5 KiB
TypeScript
170 lines
6.5 KiB
TypeScript
// Admission for one mutating `agentSession.*` call.
|
|
//
|
|
// The rules themselves live in the durable ledger and the lease adjudicator;
|
|
// this is only the fixed order they are applied in, plus the payload
|
|
// fingerprint both peers derive from the same request fields. Nothing here
|
|
// re-derives who may write — that answer comes from
|
|
// `agentSessionLeaseAdmitsWriter` alone.
|
|
|
|
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'
|
|
|
|
/**
|
|
* Stable digest over the fields that define what this call DOES. Keys are
|
|
* emitted in sorted order at every depth so two peers serializing the same
|
|
* request in different property order agree, and an undefined field is dropped
|
|
* rather than hashed as present-but-empty.
|
|
*/
|
|
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(',')}}`
|
|
}
|
|
|
|
/**
|
|
* A retry whose payload changed is a different call wearing the same id.
|
|
* Checked BEFORE the ledger is consulted, so a refused call never leaves an
|
|
* admitted row that a later honest retry would replay as already-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 }
|
|
|
|
/**
|
|
* Fixed order: fingerprint agreement, then the ledger (so a retry replays
|
|
* before anything else can refuse it), then the lease.
|
|
*
|
|
* `expectedRuntimeFence` is not checked: each write names its own target (a
|
|
* turn, an item revision, an epoch) or is last-writer-wins, so an owner restart
|
|
* the client has not seen yet refuses nothing. 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
|
|
/** 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 }
|
|
}
|
|
if (input.conversationWrite) {
|
|
return { decision: 'admit', row: ledger.row }
|
|
}
|
|
const leaseRefusal = refuseUnlessWriterAdmitted(lease)
|
|
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.'
|
|
)
|
|
}
|