mirror of
https://github.com/stablyai/orca.git
synced 2026-11-15 18:37:24 +00:00
fix(native-chat): retain full work projections with saved Stop
Reuse the existing work-answer validity lookup when a narrow read returns.
A full status projection includes the saved Stop, while narrow work leaves
that overlay unrequested. Comparing only the authority work actually read
preserves a valid arriving full projection and retires stale authority or
context without a second cache or comparator.
Extend the held-read owner scenario with a real saved Stop. Private point
attribution shows unchanged full demand going from one reader point to
zero, with full and work outputs equal; invalidation reuse remains one
read followed by three cache hits. Owning 53 tests pass on both entries and
static gates pass. Frozen721 is parent plus two W3 files only; paused SQL
diagnostics work is excluded. Historical P7 benchmark costs remain small
measured overhead as corrected in 12d479e640 and report716.
This commit is contained in:
+34
-22
@@ -98,6 +98,33 @@ export class StructuredAgentSessionJournalProjections {
|
||||
: null
|
||||
}
|
||||
|
||||
private currentWorkSql(
|
||||
journal: SqlAgentSessionJournal,
|
||||
captured: CapturedWork
|
||||
): JournalSqlWorkResult | undefined {
|
||||
const authority = copyJournalCurrentWorkAuthority(captured.authority)
|
||||
const entry = this.currentEntry(journal)
|
||||
const answer: JournalSqlWorkResult | undefined =
|
||||
entry?.kind === 'work-only'
|
||||
? entry.contextId === captured.contextId
|
||||
? entry.answer
|
||||
: undefined
|
||||
: entry?.projection.work &&
|
||||
entry.projection.contextId === captured.contextId &&
|
||||
entry.projection.authority
|
||||
? {
|
||||
cursor: { epoch: entry.projection.epoch, sequence: entry.projection.sequence },
|
||||
physicalCursor: entry.physicalCursor,
|
||||
authority: copyJournalCurrentWorkAuthority(entry.projection.authority),
|
||||
result: { kind: 'state', state: entry.projection.work }
|
||||
}
|
||||
: undefined
|
||||
return answer &&
|
||||
journalSqlWorkAnswerIsCurrent(journal, () => captured.authority, authority, answer)
|
||||
? answer
|
||||
: undefined
|
||||
}
|
||||
|
||||
/** Narrow work and full status occupy the same validated per-journal entry. */
|
||||
async readWorkSql(journal: SqlAgentSessionJournal, capture: () => CapturedWork | null) {
|
||||
for (;;) {
|
||||
@@ -106,29 +133,11 @@ export class StructuredAgentSessionJournalProjections {
|
||||
this.byJournal.delete(journal)
|
||||
return null
|
||||
}
|
||||
const authority = copyJournalCurrentWorkAuthority(captured.authority)
|
||||
const entry = this.currentEntry(journal)
|
||||
const answer: JournalSqlWorkResult | undefined =
|
||||
entry?.kind === 'work-only'
|
||||
? entry.contextId === captured.contextId
|
||||
? entry.answer
|
||||
: undefined
|
||||
: entry?.projection.work &&
|
||||
entry.projection.contextId === captured.contextId &&
|
||||
entry.projection.authority
|
||||
? {
|
||||
cursor: { epoch: entry.projection.epoch, sequence: entry.projection.sequence },
|
||||
physicalCursor: entry.physicalCursor,
|
||||
authority: copyJournalCurrentWorkAuthority(entry.projection.authority),
|
||||
result: { kind: 'state', state: entry.projection.work }
|
||||
}
|
||||
: undefined
|
||||
if (
|
||||
answer &&
|
||||
journalSqlWorkAnswerIsCurrent(journal, () => captured.authority, authority, answer)
|
||||
) {
|
||||
const answer = this.currentWorkSql(journal, captured)
|
||||
if (answer) {
|
||||
return answer
|
||||
}
|
||||
const authority = copyJournalCurrentWorkAuthority(captured.authority)
|
||||
const result = await journal.reads.work({ authority, operation: { kind: 'state' } })
|
||||
const recaptured = capture()
|
||||
if (!recaptured) {
|
||||
@@ -141,7 +150,10 @@ export class StructuredAgentSessionJournalProjections {
|
||||
) {
|
||||
continue
|
||||
}
|
||||
if (!this.currentSql(journal, recaptured)) {
|
||||
if (
|
||||
this.currentEntry(journal)?.kind !== 'full' ||
|
||||
!this.currentWorkSql(journal, recaptured)
|
||||
) {
|
||||
this.byJournal.set(journal, {
|
||||
kind: 'work-only',
|
||||
answer: result,
|
||||
|
||||
+16
-4
@@ -123,6 +123,7 @@ it('discards row-free authority changes before publishing an asynchronous observ
|
||||
const projections = new StructuredAgentSessionJournalProjections()
|
||||
let revision = 0
|
||||
let contextId: string | null = null
|
||||
let latestStop: JournalWorkAuthority['stop'] = null
|
||||
let attached = true
|
||||
const held: string[] = []
|
||||
const capture = () => {
|
||||
@@ -142,8 +143,8 @@ it('discards row-free authority changes before publishing an asynchronous observ
|
||||
openedThrough: journal.openingCursor(),
|
||||
reopenFloor: null,
|
||||
operationalRevision: revision,
|
||||
stopRevision: 0,
|
||||
stop: null
|
||||
stopRevision: journal.stopMarks.revision(),
|
||||
stop: journal.stopMarks.withCurrentSettlement(latestStop)
|
||||
}
|
||||
return { authority, contextId }
|
||||
}
|
||||
@@ -234,6 +235,13 @@ it('discards row-free authority changes before publishing an asynchronous observ
|
||||
])
|
||||
expect(await projections.readSql(journal, capture, () => {})).toMatchObject({ workRevision: 2 })
|
||||
revision = 3
|
||||
await journal.appendStopEvent({ reason: 'user-stop' }, 1)
|
||||
latestStop = await journal.stopMarks.latest()
|
||||
expect(latestStop).not.toBeNull()
|
||||
const captureNarrow = () => {
|
||||
const captured = capture()!
|
||||
return { ...captured, authority: { ...captured.authority, stop: null } }
|
||||
}
|
||||
const workEntered = Promise.withResolvers<void>()
|
||||
const workRelease = Promise.withResolvers<void>()
|
||||
const workRead = journal.reads.work.bind(journal.reads)
|
||||
@@ -243,11 +251,15 @@ it('discards row-free authority changes before publishing an asynchronous observ
|
||||
await workRelease.promise
|
||||
return answer
|
||||
})
|
||||
const pendingWork = projections.readWorkSql(journal, capture)
|
||||
const pendingWork = projections.readWorkSql(journal, captureNarrow)
|
||||
await workEntered.promise
|
||||
const published = await projections.readSql(journal, capture, () => {})
|
||||
workRelease.resolve()
|
||||
await pendingWork
|
||||
expect(projections.currentSql(journal, capture())).toBe(published)
|
||||
expect(projections.savedStop(journal, contextId)).toEqual({ value: null })
|
||||
expect(await projections.readWorkSql(journal, captureNarrow)).toMatchObject({
|
||||
result: { kind: 'state' }
|
||||
})
|
||||
expect(projections.currentSql(journal, capture())).toBe(published)
|
||||
expect(projections.savedStop(journal, contextId)).toEqual({ value: latestStop })
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user