Pine/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 1 Wiki Activity

fix(schedule): reconcile latest master contracts

Browse Source
This commit is contained in:
Tianyi Cui
2026-08-08 22:39:07 +08:00
parent 31e83f8ee7
commit 1f88085b53
35 changed files with 260 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/host/apiproxy/src/api-proxy.ts
+19 -4
View File
@@ -1115,13 +1115,13 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
const persistence = ctx.get('sessionPersistence')
if (persistence !== undefined) {
try {
const stored = await persistence.inspect(sessionId)
const stored = await persistence.readFrom(sessionId, 0)
presentedThroughSeq = identityMatchingStoredPrefix(attached, events, stored)
} catch (error: unknown) {
// Attached history remains available from the live Session. A
// failed or not-yet-materialized inspection only withholds
// failed or not-yet-materialized physical read only withholds
// commit-gated event presentation sidecars.
ctx.logger.warn(`session.history: persistence inspection for attached "${sessionId}" failed; serving raw events: ${String(error)}`)
ctx.logger.warn(`session.history: physical persistence read for attached "${sessionId}" failed; serving raw events: ${String(error)}`)
}
}
return {
@@ -1133,10 +1133,25 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
}
const inspected = await inspectServable(sessionId)
const projections = includeProjections ? detachedProjectionsFor(ctx, inspected.events) : undefined
let presentedThroughSeq = 0
const persistence = ctx.get('sessionPersistence')
/* v8 ignore next -- inspectServable already rejects when persistence is absent */
if (persistence !== undefined) {
try {
const stored = await persistence.readFrom(sessionId, 0)
presentedThroughSeq = identityMatchingStoredPrefix(
{ header: inspected.meta },
inspected.events,
stored,
)
} catch (error: unknown) {
ctx.logger.warn(`session.history: physical persistence read for detached "${sessionId}" failed; serving raw events: ${String(error)}`)
}
}
return {
header: inspected.meta,
events: inspected.events,
presentedThroughSeq: inspected.events.length,
presentedThroughSeq,
...projections === undefined ? {} : { projections },
}
}
packages/host/apiproxy/tests/api-proxy-schedule-view.spec.ts
+46 -14
View File
@@ -83,7 +83,7 @@ describe('commit-aware Schedule live views', () => {
const ctx = await harness({
handler: () => ++calls === 1 ? first.promise : true,
})
const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
const api = createApiProxy(ctx, { defaultTarget: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp', workspaceRoot: '/tmp' })
const abort = new AbortController()
const collected = collectEvents(
api.events.mux({ rpcId: RpcId('schedule-live'), payload: {} }, abort.signal),
@@ -110,14 +110,14 @@ describe('commit-aware Schedule live views', () => {
for: 'event',
view: {
scheduleId: 'schedule-1', prompt: 'first',
occurrenceAt: '2026-08-05T12:00:01.000Z', deliveryMode: 'session-local',
occurrenceAt: '2026-08-05T12:00:01.000Z',
},
},
{
for: 'event',
view: {
scheduleId: 'schedule-2', prompt: 'second',
occurrenceAt: '2026-08-05T12:00:01.000Z', deliveryMode: 'session-local',
occurrenceAt: '2026-08-05T12:00:01.000Z',
},
},
])
@@ -130,7 +130,7 @@ describe('commit-aware Schedule live views', () => {
const ctx = await harness({
handler: () => ++calls === 1 ? Promise.reject(new Error('disk unavailable')) : true,
})
const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
const api = createApiProxy(ctx, { defaultTarget: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp', workspaceRoot: '/tmp' })
const abort = new AbortController()
const collected = collectEvents(
api.events.mux({ rpcId: RpcId('schedule-retry'), payload: {} }, abort.signal),
@@ -171,9 +171,9 @@ describe('Schedule history views', () => {
})
const child = ctx.sessions.fork(resumed, undefined, SessionId('schedule-fork'))
ctx.provide('sessionPersistence', {
inspect: () => Promise.resolve({ meta: child.header, events: [...child.events] }),
readFrom: () => Promise.resolve({ meta: child.header, events: [...child.events] }),
} as never)
const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
const api = createApiProxy(ctx, { defaultTarget: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp', workspaceRoot: '/tmp' })
const response = await api.sessions.history({
rpcId: RpcId('schedule-resumed-fork'), payload: { sessionId: child.id },
@@ -185,7 +185,6 @@ describe('Schedule history views', () => {
scheduleId,
prompt: 'after restart',
occurrenceAt: '2026-08-05T12:00:01.000Z',
deliveryMode: 'session-local',
},
})
await ctx.fiber.dispose()
@@ -199,14 +198,14 @@ describe('Schedule history views', () => {
seed: [...parent.events],
meta: { cwd: '/tmp', parentSession: parent.id, seedLength: 2 },
})
let inspect = (): Promise<{ meta: SessionHeader; events: SessionEvent[] }> => Promise.resolve({
let readFrom = (): Promise<{ meta: SessionHeader; events: SessionEvent[] }> => Promise.resolve({
meta: session.header,
events: [...session.events.slice(0, 1)],
})
ctx.provide('sessionPersistence', {
inspect: () => inspect(),
readFrom: () => readFrom(),
} as never)
const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
const api = createApiProxy(ctx, { defaultTarget: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp', workspaceRoot: '/tmp' })
const history = async () => {
const response = await api.sessions.history({
rpcId: RpcId('schedule-history'), payload: { sessionId: session.id },
@@ -216,19 +215,19 @@ describe('Schedule history views', () => {
}
expect((await history()).find(entry => entry.event.seq === 1)?.view).toBeUndefined()
inspect = () => Promise.resolve({
readFrom = () => Promise.resolve({
meta: { ...session.header, delegationDepth: 0 },
events: [...session.events.slice(0, 2)],
})
expect((await history()).find(entry => entry.event.seq === 1)?.view).toMatchObject({
for: 'event',
})
inspect = () => Promise.resolve({
readFrom = () => Promise.resolve({
meta: { ...session.header, cwd: '/different', delegationDepth: 0 },
events: [...session.events.slice(0, 2)],
})
expect((await history()).find(entry => entry.event.seq === 1)?.view).toBeUndefined()
inspect = () => Promise.reject(new Error('inspect unavailable'))
readFrom = () => Promise.reject(new Error('physical read unavailable'))
expect((await history()).find(entry => entry.event.seq === 1)?.view).toBeUndefined()
await ctx.fiber.dispose()
})
@@ -247,8 +246,9 @@ describe('Schedule history views', () => {
ctx.provide('sessionPersistence', {
list: () => Promise.resolve([meta]),
inspect: () => Promise.resolve({ meta, events }),
readFrom: () => Promise.resolve({ meta, events }),
} as never)
const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
const api = createApiProxy(ctx, { defaultTarget: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp', workspaceRoot: '/tmp' })
const response = await api.sessions.history({
rpcId: RpcId('schedule-cold'), payload: { sessionId: meta.id },
})
@@ -258,4 +258,36 @@ describe('Schedule history views', () => {
})
await ctx.fiber.dispose()
})
it('withholds a detached view that exists only in a logical inspection', async () => {
const ctx = await harness()
let source: Session | undefined
const owner = await ctx.plugin(Object.assign((inner: Context) => {
source = inner.sessions.create(SessionId('schedule-logical-only'), { meta: { cwd: '/tmp' } })
}, { inject: ['sessions'] }))
if (source === undefined) throw new Error('session owner did not publish its session')
appendReminder(source, 'schedule-logical', 'not physically committed')
const meta = source.header
const events = [...source.events]
await owner.dispose()
let physicalEvents = events.slice(0, 1)
ctx.provide('sessionPersistence', {
list: () => Promise.resolve([meta]),
inspect: () => Promise.resolve({ meta, events }),
readFrom: () => Promise.resolve({ meta, events: physicalEvents }),
} as never)
const api = createApiProxy(ctx, { defaultTarget: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp', workspaceRoot: '/tmp' })
const history = async () => {
const response = await api.sessions.history({
rpcId: RpcId('schedule-logical-only-history'), payload: { sessionId: meta.id },
})
if (!response.result.ok) throw new Error(response.result.error.message)
return response.result.value.events
}
expect((await history()).find(entry => entry.event.seq === 1)?.view).toBeUndefined()
physicalEvents = events
expect((await history()).find(entry => entry.event.seq === 1)?.view).toMatchObject({ for: 'event' })
await ctx.fiber.dispose()
})
})