mirror of
https://github.com/stablyai/orca.git
synced 2026-11-15 10:27:33 +00:00
perf(native-chat): batch background reply status publications (#42550)
* perf(native-chat): reuse deterministic session pane identity * perf(native-chat): batch streaming status facts between transitions * perf(native-chat): keep pending prose off focus reconciliation * test(native-chat): model live panes in switch regression fixture
This commit is contained in:
@@ -49,6 +49,9 @@ export const CONSUMERS = new Map([
|
||||
'ownerInputChanged',
|
||||
'ownerInputs',
|
||||
'createStructuredSessionMembership',
|
||||
'createStructuredSessionStatusBurst',
|
||||
'sameCapturedInput',
|
||||
'sameStatusInput',
|
||||
'sameSemanticTab'
|
||||
])
|
||||
}
|
||||
|
||||
+35
-28
@@ -1,3 +1,4 @@
|
||||
import { createStructuredSessionStatusBurst } from './structured-agent-session-status-burst'
|
||||
import { useEffect } from 'react'
|
||||
import { projectStatus } from './structured-agent-session-status-projection'
|
||||
import { getStructuredAgentSessionLaunchLifecycle } from '@/lib/agent-session/launch/structured-agent-session-launch-registry'
|
||||
@@ -27,32 +28,11 @@ import {
|
||||
type HostGroup,
|
||||
type Member,
|
||||
type ProjectionInput,
|
||||
sameSemanticTab
|
||||
sameSemanticTab,
|
||||
sameCapturedInput,
|
||||
sameStatusInput
|
||||
} from './structured-agent-session-status-membership'
|
||||
|
||||
/** Unchanged feed facts for an unchanged tab cannot change its projection or companions. */
|
||||
function sameCapturedInput(previous: ProjectionInput | undefined, next: ProjectionInput): boolean {
|
||||
return (
|
||||
previous !== undefined &&
|
||||
sameSemanticTab(previous.tab, next.tab) &&
|
||||
previous.summary === next.summary &&
|
||||
previous.observation === next.observation &&
|
||||
previous.failedAt === next.failedAt &&
|
||||
previous.sleptAt === next.sleptAt &&
|
||||
previous.directory === next.directory
|
||||
)
|
||||
}
|
||||
|
||||
function sameStatusInput(previous: ProjectionInput | undefined, next: ProjectionInput): boolean {
|
||||
return (
|
||||
previous !== undefined &&
|
||||
sameSemanticTab(previous.tab, next.tab) &&
|
||||
previous.summary === next.summary &&
|
||||
previous.observation === next.observation &&
|
||||
previous.failedAt === next.failedAt
|
||||
)
|
||||
}
|
||||
|
||||
function installStatusProjection(): () => void {
|
||||
const membership = createStructuredSessionMembership()
|
||||
const { members, membersByTabId, legacy } = membership
|
||||
@@ -192,6 +172,19 @@ function installStatusProjection(): () => void {
|
||||
membership.settle(input.member)
|
||||
}
|
||||
}
|
||||
const publish = (inputs: ProjectionInput[]): void => {
|
||||
try {
|
||||
project(inputs)
|
||||
} catch (error) {
|
||||
for (const input of inputs) {
|
||||
if (current(input)) {
|
||||
membership.owe(input.member)
|
||||
}
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
const burst = createStructuredSessionStatusBurst(publish, enqueue)
|
||||
const activateGroups = (): void => {
|
||||
for (const [hostKey, group] of groups) {
|
||||
if (groups.get(hostKey) === group && group.members.size && !group.release) {
|
||||
@@ -223,17 +216,25 @@ function installStatusProjection(): () => void {
|
||||
directory: member.hostKey === 'local' ? summary?.launchDirectory : undefined
|
||||
}
|
||||
// Launch ticks and full host ticks carry every member; skip the unchanged ones.
|
||||
if (!sameCapturedInput(member.captured, input)) {
|
||||
if (!sameCapturedInput(burst.peek(member) ?? member.captured, input)) {
|
||||
inputs.push(input)
|
||||
membership.owe(member)
|
||||
} else {
|
||||
membership.settle(member)
|
||||
}
|
||||
}
|
||||
enqueue(
|
||||
() => {
|
||||
activateGroups()
|
||||
project(inputs)
|
||||
try {
|
||||
activateGroups()
|
||||
burst.add(inputs.filter(current))
|
||||
} catch (error) {
|
||||
for (const input of inputs) {
|
||||
if (current(input)) {
|
||||
membership.owe(input.member)
|
||||
}
|
||||
}
|
||||
throw error
|
||||
}
|
||||
},
|
||||
inputs.map(({ member }) => ({ host: member.hostKey, paneKey: member.paneKey }))
|
||||
)
|
||||
@@ -280,7 +281,11 @@ function installStatusProjection(): () => void {
|
||||
}
|
||||
const place = (state: AppState, tab: StructuredTab, paneKey: string, changed: Set<Member>) => {
|
||||
const { hostKey, group } = groupFor(state, tab)
|
||||
const old = members.get(paneKey)
|
||||
const member = membership.place(tab, paneKey, hostKey, group)
|
||||
if (old && members.get(paneKey) !== old) {
|
||||
burst.forget(old)
|
||||
}
|
||||
if (member) {
|
||||
changed.add(member)
|
||||
}
|
||||
@@ -317,6 +322,7 @@ function installStatusProjection(): () => void {
|
||||
}
|
||||
}
|
||||
for (const member of retiring) {
|
||||
burst.forget(member)
|
||||
membership.detach(member)
|
||||
retired.add(member)
|
||||
}
|
||||
@@ -373,6 +379,7 @@ function installStatusProjection(): () => void {
|
||||
boundary(() => reconcile(false))
|
||||
return () => {
|
||||
disposed = true
|
||||
burst.dispose()
|
||||
unsubscribeApp()
|
||||
unsubscribeLaunch()
|
||||
for (const group of groups.values()) {
|
||||
|
||||
+96
@@ -0,0 +1,96 @@
|
||||
import { structuralValuesEqualIgnoringUndefined } from '../../../../../../shared/structural-value-equality'
|
||||
import {
|
||||
sameSemanticTab,
|
||||
type Member,
|
||||
type ProjectionInput
|
||||
} from './structured-agent-session-status-membership'
|
||||
|
||||
const STREAMING_BURST_WINDOW_MS = 33
|
||||
|
||||
function streamingOnly(previous: ProjectionInput | undefined, next: ProjectionInput): boolean {
|
||||
if (
|
||||
!previous ||
|
||||
previous.summary?.status !== 'working' ||
|
||||
next.summary?.status !== 'working' ||
|
||||
!sameSemanticTab(previous.tab, next.tab) ||
|
||||
previous.observation !== next.observation ||
|
||||
previous.failedAt !== next.failedAt ||
|
||||
previous.sleptAt !== next.sleptAt ||
|
||||
previous.directory !== next.directory
|
||||
) {
|
||||
return false
|
||||
}
|
||||
const {
|
||||
lastAssistantMessage: _previousText,
|
||||
updatedAt: _previousTime,
|
||||
...before
|
||||
} = previous.summary
|
||||
const { lastAssistantMessage: _nextText, updatedAt: _nextTime, ...after } = next.summary
|
||||
return structuralValuesEqualIgnoringUndefined(before, after)
|
||||
}
|
||||
|
||||
/** Retain only the latest prose tick per member; publish every semantic transition synchronously. */
|
||||
export function createStructuredSessionStatusBurst(
|
||||
publish: (inputs: ProjectionInput[]) => void,
|
||||
enqueue: (operation: () => void) => void
|
||||
) {
|
||||
const pending = new Map<Member, ProjectionInput>()
|
||||
let timer: ReturnType<typeof setTimeout> | null = null
|
||||
let lastPublishedAt = 0
|
||||
let disposed = false
|
||||
const cancelTimer = (): void => {
|
||||
if (timer !== null) {
|
||||
clearTimeout(timer)
|
||||
timer = null
|
||||
}
|
||||
}
|
||||
const apply = (inputs: ProjectionInput[]): void => {
|
||||
lastPublishedAt = Date.now()
|
||||
publish(inputs)
|
||||
}
|
||||
const flush = (): void => {
|
||||
cancelTimer()
|
||||
const inputs = [...pending.values()]
|
||||
pending.clear()
|
||||
if (!disposed && inputs.length) {
|
||||
apply(inputs)
|
||||
}
|
||||
}
|
||||
return {
|
||||
flush,
|
||||
peek: (member: Member): ProjectionInput | undefined => pending.get(member),
|
||||
add: (inputs: ProjectionInput[]): void => {
|
||||
if (disposed) {
|
||||
return
|
||||
}
|
||||
if (!inputs.length) {
|
||||
publish(inputs)
|
||||
return
|
||||
}
|
||||
if (inputs.some((input) => !streamingOnly(input.member.captured, input))) {
|
||||
flush()
|
||||
apply(inputs)
|
||||
return
|
||||
}
|
||||
if (timer === null && Date.now() - lastPublishedAt >= STREAMING_BURST_WINDOW_MS) {
|
||||
apply(inputs)
|
||||
return
|
||||
}
|
||||
for (const input of inputs) {
|
||||
pending.set(input.member, input)
|
||||
}
|
||||
timer ??= setTimeout(() => enqueue(flush), STREAMING_BURST_WINDOW_MS)
|
||||
},
|
||||
forget: (member: Member): void => {
|
||||
pending.delete(member)
|
||||
if (!pending.size) {
|
||||
cancelTimer()
|
||||
}
|
||||
},
|
||||
dispose: (): void => {
|
||||
disposed = true
|
||||
cancelTimer()
|
||||
pending.clear()
|
||||
}
|
||||
}
|
||||
}
|
||||
+29
@@ -45,6 +45,35 @@ export function sameSemanticTab(previous: StructuredTab, next: StructuredTab): b
|
||||
)
|
||||
}
|
||||
|
||||
/** Unchanged feed facts for an unchanged tab cannot change its projection or companions. */
|
||||
export function sameCapturedInput(
|
||||
previous: ProjectionInput | undefined,
|
||||
next: ProjectionInput
|
||||
): boolean {
|
||||
return (
|
||||
previous !== undefined &&
|
||||
sameSemanticTab(previous.tab, next.tab) &&
|
||||
previous.summary === next.summary &&
|
||||
previous.observation === next.observation &&
|
||||
previous.failedAt === next.failedAt &&
|
||||
previous.sleptAt === next.sleptAt &&
|
||||
previous.directory === next.directory
|
||||
)
|
||||
}
|
||||
|
||||
export function sameStatusInput(
|
||||
previous: ProjectionInput | undefined,
|
||||
next: ProjectionInput
|
||||
): boolean {
|
||||
return (
|
||||
previous !== undefined &&
|
||||
sameSemanticTab(previous.tab, next.tab) &&
|
||||
previous.summary === next.summary &&
|
||||
previous.observation === next.observation &&
|
||||
previous.failedAt === next.failedAt
|
||||
)
|
||||
}
|
||||
|
||||
/** Which chat tabs the status bridge projects, updated per changed tab rather than rebuilt. */
|
||||
export function createStructuredSessionMembership() {
|
||||
const members = new Map<string, Member>()
|
||||
|
||||
Reference in New Issue
Block a user