mirror of
https://github.com/wu736139669/hapi.git
synced 2026-08-05 06:24:37 +00:00
fix(web,hub): queued bar SSE + never-started inactive resume (#761)
* fix(web): apply messages-consumed on global SSE connection The global all-sessions SSE subscription returned early on message-stream events without updating the message-window store. When session-scoped SSE was reconnecting or the user had another session selected, messages-consumed never cleared the queued bar even though the hub had stamped invoked_at. Also harden mergeMessages so a stale invokedAt:null snapshot cannot clobber an existing ack timestamp. Fixes tiann/hapi#758 Co-authored-by: Cursor <cursoragent@cursor.com> * fix(web,hub): resume never-started inactive sessions on first send Hub fresh-spawns when inactive session has path but no agent thread id and zero messages. Web guards resume, updates inactive banner copy, and surfaces resume_unavailable before POST /resume when resume is impossible. Fixes tiann/hapi#759 Co-authored-by: Cursor <cursoragent@cursor.com> * fix(web): scope sessionResume guard to current flavor only Hub `resolveAgentResumeId` only honors the metadata.flavor's id; the web guard was falling back across all flavors so a cursor session with a stale codexSessionId still tried to resume and 409'd. Mirror the hub switch and default to claude when flavor is unknown. Addresses HAPI Bot review on tiann/hapi#761. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(web): allow claude session resume via hub message-id recovery Hub `resolveAgentResumeId` falls back to `recoverClaudeSessionIdFromMessages` on the claude branch when `metadata.claudeSessionId` is absent, so the web guard must not block inactive claude sessions that have stored messages but no metadata id. Other flavors have no such recovery path and stay rejected. Addresses second HAPI Bot review thread on tiann/hapi#761 (`web/src/lib/sessionResume.ts:41`). Co-authored-by: Cursor <cursoragent@cursor.com> --------- Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
HeavyGee
@@ -1148,6 +1148,61 @@ describe('session model', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('resumeSession fresh-spawns when inactive cursor session has no agent id and no user messages', async () => {
|
||||
const store = new Store(':memory:')
|
||||
const engine = new SyncEngine(
|
||||
store,
|
||||
{} as never,
|
||||
new RpcRegistry(),
|
||||
{ broadcast() {} } as never
|
||||
)
|
||||
|
||||
try {
|
||||
const session = engine.getOrCreateSession(
|
||||
'never-started-cursor',
|
||||
{
|
||||
path: '/tmp/project',
|
||||
host: 'localhost',
|
||||
machineId: 'machine-1',
|
||||
flavor: 'cursor'
|
||||
},
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
engine.getOrCreateMachine(
|
||||
'machine-1',
|
||||
{ host: 'localhost', platform: 'linux', happyCliVersion: '0.1.0' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
engine.handleMachineAlive({ machineId: 'machine-1', time: Date.now() })
|
||||
|
||||
let capturedResumeSessionId: string | undefined = 'unset'
|
||||
;(engine as any).rpcGateway.spawnSession = async (
|
||||
_machineId: string,
|
||||
_directory: string,
|
||||
_agent: string,
|
||||
_model?: string,
|
||||
_modelReasoningEffort?: string,
|
||||
_yolo?: boolean,
|
||||
_sessionType?: string,
|
||||
_worktreeName?: string,
|
||||
resumeSessionId?: string
|
||||
) => {
|
||||
capturedResumeSessionId = resumeSessionId
|
||||
return { type: 'success', sessionId: session.id }
|
||||
}
|
||||
;(engine as any).waitForSessionActive = async () => true
|
||||
|
||||
const result = await engine.resumeSession(session.id, 'default')
|
||||
|
||||
expect(result).toEqual({ type: 'success', sessionId: session.id })
|
||||
expect(capturedResumeSessionId).toBeUndefined()
|
||||
} finally {
|
||||
engine.stop()
|
||||
}
|
||||
})
|
||||
|
||||
it('includes first user message in local resumable sessions', () => {
|
||||
const store = new Store(':memory:')
|
||||
const engine = new SyncEngine(
|
||||
|
||||
@@ -617,6 +617,18 @@ export class SyncEngine {
|
||||
return undefined
|
||||
}
|
||||
|
||||
/** Inactive session with directory path but no agent thread and no prior user turn. */
|
||||
private canFreshSpawnNeverStartedSession(session: Session, sessionId: string, namespace: string): boolean {
|
||||
const metadata = session.metadata
|
||||
if (!metadata || typeof metadata.path !== 'string' || metadata.path.length === 0) {
|
||||
return false
|
||||
}
|
||||
if (this.resolveAgentResumeId(session, namespace)) {
|
||||
return false
|
||||
}
|
||||
return this.store.messages.getFirstMessages(sessionId, 1).length === 0
|
||||
}
|
||||
|
||||
async resumeSession(sessionId: string, namespace: string, opts?: { permissionMode?: PermissionMode }): Promise<ResumeSessionResult> {
|
||||
const access = this.sessionCache.resolveSessionAccess(sessionId, namespace)
|
||||
if (!access.ok) {
|
||||
@@ -633,14 +645,27 @@ export class SyncEngine {
|
||||
}
|
||||
|
||||
const targetResult = this.resolveLocalResumeTarget(access.sessionId, namespace)
|
||||
if (targetResult.type === 'error') {
|
||||
let flavor: AgentFlavor
|
||||
let resumeToken: string | undefined
|
||||
let directory: string
|
||||
|
||||
if (targetResult.type === 'success') {
|
||||
flavor = targetResult.target.flavor
|
||||
resumeToken = targetResult.target.agentSessionId
|
||||
directory = targetResult.target.directory
|
||||
} else if (
|
||||
targetResult.code === 'resume_unavailable'
|
||||
&& this.canFreshSpawnNeverStartedSession(session, access.sessionId, namespace)
|
||||
) {
|
||||
const metadata = session.metadata!
|
||||
flavor = this.resolveFlavor(session)
|
||||
resumeToken = undefined
|
||||
directory = metadata.path
|
||||
} else {
|
||||
return targetResult
|
||||
}
|
||||
|
||||
const target = targetResult.target
|
||||
const metadata = session.metadata!
|
||||
const flavor = target.flavor
|
||||
const resumeToken = target.agentSessionId
|
||||
|
||||
const onlineMachines = this.machineCache.getOnlineMachinesByNamespace(namespace)
|
||||
if (onlineMachines.length === 0) {
|
||||
@@ -668,7 +693,7 @@ export class SyncEngine {
|
||||
?? session.metadata?.preferredPermissionMode
|
||||
const spawnResult = await this.rpcGateway.spawnSession(
|
||||
targetMachine.id,
|
||||
target.directory,
|
||||
directory,
|
||||
flavor,
|
||||
session.model ?? undefined,
|
||||
session.modelReasoningEffort ?? undefined,
|
||||
|
||||
@@ -18,6 +18,7 @@ import { reconcileChatBlocks } from '@/chat/reconcile'
|
||||
import { buildConversationOutline } from '@/chat/outline'
|
||||
import { buildVisibleChatBlocks, isToolGroupBlock, type ToolGroupBlock } from '@/chat/toolGroups'
|
||||
import { isQueuedForInvocation, mergeMessages } from '@/lib/messages'
|
||||
import { inactiveSessionCanResume } from '@/lib/sessionResume'
|
||||
import { HappyComposer } from '@/components/AssistantChat/HappyComposer'
|
||||
import type { PendingSchedule } from '@/components/AssistantChat/ScheduleTimePicker'
|
||||
import { resolvePendingSchedule } from '@/components/AssistantChat/ScheduleTimePicker'
|
||||
@@ -125,6 +126,7 @@ export function SessionChat(props: {
|
||||
const { t } = useTranslation()
|
||||
const navigate = useNavigate()
|
||||
const sessionInactive = !props.session.active
|
||||
const inactiveCanResume = inactiveSessionCanResume(props.session, props.messages.length)
|
||||
const terminalSupported = isRemoteTerminalSupported(props.session.metadata)
|
||||
const normalizedCacheRef = useRef<Map<string, { source: DecryptedMessage; normalized: NormalizedMessage | null }>>(new Map())
|
||||
const blocksByIdRef = useRef<Map<string, ChatBlock>>(new Map())
|
||||
@@ -560,7 +562,9 @@ export function SessionChat(props: {
|
||||
{sessionInactive ? (
|
||||
<div className="px-3 pt-3">
|
||||
<div className="mx-auto w-full max-w-content rounded-md bg-[var(--app-subtle-bg)] p-3 text-sm text-[var(--app-hint)]">
|
||||
Session is inactive. Sending will resume it automatically.
|
||||
{inactiveCanResume
|
||||
? t('session.inactive.autoResume')
|
||||
: t('session.inactive.cannotResume')}
|
||||
</div>
|
||||
</div>
|
||||
) : null}
|
||||
|
||||
@@ -458,6 +458,16 @@ export function useSSE(options: {
|
||||
) {
|
||||
queueSessionListInvalidation()
|
||||
}
|
||||
// The global `all` subscription also receives message-stream events.
|
||||
// Session-scoped SSE normally drives the message window, but during
|
||||
// reconnect gaps or while another session is selected, only the global
|
||||
// connection may be alive — still clear the queued bar / optimistic rows.
|
||||
if (event.type === 'messages-consumed') {
|
||||
markMessagesConsumed(event.sessionId, event.localIds, event.invokedAt)
|
||||
}
|
||||
if (event.type === 'message-cancelled') {
|
||||
removeOptimisticMessage(event.sessionId, event.messageId)
|
||||
}
|
||||
onEventRef.current(event)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -362,6 +362,9 @@ export default {
|
||||
'send.blocked.title': 'Cannot send message',
|
||||
'send.blocked.noConnection': 'Not connected to server',
|
||||
'resume.failed.title': 'Resume failed',
|
||||
'resume.unavailable.noTarget': 'This session cannot be resumed. Start a new session in this directory.',
|
||||
'session.inactive.autoResume': 'Session is inactive. Sending will resume it automatically.',
|
||||
'session.inactive.cannotResume': 'Session is inactive and cannot be resumed from here. Start a new session in this directory.',
|
||||
'toast.ready.title': 'Ready for input',
|
||||
'toast.ready.body': '{agent} is waiting in {session}',
|
||||
'toast.permission.title': 'Permission Request',
|
||||
|
||||
@@ -364,6 +364,9 @@ export default {
|
||||
'send.blocked.title': '无法发送消息',
|
||||
'send.blocked.noConnection': '未连接到服务器',
|
||||
'resume.failed.title': '恢复会话失败',
|
||||
'resume.unavailable.noTarget': '无法恢复此会话。请在该目录下新建会话。',
|
||||
'session.inactive.autoResume': '会话已停用。发送消息将自动恢复会话。',
|
||||
'session.inactive.cannotResume': '会话已停用且无法在此恢复。请在该目录下新建会话。',
|
||||
'toast.ready.title': '等待输入',
|
||||
'toast.ready.body': '{agent} 正在 {session} 等待你的输入',
|
||||
'toast.permission.title': '权限请求',
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import type { DecryptedMessage } from '@/types/api'
|
||||
import { mergeMessages } from '@/lib/messages'
|
||||
|
||||
function userMessage(partial: Partial<DecryptedMessage> & { id: string }): DecryptedMessage {
|
||||
return {
|
||||
id: partial.id,
|
||||
localId: partial.localId ?? partial.id,
|
||||
seq: partial.seq ?? 1,
|
||||
createdAt: partial.createdAt ?? 1_000,
|
||||
invokedAt: partial.invokedAt ?? null,
|
||||
status: partial.status,
|
||||
content: { role: 'user', content: [{ type: 'text', text: 'hi' }] },
|
||||
}
|
||||
}
|
||||
|
||||
describe('mergeMessages', () => {
|
||||
it('preserves invokedAt when a stale snapshot omits the ack timestamp', () => {
|
||||
const invokedAt = 2_000
|
||||
const existing = [userMessage({ id: 'server-1', localId: 'local-1', invokedAt })]
|
||||
const incoming = [userMessage({ id: 'server-1', localId: 'local-1', invokedAt: null })]
|
||||
|
||||
const merged = mergeMessages(existing, incoming)
|
||||
expect(merged).toHaveLength(1)
|
||||
expect(merged[0]?.invokedAt).toBe(invokedAt)
|
||||
})
|
||||
})
|
||||
@@ -56,7 +56,12 @@ export function mergeMessages(existing: DecryptedMessage[], incoming: DecryptedM
|
||||
byId.set(msg.id, msg)
|
||||
}
|
||||
for (const msg of incoming) {
|
||||
byId.set(msg.id, msg)
|
||||
const existing = byId.get(msg.id)
|
||||
if (existing && existing.invokedAt != null && msg.invokedAt == null) {
|
||||
byId.set(msg.id, { ...msg, invokedAt: existing.invokedAt })
|
||||
} else {
|
||||
byId.set(msg.id, msg)
|
||||
}
|
||||
}
|
||||
|
||||
let merged = Array.from(byId.values())
|
||||
|
||||
@@ -0,0 +1,114 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import type { Session } from '@/types/api'
|
||||
import { inactiveSessionCanResume, resolveAgentSessionIdFromMetadata } from './sessionResume'
|
||||
|
||||
function makeSession(overrides: Partial<Session> = {}): Session {
|
||||
return {
|
||||
id: 'session-1',
|
||||
active: false,
|
||||
thinking: false,
|
||||
activeAt: 0,
|
||||
updatedAt: 0,
|
||||
metadata: { path: '/tmp/project', host: 'localhost', flavor: 'cursor' },
|
||||
...overrides,
|
||||
} as Session
|
||||
}
|
||||
|
||||
describe('sessionResume', () => {
|
||||
it('resolveAgentSessionIdFromMetadata picks the id matching the session flavor', () => {
|
||||
expect(resolveAgentSessionIdFromMetadata({
|
||||
path: '/p',
|
||||
host: 'h',
|
||||
flavor: 'codex',
|
||||
codexSessionId: 'codex-1',
|
||||
cursorSessionId: 'cursor-1',
|
||||
})).toBe('codex-1')
|
||||
expect(resolveAgentSessionIdFromMetadata({
|
||||
path: '/p',
|
||||
host: 'h',
|
||||
flavor: 'cursor',
|
||||
cursorSessionId: 'cursor-1',
|
||||
})).toBe('cursor-1')
|
||||
})
|
||||
|
||||
it('resolveAgentSessionIdFromMetadata ignores stale cross-flavor ids', () => {
|
||||
expect(resolveAgentSessionIdFromMetadata({
|
||||
path: '/p',
|
||||
host: 'h',
|
||||
flavor: 'cursor',
|
||||
codexSessionId: 'codex-1',
|
||||
})).toBeUndefined()
|
||||
})
|
||||
|
||||
it('resolveAgentSessionIdFromMetadata defaults to claude when flavor is missing', () => {
|
||||
expect(resolveAgentSessionIdFromMetadata({
|
||||
path: '/p',
|
||||
host: 'h',
|
||||
claudeSessionId: 'claude-1',
|
||||
})).toBe('claude-1')
|
||||
})
|
||||
|
||||
it('inactiveSessionCanResume is true for active sessions', () => {
|
||||
expect(inactiveSessionCanResume(makeSession({ active: true }), 0)).toBe(true)
|
||||
})
|
||||
|
||||
it('inactiveSessionCanResume allows fresh spawn when no agent id and no messages', () => {
|
||||
expect(inactiveSessionCanResume(makeSession(), 0)).toBe(true)
|
||||
})
|
||||
|
||||
it('inactiveSessionCanResume allows resume when agent id exists', () => {
|
||||
expect(inactiveSessionCanResume(makeSession({
|
||||
metadata: {
|
||||
path: '/tmp/project',
|
||||
host: 'localhost',
|
||||
flavor: 'cursor',
|
||||
cursorSessionId: 'cursor-thread-1',
|
||||
},
|
||||
}), 5)).toBe(true)
|
||||
})
|
||||
|
||||
it('inactiveSessionCanResume rejects inactive sessions with messages but no agent id', () => {
|
||||
expect(inactiveSessionCanResume(makeSession(), 3)).toBe(false)
|
||||
})
|
||||
|
||||
it('inactiveSessionCanResume rejects when stale cross-flavor agent id is present but no messages', () => {
|
||||
expect(inactiveSessionCanResume(makeSession({
|
||||
metadata: {
|
||||
path: '/tmp/project',
|
||||
host: 'localhost',
|
||||
flavor: 'cursor',
|
||||
codexSessionId: 'stale-codex-1',
|
||||
},
|
||||
}), 0)).toBe(true)
|
||||
expect(inactiveSessionCanResume(makeSession({
|
||||
metadata: {
|
||||
path: '/tmp/project',
|
||||
host: 'localhost',
|
||||
flavor: 'cursor',
|
||||
codexSessionId: 'stale-codex-1',
|
||||
},
|
||||
}), 3)).toBe(false)
|
||||
})
|
||||
|
||||
it('inactiveSessionCanResume rejects when metadata path is missing', () => {
|
||||
expect(inactiveSessionCanResume(makeSession({ metadata: { path: '', host: 'localhost' } }), 0)).toBe(false)
|
||||
})
|
||||
|
||||
it('inactiveSessionCanResume allows claude resume by message recovery when no claudeSessionId is stored', () => {
|
||||
expect(inactiveSessionCanResume(makeSession({
|
||||
metadata: { path: '/tmp/project', host: 'localhost', flavor: 'claude' },
|
||||
}), 3)).toBe(true)
|
||||
})
|
||||
|
||||
it('inactiveSessionCanResume allows claude recovery when flavor is missing (defaults to claude)', () => {
|
||||
expect(inactiveSessionCanResume(makeSession({
|
||||
metadata: { path: '/tmp/project', host: 'localhost' },
|
||||
}), 3)).toBe(true)
|
||||
})
|
||||
|
||||
it('inactiveSessionCanResume rejects non-claude flavors with messages but no flavor-specific id (no recovery path)', () => {
|
||||
expect(inactiveSessionCanResume(makeSession({
|
||||
metadata: { path: '/tmp/project', host: 'localhost', flavor: 'codex' },
|
||||
}), 3)).toBe(false)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,49 @@
|
||||
import { isKnownFlavor } from '@hapi/protocol'
|
||||
import type { Session } from '@/types/api'
|
||||
|
||||
/** Agent thread id used by hub `resolveAgentResumeId`, flavor-specific.
|
||||
* Mirrors hub: cross-flavor ids are ignored to avoid the web layer claiming a
|
||||
* session is resumable when the hub will only honor the current flavor's id. */
|
||||
export function resolveAgentSessionIdFromMetadata(
|
||||
metadata: Session['metadata'] | null | undefined,
|
||||
): string | undefined {
|
||||
if (!metadata) {
|
||||
return undefined
|
||||
}
|
||||
const flavor = isKnownFlavor(metadata.flavor) ? metadata.flavor : 'claude'
|
||||
switch (flavor) {
|
||||
case 'codex': return metadata.codexSessionId ?? undefined
|
||||
case 'gemini': return metadata.geminiSessionId ?? undefined
|
||||
case 'opencode': return metadata.opencodeSessionId ?? undefined
|
||||
case 'cursor': return metadata.cursorSessionId ?? undefined
|
||||
case 'kimi': return metadata.kimiSessionId ?? undefined
|
||||
default: return metadata.claudeSessionId ?? undefined
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether an inactive session can be activated via resume (or fresh spawn on first send).
|
||||
* Matches hub: resume with agent id, or fresh spawn when path exists, no agent id, no user messages.
|
||||
* Claude with messages but no `claudeSessionId` is allowed because hub
|
||||
* `recoverClaudeSessionIdFromMessages` reconstructs the resume id from the
|
||||
* stored message log (only the claude path has this recovery fallback).
|
||||
*/
|
||||
export function inactiveSessionCanResume(
|
||||
session: Session,
|
||||
userMessageCount: number,
|
||||
): boolean {
|
||||
if (session.active) {
|
||||
return true
|
||||
}
|
||||
if (!session.metadata?.path) {
|
||||
return false
|
||||
}
|
||||
if (resolveAgentSessionIdFromMetadata(session.metadata)) {
|
||||
return true
|
||||
}
|
||||
const flavor = isKnownFlavor(session.metadata.flavor) ? session.metadata.flavor : 'claude'
|
||||
if (flavor === 'claude' && userMessageCount > 0) {
|
||||
return true
|
||||
}
|
||||
return userMessageCount === 0
|
||||
}
|
||||
@@ -34,6 +34,7 @@ import { useToast } from '@/lib/toast-context'
|
||||
import { useTranslation } from '@/lib/use-translation'
|
||||
import { fetchLatestMessages, seedMessageWindowFromSession } from '@/lib/message-window-store'
|
||||
import { clearDraftsAfterSend } from '@/lib/clearDraftsAfterSend'
|
||||
import { inactiveSessionCanResume } from '@/lib/sessionResume'
|
||||
import { markSessionSeen } from '@/lib/sessionLastSeen'
|
||||
import type { Machine } from '@/types/api'
|
||||
import FilesPage from '@/routes/sessions/files'
|
||||
@@ -296,6 +297,9 @@ function SessionPage() {
|
||||
if (!api || !session || session.active) {
|
||||
return currentSessionId
|
||||
}
|
||||
if (!inactiveSessionCanResume(session, messages.length)) {
|
||||
throw new Error(t('resume.unavailable.noTarget'))
|
||||
}
|
||||
try {
|
||||
return await api.resumeSession(currentSessionId, { permissionMode: session.permissionMode ?? undefined })
|
||||
} catch (error) {
|
||||
|
||||
Reference in New Issue
Block a user