diff --git a/package-lock.json b/package-lock.json index 7a028cd..31ede4e 100644 --- a/package-lock.json +++ b/package-lock.json @@ -8,7 +8,7 @@ "name": "openframe-frontend", "version": "0.1.0", "dependencies": { - "@flamingo-stack/openframe-frontend-core": "^0.0.446", + "@flamingo-stack/openframe-frontend-core": "^0.0.448", "@hookform/resolvers": "^5.2.2", "@monaco-editor/react": "^4.7.0", "@tanstack/react-query": "^5.90.16", @@ -498,9 +498,9 @@ } }, "node_modules/@flamingo-stack/openframe-frontend-core": { - "version": "0.0.446", - "resolved": "https://registry.npmjs.org/@flamingo-stack/openframe-frontend-core/-/openframe-frontend-core-0.0.446.tgz", - "integrity": "sha512-WN81ghVxAywiF4dglDfHpOSqJ0fv0thkEVlDVKz86i9MN0iVCFv18otQen9PWwkfn2Xrm99xuKGWhwC7/ex3Gw==", + "version": "0.0.448", + "resolved": "https://registry.npmjs.org/@flamingo-stack/openframe-frontend-core/-/openframe-frontend-core-0.0.448.tgz", + "integrity": "sha512-QFCbnJIh9NB/HeTsOLwlDVxgYF2CuZRQlBCuHKsT3QUEFLsILrUERFazct0Gh1Seau9G97+AzW0oZe3F3W1Nkg==", "dependencies": { "@dnd-kit/core": "^6.3.1", "@dnd-kit/sortable": "^10.0.0", diff --git a/package.json b/package.json index 3bdbc9b..dedac64 100644 --- a/package.json +++ b/package.json @@ -24,7 +24,7 @@ "core:unlink": "yalc remove @flamingo-stack/openframe-frontend-core" }, "dependencies": { - "@flamingo-stack/openframe-frontend-core": "^0.0.446", + "@flamingo-stack/openframe-frontend-core": "^0.0.448", "@hookform/resolvers": "^5.2.2", "@monaco-editor/react": "^4.7.0", "@tanstack/react-query": "^5.90.16", diff --git a/schema.graphql b/schema.graphql index 987b30d..7164ada 100644 --- a/schema.graphql +++ b/schema.graphql @@ -1,6 +1,6 @@ # Auto-generated by scripts/fetch-schema.mjs via introspection -# Source: https://test-frontend.openframe.build/api/graphql -# Generated: 2026-07-15T10:27:33.577Z +# Source: https://test-dev.openframe.build/api/graphql +# Generated: 2026-07-17T22:22:39.836Z # # Do not edit manually. Re-run: npm run fetch-schema directive @extends on OBJECT | INTERFACE @@ -1034,11 +1034,6 @@ type Machine implements Node & AssignableTarget { } type Mutation { - """ - Update the current authenticated tenant's name and/or website. saas-api only. - """ - updateTenantInfo(input: UpdateTenantInfoInput!): TenantInfoPayload! - """Create a Stripe checkout session for subscribing to selected products.""" createCheckoutSession(input: CheckoutInput!): CheckoutResult! @@ -1068,6 +1063,11 @@ type Mutation { """Detach all payment methods. Returns the count of detached methods.""" detachPaymentMethods: Int! + """ + Update the current authenticated tenant's name and/or website. saas-api only. + """ + updateTenantInfo(input: UpdateTenantInfoInput!): TenantInfoPayload! + """Mark one Initial Setup step done. Idempotent.""" completeTenantOnboardingStep(step: TenantOnboardingStep!): TenantOnboardingProgress! @@ -1089,10 +1089,45 @@ type Mutation { Reset personal onboarding to a clean state (no steps, not skipped, not completed). """ resetUserOnboarding: UserOnboardingProgress! - markNotificationAsRead(notificationId: ID!): Boolean! - markAllNotificationsAsRead: Int! - deleteNotification(notificationId: ID!): Boolean! - deleteAllReadNotifications: Int! + + """ + Create a new script. Returns the created script with its server-assigned id. + """ + createScript(input: CreateScriptInput!): Script! + + """ + Full replacement (PUT semantics) of an existing script: every writable field on the input overwrites the stored value, including nulls which clear optional fields. Required fields (name, shell, scriptBody) must be present. The target script id travels inside the input. + """ + updateScript(input: UpdateScriptInput!): Script! + + """ + Soft-delete a script (sets status to DELETED). Returns the id of the deleted script; idempotent on already-deleted scripts. Throws if the script id is not found in the tenant. + """ + deleteScript(id: ID!): ID! + + """ + Archive a script (sets status to ARCHIVED). Idempotent on already-archived scripts. Returns the updated script. Throws if the script id is not found or is soft-deleted in the tenant. + """ + archiveScript(id: ID!): Script! + + """ + Restore an archived script back to ACTIVE. Idempotent on scripts that are not archived. Returns the updated script. Throws if the script id is not found or is soft-deleted in the tenant. + """ + unarchiveScript(id: ID!): Script! + + """ + Dispatch a saved script to a target machine over NATS. Nothing is persisted; + the script body / shell / env-vars are resolved server-side from the stored script. + Returns an executionId for correlating the agent's asynchronous result. + """ + runScript(input: RunScriptInput!): DispatchResponse! + + """ + Dispatch one saved script to several machines at once over NATS, under a single + shared executionId. Nothing is persisted; the script body / shell / env-vars are + resolved server-side. Returns the shared executionId. + """ + batchRunScript(input: BatchRunScriptInput!): DispatchResponse! """ Create a new schedule. Returns the created schedule with its server-assigned id. @@ -1124,6 +1159,9 @@ type Mutation { machineIds are Machine global ids. Returns the updated schedule. Throws if the schedule is not found or soft-deleted. """ setScriptScheduleDevices(scheduleId: ID!, machineIds: [ID!]!): ScriptSchedule! + runCommand(input: RunCommandInput!): DispatchResponse! + batchRunCommand(input: BatchRunCommandInput!): DispatchResponse! + cancelExecution(input: CancelExecutionInput!): DispatchResponse! createFolder(name: String!, parentId: ID): KnowledgeBaseItem! renameFolder(id: ID!, name: String!): KnowledgeBaseItem! createArticle(input: CreateArticleInput!): KnowledgeBaseItem! @@ -1141,50 +1179,11 @@ type Mutation { linkKnowledgeBaseTempAttachmentsToArticle(input: LinkKnowledgeBaseTempAttachmentsInput!): [KnowledgeBaseItemAttachment!]! createKnowledgeBaseAttachmentUploadUrl(input: CreateKnowledgeBaseAttachmentInput!): KnowledgeBaseAttachmentUploadPayload! deleteKnowledgeBaseAttachment(input: MutationDeleteInput!): MutationDeletePayload! - - """ - Create a new script. Returns the created script with its server-assigned id. - """ - createScript(input: CreateScriptInput!): Script! - - """ - Full replacement (PUT semantics) of an existing script: every writable field on the input overwrites the stored value, including nulls which clear optional fields. Required fields (name, shell, scriptBody) must be present. The target script id travels inside the input. - """ - updateScript(input: UpdateScriptInput!): Script! - - """ - Soft-delete a script (sets status to DELETED). Returns the id of the deleted script; idempotent on already-deleted scripts. Throws if the script id is not found in the tenant. - """ - deleteScript(id: ID!): ID! - - """ - Archive a script (sets status to ARCHIVED). Idempotent on already-archived scripts. Returns the updated script. Throws if the script id is not found or is soft-deleted in the tenant. - """ - archiveScript(id: ID!): Script! - - """ - Restore an archived script back to ACTIVE. Idempotent on scripts that are not archived. Returns the updated script. Throws if the script id is not found or is soft-deleted in the tenant. - """ - unarchiveScript(id: ID!): Script! - - """ - Dispatch a saved script to a target machine over NATS. Nothing is persisted; - the script body / shell / env-vars are resolved server-side from the stored script. - Returns an executionId for correlating the agent's asynchronous result. - """ - runScript(input: RunScriptInput!): DispatchResponse! - - """ - Dispatch one saved script to several machines at once over NATS, under a single - shared executionId. Nothing is persisted; the script body / shell / env-vars are - resolved server-side. Returns the shared executionId. - """ - batchRunScript(input: BatchRunScriptInput!): DispatchResponse! createEvent(input: CreateEventInput!): Event updateEvent(id: ID!, input: CreateEventInput!): Event - createTag(key: String!, entityType: String!, description: String, color: String): Tag! - updateTag(id: ID!, key: String, description: String, color: String): Tag! - deleteTag(id: ID!): Boolean! + assignItem(itemId: ID!, itemType: AssignmentItemType!, targetType: AssignmentTargetType!, targetId: ID!): ItemAssignment! + unassignItem(itemId: ID!, targetType: AssignmentTargetType!, targetId: ID!): Boolean! + unassignAllByType(itemId: ID!, targetType: AssignmentTargetType!): Boolean! startTimer(input: StartTimerInput): TimeEntry! pauseTimer: TimeEntry! resumeTimer: TimeEntry! @@ -1194,12 +1193,23 @@ type Mutation { updateTimeEntry(input: UpdateTimeEntryInput!): TimeEntry! unlinkTicketFromTimeEntry(id: ID!): TimeEntry! deleteTimeEntry(id: ID!): Boolean! - runCommand(input: RunCommandInput!): DispatchResponse! - batchRunCommand(input: BatchRunCommandInput!): DispatchResponse! - cancelExecution(input: CancelExecutionInput!): DispatchResponse! - assignItem(itemId: ID!, itemType: AssignmentItemType!, targetType: AssignmentTargetType!, targetId: ID!): ItemAssignment! - unassignItem(itemId: ID!, targetType: AssignmentTargetType!, targetId: ID!): Boolean! - unassignAllByType(itemId: ID!, targetType: AssignmentTargetType!): Boolean! + createTag(key: String!, entityType: String!, description: String, color: String): Tag! + updateTag(id: ID!, key: String, description: String, color: String): Tag! + deleteTag(id: ID!): Boolean! + markNotificationAsRead(notificationId: ID!): Boolean! + markAllNotificationsAsRead: Int! + deleteNotification(notificationId: ID!): Boolean! + deleteAllReadNotifications: Int! + + """ + Advance this tenant's Stripe test clock by N days (dev/stage only). Attaches a clock to the customer on first call, flushes current metered usage to Stripe, then advances. + """ + advanceTestClock(days: Int!): TestClockResult + + """ + Delete this tenant's Stripe test clock (dev/stage only). Cascades to delete the customer, resetting the tenant to trial. + """ + resetTestClock: Boolean! } input MutationDeleteInput { @@ -1505,54 +1515,42 @@ type Query { """Current user's Get Started progress.""" userOnboardingProgress: UserOnboardingProgress! - deviceFilters(filter: DeviceFilterInput): DeviceFilters! - devices(filter: DeviceFilterInput, first: Int, after: String, last: Int, before: String, search: String, sort: SortInput): DeviceConnection! - - """ Get single device by Relay global ID""" - deviceById(id: ID!): Machine - - """ Get single device by machineId""" - device(machineId: String!): Machine - notifications(filter: NotificationFilterInput, search: String, first: Int, after: String, last: Int, before: String, sort: SortInput): NotificationConnection! - hasUnreadNotifications: Boolean! - unreadCountsByCategory: [UnreadCategoryCount!]! """ - Get a single schedule by id within the current tenant. Throws (not null) if the id is absent, soft-deleted, or in another tenant. + Get a single script by id within the current tenant. Throws (not null) if the id is absent, soft-deleted, or in another tenant. """ - scriptSchedule(id: ID!): ScriptSchedule! + script(id: ID!): Script! """ - Cursor-paginated list of schedules within the current tenant (Relay Connection Spec). + Cursor-paginated list of scripts within the current tenant (Relay Connection Spec). Default order is newest-first by _id. Optional filter / search / sort. Sortable fields: _id (default), name, createdAt, updatedAt. Search is a case-insensitive substring match on name. """ - scriptSchedules(filter: ScriptScheduleFilterInput, search: String, sort: SortInput, first: Int, after: String, last: Int, before: String): ScriptScheduleConnection! - scriptScheduleFilters(filter: ScriptScheduleFilterInput): ScriptScheduleFilters! - knowledgeBaseItems(filter: KnowledgeBaseFilterInput, search: String, first: Int, after: String): KnowledgeBaseItemConnection! - knowledgeBaseItem(id: ID!): KnowledgeBaseItem - knowledgeBaseTags(folderId: ID, archived: Boolean): [Tag!]! - knowledgeBaseFolderTree: [KnowledgeBaseItem!]! - knowledgeBaseArticleTree: [KnowledgeBaseItem!]! - archivedArticles(search: String, tagIds: [ID], first: Int, after: String): KnowledgeBaseItemConnection! - knowledgeBaseAttachmentDownloadUrl(attachmentId: ID!): String! + scripts(filter: ScriptFilterInput, search: String, sort: SortInput, first: Int, after: String, last: Int, before: String): ScriptConnection! + scriptFilters(filter: ScriptFilterInput): ScriptFilters! scriptExecutions(scriptId: ID!, filter: ScriptExecutionFilterInput, search: String, sort: SortInput, first: Int, after: String, last: Int, before: String): ScriptExecutionConnection! scriptExecutionFilters(scriptId: ID!, filter: ScriptExecutionFilterInput, search: String): ScriptExecutionFilters! """ - Get a single script by id within the current tenant. Throws (not null) if the id is absent, soft-deleted, or in another tenant. + Get a single schedule by id within the current tenant. Throws (not null) if the id is absent, soft-deleted, or in another tenant. """ - script(id: ID!): Script! + scriptSchedule(id: ID!): ScriptSchedule! """ - Cursor-paginated list of scripts within the current tenant (Relay Connection Spec). + Cursor-paginated list of schedules within the current tenant (Relay Connection Spec). Default order is newest-first by _id. Optional filter / search / sort. Sortable fields: _id (default), name, createdAt, updatedAt. Search is a case-insensitive substring match on name. """ - scripts(filter: ScriptFilterInput, search: String, sort: SortInput, first: Int, after: String, last: Int, before: String): ScriptConnection! - scriptFilters(filter: ScriptFilterInput): ScriptFilters! + scriptSchedules(filter: ScriptScheduleFilterInput, search: String, sort: SortInput, first: Int, after: String, last: Int, before: String): ScriptScheduleConnection! + scriptScheduleFilters(filter: ScriptScheduleFilterInput): ScriptScheduleFilters! + + " Get all integrated tools with optional filtering\n Returns a simple list since tools are limited in number (configuration data)" + integratedTools(filter: ToolFilterInput, search: String, sort: SortInput): ToolList! + + """ Get available tool filters""" + toolFilters: ToolFilters! """ Get all organizations with optional filtering and pagination""" organizations(filter: OrganizationFilterInput, first: Int, after: String, last: Int, before: String, search: String, orderBy: OrganizationSortInput): OrganizationConnection! @@ -1562,9 +1560,33 @@ type Query { """ Get organization by organizationId""" organizationByOrganizationId(organizationId: String!): Organization + knowledgeBaseItems(filter: KnowledgeBaseFilterInput, search: String, first: Int, after: String): KnowledgeBaseItemConnection! + knowledgeBaseItem(id: ID!): KnowledgeBaseItem + knowledgeBaseTags(folderId: ID, archived: Boolean): [Tag!]! + knowledgeBaseFolderTree: [KnowledgeBaseItem!]! + knowledgeBaseArticleTree: [KnowledgeBaseItem!]! + archivedArticles(search: String, tagIds: [ID], first: Int, after: String): KnowledgeBaseItemConnection! + knowledgeBaseAttachmentDownloadUrl(attachmentId: ID!): String! events(filter: EventFilterInput, first: Int, after: String, last: Int, before: String, search: String, sort: SortInput): EventConnection! eventById(id: ID!): Event eventFilters(filter: EventFilterInput): EventFilters! + deviceFilters(filter: DeviceFilterInput): DeviceFilters! + devices(filter: DeviceFilterInput, first: Int, after: String, last: Int, before: String, search: String, sort: SortInput): DeviceConnection! + + """ Get single device by Relay global ID""" + deviceById(id: ID!): Machine + + """ Get single device by machineId""" + device(machineId: String!): Machine + assignedItemCounts(itemId: ID!): [AssignedItemCount!]! + assignedItems(itemId: ID!, targetType: AssignmentTargetType!, search: String, sort: SortInput, first: Int, after: String): ItemAssignmentConnection! + currentTimer: TimeEntry + myTimeEntries(period: DateRangeInput, search: String, sort: SortInput, first: Int, after: String): TimeEntryConnection! + + " Tenant-wide time entries. Filter groups all criteria: employeeIds (all users if null/empty),\n organizationIds (customer filter), startedFrom/startedTo (period)." + employeeTimeEntries(filter: TimeEntryFilterInput, search: String, sort: SortInput, first: Int, after: String): TimeEntryConnection! + employeeTimeStats(filter: TimeEntryFilterInput): EmployeeTimeStats! + timeEntry(id: ID!): TimeEntry """ Tag key queries (tenant-wide, scoped to DEVICE entity type)""" tags: [Tag!]! @@ -1583,19 +1605,6 @@ type Query { " Tag keys assigned to scripts. When archived=true returns tags assigned to\n archived scripts; when false/null returns tags assigned to active scripts." scriptsTags(archived: Boolean): [Tag!]! - currentTimer: TimeEntry - myTimeEntries(period: DateRangeInput, search: String, sort: SortInput, first: Int, after: String): TimeEntryConnection! - - " Tenant-wide time entries. Filter groups all criteria: employeeIds (all users if null/empty),\n organizationIds (customer filter), startedFrom/startedTo (period)." - employeeTimeEntries(filter: TimeEntryFilterInput, search: String, sort: SortInput, first: Int, after: String): TimeEntryConnection! - employeeTimeStats(filter: TimeEntryFilterInput): EmployeeTimeStats! - timeEntry(id: ID!): TimeEntry - - " Get all integrated tools with optional filtering\n Returns a simple list since tools are limited in number (configuration data)" - integratedTools(filter: ToolFilterInput, search: String, sort: SortInput): ToolList! - - """ Get available tool filters""" - toolFilters: ToolFilters! logFilters(filter: LogFilterInput): LogFilters! logs(filter: LogFilterInput, first: Int, after: String, last: Int, before: String, search: String, sort: LogSortInput): LogConnection! @@ -1615,8 +1624,9 @@ type Query { """ Timestamp of the audit""" toolEventId: String! ): LogDetails - assignedItemCounts(itemId: ID!): [AssignedItemCount!]! - assignedItems(itemId: ID!, targetType: AssignmentTargetType!, search: String, sort: SortInput, first: Int, after: String): ItemAssignmentConnection! + notifications(filter: NotificationFilterInput, search: String, first: Int, after: String, last: Int, before: String, sort: SortInput): NotificationConnection! + hasUnreadNotifications: Boolean! + unreadCountsByCategory: [UnreadCategoryCount!]! """Current authenticated tenant's info (name, website, image).""" tenantInfo: TenantInfo @@ -2125,6 +2135,10 @@ enum TenantOnboardingStep { COMPANY_TEAM } +type TestClockResult { + frozenTime: String! +} + " Ticket fields exposed for AssignableTarget resolution. Full type in ai-agent (ticket.graphqls).\n All fields below come from the Ticket @Document entity (denormalized at write time).\n Fields requiring DataLoaders (assignee User, attachments, notes, dialog, labels) are\n only on the ai-agent side; query ticket(id) there for full detail.\n TODO: Remove after Apollo Federation migration — single shared Ticket type." type Ticket implements AssignableTarget & Node { id: ID! diff --git a/src/app/(app)/mingo/hooks/use-mingo-dialog-selection.ts b/src/app/(app)/mingo/hooks/use-mingo-dialog-selection.ts index b08ac74..4b1a39a 100644 --- a/src/app/(app)/mingo/hooks/use-mingo-dialog-selection.ts +++ b/src/app/(app)/mingo/hooks/use-mingo-dialog-selection.ts @@ -344,6 +344,9 @@ export function useMingoDialogSelection() { createdAt: msg.createdAt, owner: msg.owner, messageData: msg.messageData, + // Feed the merge's per-role seq coverage so a post-reload live synthetic + // isn't dropped by an unrelated row's seq (message-loss on stop/reload). + lastChunkStreamSeq: msg.lastChunkStreamSeq, })); const assistantConfig = ASSISTANT_CONFIG.MINGO; @@ -355,6 +358,11 @@ export function useMingoDialogSelection() { onReject: handleRejectRef.current, approvalStatuses: Object.fromEntries(Object.entries(approvalStatusesRef.current).map(([k, v]) => [k, v as any])), batchApprovalsEnabled: featureFlags.batchApproval.enabled(), + // Must match the realtime processor (use-mingo-realtime-subscription): + // an omitted option is not guaranteed to include ADMIN, and a history + // reprocess (reopen, reconnect invalidation, pagination) would silently + // drop pending ADMIN approval cards that rendered live. + displayApprovalTypes: ['CLIENT', 'ADMIN'], }); const allProcessedMessages = foldPendingApprovalsEnvelope(rawProcessedMessages as Message[]); diff --git a/src/app/(app)/mingo/hooks/use-mingo-realtime-subscription.ts b/src/app/(app)/mingo/hooks/use-mingo-realtime-subscription.ts index d92249e..c4c9d45 100644 --- a/src/app/(app)/mingo/hooks/use-mingo-realtime-subscription.ts +++ b/src/app/(app)/mingo/hooks/use-mingo-realtime-subscription.ts @@ -354,8 +354,16 @@ function useDialogChunkProcessor(dialogId: string, options: UseDialogChunkProces }, onToolExecuted: (segment: ToolExecutionSegment) => { + // Merge into an existing card first; when NOTHING matches — the + // common case for the FIRST post-MESSAGE_END EXECUTING chunk of an + // approved command, which has no prior segment to merge into — + // append the segment to the last assistant bubble instead of + // dropping it, or the whole tool run stays invisible. const execId = segment.data.toolExecutionRequestId; - if (execId) updateToolExecutionInMessages(dialogId, execId, segment.data); + const matched = updateToolExecutionInMessages(dialogId, execId, segment.data); + if (!matched) { + appendSegmentsToLastAssistant(dialogId, [segment]); + } }, onUserMessage: ( @@ -476,6 +484,16 @@ export function DialogSubscription({ // Rejects out-of-order JetStream redeliveries. const lastAppliedStreamSeqRef = useRef(-1); + // Dispatch-level redelivery gate (mirrors the tickets subscription): + // JetStream is at-least-once, so a chunk with an equal-or-older streamSeq + // has already been processed — letting it through duplicates accumulator + // text / participant rows. The streamState guard above only protects the + // query-cache write, not the processor. + const lastDispatchedStreamSeqRef = useRef(-1); + useEffect(() => { + lastDispatchedStreamSeqRef.current = -1; + }, [dialogId]); + const syncStreamStateFromChunk = useCallback( (chunk: ChunkData) => { if (typeof chunk.streamSeq === 'number') { @@ -500,6 +518,10 @@ export function DialogSubscription({ if (featureFlags.debugNatsChunks.enabled()) { console.log('[mingo-js] chunk received', { dialogId, streamSeq: chunk.streamSeq, chunk }); } + if (typeof chunk.streamSeq === 'number') { + if (chunk.streamSeq <= lastDispatchedStreamSeqRef.current) return; + lastDispatchedStreamSeqRef.current = chunk.streamSeq; + } syncStreamStateFromChunk(chunk); processorRef.current(chunk); }, @@ -514,7 +536,7 @@ export function DialogSubscription({ onConnectionChange?.(dialogId, false); }, [dialogId, onConnectionChange]); - useJetStreamDialogSubscription({ + const { reconnectionCount } = useJetStreamDialogSubscription({ enabled: isInitialOptStartSeqReady, dialogId, streamName: CHAT_CHUNKS_STREAM, @@ -527,5 +549,16 @@ export function DialogSubscription({ getNatsWsUrl: getWsUrl, }); + // NATS reconnect: JetStream replays only ~10 minutes of CHAT_CHUNKS, so an + // outage longer than that leaves a gap the resume-by-seq cannot fill. + // Refetch persisted history — the merge layer dedupes what replay covers. + const lastHandledReconnectRef = useRef(0); + useEffect(() => { + if (reconnectionCount <= lastHandledReconnectRef.current) return; + lastHandledReconnectRef.current = reconnectionCount; + void queryClient.invalidateQueries({ queryKey: ['mingo-dialog-messages', dialogId] }); + void queryClient.invalidateQueries({ queryKey: ['mingo-dialog', dialogId] }); + }, [reconnectionCount, queryClient, dialogId]); + return null; } diff --git a/src/app/(app)/mingo/stores/mingo-messages-store.ts b/src/app/(app)/mingo/stores/mingo-messages-store.ts index 1b99e95..2167a02 100644 --- a/src/app/(app)/mingo/stores/mingo-messages-store.ts +++ b/src/app/(app)/mingo/stores/mingo-messages-store.ts @@ -65,11 +65,16 @@ interface MingoMessagesStore { status: 'approved' | 'rejected', resolvedByName?: string | null, ) => void; + /** Merge an EXECUTING/EXECUTED event into the matching tool segment or + * batch `executions` slot. `executionRequestId` is optional — without it, + * an EXECUTED chunk pairs with the newest EXECUTING of the same + * (integratedToolType, toolFunction). Returns whether anything matched so + * the caller can append a fresh segment instead of dropping the event. */ updateToolExecutionInMessages: ( dialogId: string, - executionRequestId: string, + executionRequestId: string | undefined, executedData: ToolExecutionSegment['data'], - ) => void; + ) => boolean; getMessages: (dialogId: string) => Message[]; // Real-time State Management @@ -289,67 +294,99 @@ export const useMingoMessagesStore = create()( updateToolExecutionInMessages: ( dialogId: string, - executionRequestId: string, + executionRequestId: string | undefined, executedData: ToolExecutionSegment['data'], ) => { + let matched = false; set(state => { const newMap = new Map(state.messagesByDialog); const currentMessages = newMap.get(dialogId) || []; - let matched = false; - const updatedMessages = currentMessages.map(message => { - if (matched) return message; - if (message.role !== 'assistant' || !Array.isArray(message.content)) return message; - - let messageChanged = false; - const updatedContent = message.content.map(segment => { - if (matched) return segment; - - if ( - segment.type === 'tool_execution' && - segment.data.type === 'EXECUTING_TOOL' && - segment.data.toolExecutionRequestId === executionRequestId - ) { - matched = true; - messageChanged = true; - const merged: ToolExecutionSegment = { - type: 'tool_execution', - data: { - ...executedData, - toolTitle: executedData.toolTitle ?? segment.data.toolTitle, - parameters: executedData.parameters ?? segment.data.parameters, - }, - }; - return merged; - } - - if ( + // Scan NEWEST-first: an id-less EXECUTED must pair with the newest + // in-flight EXECUTING of the same tool (a stale interrupted card + // from an earlier turn must not swallow the current run); execId + // matches are unique so direction doesn't change them. + let msgIdx = -1; + let segIdx = -1; + for (let i = currentMessages.length - 1; i >= 0 && msgIdx === -1; i--) { + const message = currentMessages[i]; + if (message.role !== 'assistant' || !Array.isArray(message.content)) continue; + const content = message.content as MessageSegment[]; + for (let j = content.length - 1; j >= 0; j--) { + const segment = content[j]; + if (segment.type === 'tool_execution') { + const idMatches = executionRequestId + ? segment.data.toolExecutionRequestId === executionRequestId + : segment.data.type === 'EXECUTING_TOOL' && + segment.data.integratedToolType === executedData.integratedToolType && + segment.data.toolFunction === executedData.toolFunction; + if (idMatches) { + msgIdx = i; + segIdx = j; + break; + } + } else if ( + executionRequestId && segment.type === 'approval_batch' && segment.data.toolCalls.some(c => c.toolExecutionRequestId === executionRequestId) ) { - matched = true; - messageChanged = true; - const prev: ApprovalBatchExecutionState | undefined = segment.data.executions?.[executionRequestId]; - const next: ApprovalBatchExecutionState = - executedData.type === 'EXECUTED_TOOL' - ? { status: 'done', result: executedData.result, success: executedData.success } - : { status: 'executing', result: prev?.result, success: prev?.success }; - return { - ...segment, - data: { - ...segment.data, - executions: { ...(segment.data.executions ?? {}), [executionRequestId]: next }, - }, - } as ApprovalBatchSegment; + msgIdx = i; + segIdx = j; + break; } + } + } - return segment; - }); - - return messageChanged ? { ...message, content: updatedContent } : message; - }); + if (msgIdx === -1) return state; + matched = true; + + const message = currentMessages[msgIdx]; + const content = message.content as MessageSegment[]; + const segment = content[segIdx]; + let nextSegment: MessageSegment; + + if (segment.type === 'tool_execution') { + // Never downgrade a completed run back to EXECUTING (JetStream + // redelivery of the EXECUTING chunk after EXECUTED landed). + // `matched` stays TRUE here on purpose: the run is already + // represented on screen, so the caller's append fallback must not + // fire — a `false` would duplicate the card. + if (executedData.type === 'EXECUTING_TOOL' && segment.data.type === 'EXECUTED_TOOL') { + return state; + } + nextSegment = { + type: 'tool_execution', + data: { + ...executedData, + toolTitle: executedData.toolTitle ?? segment.data.toolTitle, + parameters: executedData.parameters ?? segment.data.parameters, + }, + } satisfies ToolExecutionSegment; + } else { + const batch = segment as ApprovalBatchSegment; + const prev: ApprovalBatchExecutionState | undefined = batch.data.executions?.[executionRequestId as string]; + // Same no-downgrade rule for a batch slot: a redelivered EXECUTING + // must not flip a 'done' entry back (matched stays true → no append). + if (executedData.type === 'EXECUTING_TOOL' && prev?.status === 'done') { + return state; + } + const next: ApprovalBatchExecutionState = + executedData.type === 'EXECUTED_TOOL' + ? { status: 'done', result: executedData.result, success: executedData.success } + : { status: 'executing', result: prev?.result, success: prev?.success }; + nextSegment = { + ...batch, + data: { + ...batch.data, + executions: { ...(batch.data.executions ?? {}), [executionRequestId as string]: next }, + }, + } as ApprovalBatchSegment; + } - if (!matched) return state; + const nextContent = [...content]; + nextContent[segIdx] = nextSegment; + const updatedMessages = [...currentMessages]; + updatedMessages[msgIdx] = { ...message, content: nextContent }; newMap.set(dialogId, updatedMessages); @@ -362,6 +399,7 @@ export const useMingoMessagesStore = create()( return { messagesByDialog: newMap, streamingMessages: newStreamingMap }; }); + return matched; }, getMessages: (dialogId: string) => { @@ -480,16 +518,25 @@ export const useMingoMessagesStore = create()( let nextContent: MessageSegment[]; if (incomingCompaction) { - const startedIdx = existing.findIndex(s => s.type === 'context_compaction' && s.status === 'started'); - const hasAnyCompaction = existing.some(s => s.type === 'context_compaction'); + // Mirror the lib's upsertTrailingCompaction: replace the LAST + // 'started' (earlier compactions in the bubble are already + // completed-in-place), else append. The previous first-match + + // any-compaction blanket silently dropped a second compaction + // landing in the same bubble. + let startedIdx = -1; + for (let k = existing.length - 1; k >= 0; k--) { + const seg = existing[k]; + if (seg.type === 'context_compaction' && seg.status === 'started') { + startedIdx = k; + break; + } + } - if (incomingCompaction.status === 'completed' && startedIdx !== -1) { + if (startedIdx !== -1) { nextContent = [...existing]; nextContent[startedIdx] = incomingCompaction; - } else if (!hasAnyCompaction) { - nextContent = [...existing, incomingCompaction]; } else { - nextContent = existing; + nextContent = [...existing, incomingCompaction]; } } else { const accumulator = state.segmentAccumulators.get(dialogId); @@ -511,7 +558,20 @@ export const useMingoMessagesStore = create()( } } - return state; + // No assistant message anywhere (thread cleared / history not yet + // hydrated): open a fresh bubble instead of dropping the segments — + // mirrors the lib's appendToTrailingAssistant. Silently dropping the + // first post-MESSAGE_END tool chunk here is exactly the + // invisible-tool-run bug this pipeline was fixed for. + const fallback: Message = { + id: `assistant-${Date.now()}-${Math.random().toString(16).slice(2)}`, + role: 'assistant', + content: incomingCompaction ? [incomingCompaction] : [...segments], + timestamp: new Date(), + ...(streamSeq != null ? { streamSeq } : {}), + }; + newMap.set(dialogId, [...currentMessages, fallback]); + return { messagesByDialog: newMap }; }); }, diff --git a/src/app/(app)/tickets/components/ticket-details-view.tsx b/src/app/(app)/tickets/components/ticket-details-view.tsx index 8b84438..fa367b5 100644 --- a/src/app/(app)/tickets/components/ticket-details-view.tsx +++ b/src/app/(app)/tickets/components/ticket-details-view.tsx @@ -425,6 +425,16 @@ export function TicketDetailsView({ ticketId }: TicketDetailsViewProps) { const adminInitialOptStartSeq = useMemo(() => maxPersistedStreamSeq(adminChat.rawPages), [adminChat.rawPages]); const isInitialOptStartSeqReady = clientChat.isFetched && (!isTechnicianChatEnabled || adminChat.isFetched); + // NATS reconnect: JetStream replays only ~10 minutes of CHAT_CHUNKS, so an + // outage longer than that leaves a gap the resume-by-seq cannot fill. + // Refetch persisted history — the merge layer dedupes what replay covers. + const refetchClientChat = clientChat.refetch; + const refetchAdminChat = adminChat.refetch; + const handleNatsReconnected = useCallback(() => { + void refetchClientChat(); + if (isTechnicianChatEnabled) void refetchAdminChat(); + }, [refetchClientChat, refetchAdminChat, isTechnicianChatEnabled]); + const applyStatus = useCallback( (nextStatus: Dialog['status']) => { queryClient.setQueryData(ticketsQueryKeys.detail(ticketId), prev => @@ -932,6 +942,7 @@ export function TicketDetailsView({ ticketId }: TicketDetailsViewProps) { adminInitialOptStartSeq={adminInitialOptStartSeq} isInitialOptStartSeqReady={isInitialOptStartSeqReady} subscribeAdmin={isTechnicianChatEnabled} + onReconnected={handleNatsReconnected} /> {isSidebarLayout ? ( void; } export function TicketDialogSubscription({ @@ -32,6 +36,7 @@ export function TicketDialogSubscription({ adminInitialOptStartSeq, isInitialOptStartSeqReady, subscribeAdmin, + onReconnected, }: TicketDialogSubscriptionProps) { const { getWsUrl, onBeforeReconnect } = useNatsAppConfig(); @@ -69,7 +74,7 @@ export function TicketDialogSubscription({ dispatchRef.current(chunk, NATS_TOPICS.ADMIN_MESSAGE); }, []); - useJetStreamDialogSubscription({ + const { reconnectionCount: clientReconnectionCount } = useJetStreamDialogSubscription({ enabled: !!dialogId && isInitialOptStartSeqReady, dialogId, streamName: CHAT_CHUNKS_STREAM, @@ -80,7 +85,7 @@ export function TicketDialogSubscription({ getNatsWsUrl: getWsUrl, }); - useJetStreamDialogSubscription({ + const { reconnectionCount: adminReconnectionCount } = useJetStreamDialogSubscription({ enabled: !!dialogId && isInitialOptStartSeqReady && subscribeAdmin, dialogId, streamName: CHAT_CHUNKS_STREAM, @@ -91,5 +96,20 @@ export function TicketDialogSubscription({ getNatsWsUrl: getWsUrl, }); + const onReconnectedRef = useRef(onReconnected); + useEffect(() => { + onReconnectedRef.current = onReconnected; + }, [onReconnected]); + + // Both subscriptions share one NATS connection, so a reconnect bumps both + // counters together — collapse to a single parent notification per event. + const lastNotifiedReconnectRef = useRef(0); + const reconnectTotal = Math.max(clientReconnectionCount, adminReconnectionCount); + useEffect(() => { + if (reconnectTotal <= lastNotifiedReconnectRef.current) return; + lastNotifiedReconnectRef.current = reconnectTotal; + onReconnectedRef.current?.(); + }, [reconnectTotal]); + return null; } diff --git a/src/app/(app)/tickets/hooks/use-historical-messages.ts b/src/app/(app)/tickets/hooks/use-historical-messages.ts index 4cca9a3..0646b9f 100644 --- a/src/app/(app)/tickets/hooks/use-historical-messages.ts +++ b/src/app/(app)/tickets/hooks/use-historical-messages.ts @@ -87,6 +87,9 @@ export function useHistoricalMessages({ createdAt: msg.createdAt, owner: msg.owner, messageData: msg.messageData, + // Feed the merge's per-role seq coverage so a post-reload live synthetic + // isn't dropped by an unrelated row's seq (message-loss on stop/reload). + lastChunkStreamSeq: msg.lastChunkStreamSeq, })); const historicalResolutions: Record = {}; @@ -111,6 +114,11 @@ export function useHistoricalMessages({ onReject: onRejectRef.current, approvalStatuses: { ...approvalStatusesRef.current, ...historicalResolutions }, batchApprovalsEnabled: featureFlags.batchApproval.enabled(), + // Must match the realtime processor (use-side-chunk-processor): an + // omitted option is not guaranteed to include ADMIN, and a history + // reprocess (reopen, reconnect refetch, pagination) would silently + // drop pending ADMIN approval cards that rendered live. + displayApprovalTypes: ['CLIENT', 'ADMIN'], }); const storeMessages: ChatMessage[] = foldPendingApprovalsEnvelope( @@ -123,6 +131,8 @@ export function useHistoricalMessages({ authorType: msg.authorType, timestamp: msg.timestamp, avatar: msg.avatar, + // Carry the persisted per-row seq into the merge (per-role coverage). + streamSeq: msg.streamSeq, })), ); diff --git a/src/app/(app)/tickets/hooks/use-side-chunk-processor.ts b/src/app/(app)/tickets/hooks/use-side-chunk-processor.ts index 21c3042..d919b7f 100644 --- a/src/app/(app)/tickets/hooks/use-side-chunk-processor.ts +++ b/src/app/(app)/tickets/hooks/use-side-chunk-processor.ts @@ -186,10 +186,26 @@ export function useSideChunkProcessor( } }, onToolExecuted: (segment: ToolExecutionSegment) => { + // With an execId, merge into an existing card on either side (a batch + // card can live in the other side's thread) — the id is unique, so a + // cross-side lookup can't touch the wrong segment. WITHOUT an id the + // fallback pairs by (toolType, toolFunction), which on the wrong side + // could flip an unrelated same-tool run — so fuzzy-match only THIS + // side. When NOTHING matches (the common case for the FIRST + // post-MESSAGE_END EXECUTING chunk of an approved command), append + // the segment to this side's last assistant bubble instead of + // dropping it, or the whole tool run stays invisible. const execId = segment.data.toolExecutionRequestId; + let matched: boolean; if (execId) { - updateToolExecutionInMessages('client', execId, segment.data); - updateToolExecutionInMessages('admin', execId, segment.data); + const matchedClient = updateToolExecutionInMessages('client', execId, segment.data); + const matchedAdmin = updateToolExecutionInMessages('admin', execId, segment.data); + matched = matchedClient || matchedAdmin; + } else { + matched = updateToolExecutionInMessages(side, undefined, segment.data); + } + if (!matched) { + appendSegmentsToLastAssistant(side, [segment]); } }, onUserMessage: ( diff --git a/src/app/(app)/tickets/hooks/use-ticket-messages.ts b/src/app/(app)/tickets/hooks/use-ticket-messages.ts index cc98965..923f733 100644 --- a/src/app/(app)/tickets/hooks/use-ticket-messages.ts +++ b/src/app/(app)/tickets/hooks/use-ticket-messages.ts @@ -44,6 +44,10 @@ export function useTicketMessages(dialogId: string | null, chatType: ChatType) { hasNextPage: messagesQuery.hasNextPage ?? false, isFetchingNextPage: messagesQuery.isFetchingNextPage, fetchNextPage: messagesQuery.fetchNextPage, + /** Re-fetch the loaded pages. Used after a NATS reconnect: the JetStream + * CHAT_CHUNKS stream retains only ~10 minutes, so a longer outage leaves + * a gap only the persisted history can fill (the merge dedupes). */ + refetch: messagesQuery.refetch, error: messagesQuery.error?.message || null, }; } diff --git a/src/app/(app)/tickets/stores/ticket-details-store.ts b/src/app/(app)/tickets/stores/ticket-details-store.ts index 2d38624..e4a0b40 100644 --- a/src/app/(app)/tickets/stores/ticket-details-store.ts +++ b/src/app/(app)/tickets/stores/ticket-details-store.ts @@ -75,11 +75,16 @@ interface TicketDetailsStore { status: ApprovalStatus, resolvedByName?: string | null, ) => void; + /** Merge an EXECUTING/EXECUTED event into the matching tool segment or + * batch `executions` slot. `executionRequestId` is optional — without it, + * an EXECUTED chunk pairs with the newest EXECUTING of the same + * (integratedToolType, toolFunction). Returns whether anything matched so + * the caller can append a fresh segment instead of dropping the event. */ updateToolExecutionInMessages: ( side: ChatSide, - executionRequestId: string, + executionRequestId: string | undefined, executedData: ToolExecutionSegment['data'], - ) => void; + ) => boolean; setApprovalStatus: (requestId: string, status: ApprovalStatus) => void; mergeApprovalStatuses: (entries: ApprovalStatusMap) => void; @@ -222,15 +227,24 @@ export const useTicketDetailsStore = create((set, get) => ({ let nextContent: MessageSegment[]; if (incomingCompaction) { - const startedIdx = existing.findIndex(seg => seg.type === 'context_compaction' && seg.status === 'started'); - const hasAnyCompaction = existing.some(seg => seg.type === 'context_compaction'); - if (incomingCompaction.status === 'completed' && startedIdx !== -1) { + // Mirror the lib's upsertTrailingCompaction: replace the LAST + // 'started' (earlier compactions in the bubble are already + // completed-in-place), else append. The previous first-match + + // any-compaction blanket silently dropped a second compaction + // landing in the same bubble. + let startedIdx = -1; + for (let k = existing.length - 1; k >= 0; k--) { + const seg = existing[k]; + if (seg.type === 'context_compaction' && seg.status === 'started') { + startedIdx = k; + break; + } + } + if (startedIdx !== -1) { nextContent = [...existing]; nextContent[startedIdx] = incomingCompaction; - } else if (!hasAnyCompaction) { - nextContent = [...existing, incomingCompaction]; } else { - nextContent = existing; + nextContent = [...existing, incomingCompaction]; } } else { nextContent = s.accumulator.replaySegments([...existing, ...segments]); @@ -244,7 +258,19 @@ export const useTicketDetailsStore = create((set, get) => ({ }; return { ...s, messages: updated }; } - return s; + // No assistant message anywhere (thread cleared / history not yet + // hydrated): open a fresh bubble instead of dropping the segments — + // mirrors the lib's appendToTrailingAssistant. Silently dropping the + // first post-MESSAGE_END tool chunk here is exactly the + // invisible-tool-run bug this pipeline was fixed for. + const fallback: ChatMessage = { + id: `assistant-${Date.now()}-${Math.random().toString(16).slice(2)}`, + role: 'assistant', + content: incomingCompaction ? [incomingCompaction] : [...segments], + timestamp: new Date(), + ...(streamSeq != null ? { streamSeq } : {}), + }; + return { ...s, messages: [...s.messages, fallback] }; }), ); }, @@ -294,63 +320,95 @@ export const useTicketDetailsStore = create((set, get) => ({ }; }), - updateToolExecutionInMessages: (side, executionRequestId, executedData) => + updateToolExecutionInMessages: (side, executionRequestId, executedData) => { + let matched = false; set(state => { - let matched = false; const nextSides = produceSide(state, side, s => { - const updatedMessages = s.messages.map(message => { - if (matched) return message; - if (message.role !== 'assistant' || !Array.isArray(message.content)) return message; - - let changed = false; - const updatedContent = message.content.map(segment => { - if (matched) return segment; - - if ( - segment.type === 'tool_execution' && - segment.data.type === 'EXECUTING_TOOL' && - segment.data.toolExecutionRequestId === executionRequestId - ) { - matched = true; - changed = true; - const merged: ToolExecutionSegment = { - type: 'tool_execution', - data: { - ...executedData, - toolTitle: executedData.toolTitle ?? segment.data.toolTitle, - parameters: executedData.parameters ?? segment.data.parameters, - }, - }; - return merged; - } - - if ( + // Scan NEWEST-first: an id-less EXECUTED must pair with the newest + // in-flight EXECUTING of the same tool (a stale interrupted card from + // an earlier turn must not swallow the current run); execId matches + // are unique so direction doesn't change them. + let msgIdx = -1; + let segIdx = -1; + for (let i = s.messages.length - 1; i >= 0 && msgIdx === -1; i--) { + const message = s.messages[i]; + if (message.role !== 'assistant' || !Array.isArray(message.content)) continue; + const content = message.content as MessageSegment[]; + for (let j = content.length - 1; j >= 0; j--) { + const segment = content[j]; + if (segment.type === 'tool_execution') { + const idMatches = executionRequestId + ? segment.data.toolExecutionRequestId === executionRequestId + : segment.data.type === 'EXECUTING_TOOL' && + segment.data.integratedToolType === executedData.integratedToolType && + segment.data.toolFunction === executedData.toolFunction; + if (idMatches) { + msgIdx = i; + segIdx = j; + break; + } + } else if ( + executionRequestId && segment.type === 'approval_batch' && segment.data.toolCalls.some(c => c.toolExecutionRequestId === executionRequestId) ) { - matched = true; - changed = true; - const prev: ApprovalBatchExecutionState | undefined = segment.data.executions?.[executionRequestId]; - const next: ApprovalBatchExecutionState = - executedData.type === 'EXECUTED_TOOL' - ? { status: 'done', result: executedData.result, success: executedData.success } - : { status: 'executing', result: prev?.result, success: prev?.success }; - return { - ...segment, - data: { - ...segment.data, - executions: { ...(segment.data.executions ?? {}), [executionRequestId]: next }, - }, - } as ApprovalBatchSegment; + msgIdx = i; + segIdx = j; + break; } + } + } - return segment; - }); - - return changed ? { ...message, content: updatedContent } : message; - }); + if (msgIdx === -1) return s; + matched = true; + + const message = s.messages[msgIdx]; + const content = message.content as MessageSegment[]; + const segment = content[segIdx]; + let nextSegment: MessageSegment; + + if (segment.type === 'tool_execution') { + // Never downgrade a completed run back to EXECUTING (JetStream + // redelivery of the EXECUTING chunk after EXECUTED landed). + // `matched` stays TRUE here on purpose: the run is already + // represented on screen, so the caller's append fallback must not + // fire — a `false` would duplicate the card. + if (executedData.type === 'EXECUTING_TOOL' && segment.data.type === 'EXECUTED_TOOL') { + return s; + } + nextSegment = { + type: 'tool_execution', + data: { + ...executedData, + toolTitle: executedData.toolTitle ?? segment.data.toolTitle, + parameters: executedData.parameters ?? segment.data.parameters, + }, + } satisfies ToolExecutionSegment; + } else { + const batch = segment as ApprovalBatchSegment; + const prev: ApprovalBatchExecutionState | undefined = batch.data.executions?.[executionRequestId as string]; + // Same no-downgrade rule for a batch slot: a redelivered EXECUTING + // must not flip a 'done' entry back (matched stays true → no append). + if (executedData.type === 'EXECUTING_TOOL' && prev?.status === 'done') { + return s; + } + const next: ApprovalBatchExecutionState = + executedData.type === 'EXECUTED_TOOL' + ? { status: 'done', result: executedData.result, success: executedData.success } + : { status: 'executing', result: prev?.result, success: prev?.success }; + nextSegment = { + ...batch, + data: { + ...batch.data, + executions: { ...(batch.data.executions ?? {}), [executionRequestId as string]: next }, + }, + } as ApprovalBatchSegment; + } - if (!matched) return s; + const nextContent = [...content]; + nextContent[segIdx] = nextSegment; + const updatedMessages = [...s.messages]; + updatedMessages[msgIdx] = { ...message, content: nextContent }; const updatedStreaming = s.streaming ? (updatedMessages.find(m => m.id === s.streaming?.id) ?? s.streaming) @@ -366,7 +424,9 @@ export const useTicketDetailsStore = create((set, get) => ({ // this guard zustand notifies every subscriber on every no-match. if (!matched) return state; return nextSides; - }), + }); + return matched; + }, setApprovalStatus: (requestId, status) => set(state =>