Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
30 commits
Select commit Hold shift + click to select a range
c758ed6
feat(ai): cache-aware prompt assembly, breakpoints, tool stabilization
2witstudios Jun 12, 2026
173fcd2
feat(compaction): PR 2 tasks 1–6 — schema, pure cores, service glue
2witstudios Jun 12, 2026
9c708ae
Merge branch 'pu/pr2-compaction-core' into pu/sliding-window-compacte…
2witstudios Jun 12, 2026
10cd514
feat(compaction): PR 2 tasks 7–9 — entry-point wiring and invalidatio…
2witstudios Jun 12, 2026
e6e8f00
chore(db): drop compaction migration for regeneration after master merge
2witstudios Jun 12, 2026
1fe8bc5
Merge remote-tracking branch 'origin/master' into pu/sliding-window-c…
2witstudios Jun 12, 2026
bbe53e7
feat(db): regenerate conversation_compactions migration as 0160
2witstudios Jun 12, 2026
60e3d1b
feat(context): cross-request tool-result eliding + recall polish (PR 3)
2witstudios Jun 12, 2026
7d9aa16
feat(agents): durable memory page + provider-factory user-row threadi…
2witstudios Jun 12, 2026
e0f266e
chore(db): drop compaction migration for regeneration after master me…
2witstudios Jun 12, 2026
c874bd0
Merge remote-tracking branch 'origin/master' into pu/sliding-window-c…
2witstudios Jun 12, 2026
0397ab8
feat(db): regenerate conversation_compactions migration as 0161
2witstudios Jun 12, 2026
0cf8e2c
fix(ai): address bot review — UIMessage tool parts in summaries, exec…
2witstudios Jun 12, 2026
531b61a
fix(security): mask conversationId in compaction log to prevent log i…
2witstudios Jun 12, 2026
90b13dc
test(compaction): assert accumulated usage across re-condense passes
2witstudios Jun 12, 2026
0cf93e6
fix(ai): address second batch of bot review — repo invalidation, elis…
2witstudios Jun 12, 2026
cf31929
fix(ai): address third batch of bot review — summary handling and tok…
2witstudios Jun 12, 2026
3e6a8d7
fix(ai): eliminate type drift and write-tool duplication (bot review …
2witstudios Jun 12, 2026
13a4ac2
test(ai): add compaction-aware context assembly regression tests for …
2witstudios Jun 12, 2026
f4109e4
fix(ai): apply <conversation_summary> prefix guard to v1 route hasSum…
2witstudios Jun 12, 2026
0a4439b
Merge remote-tracking branch 'origin/master' into pu/sliding-window-c…
2witstudios Jun 12, 2026
b463611
fix(ai): address human review — integrity gate, shared summary sentin…
2witstudios Jun 12, 2026
e585fb1
chore(db): drop compaction migration for regeneration after master me…
2witstudios Jun 12, 2026
adae3fc
Merge remote-tracking branch 'origin/master' into pu/sliding-window-c…
2witstudios Jun 12, 2026
b746b35
feat(db): regenerate conversation_compactions migration as 0162
2witstudios Jun 12, 2026
852aabe
fix(ai): address second review round — estimator coverage, billing on…
2witstudios Jun 12, 2026
9bf07ea
fix(ai): address third review round — sanitized masking, guarded glob…
2witstudios Jun 12, 2026
b6a377a
chore(db): drop compaction migration for regeneration after master me…
2witstudios Jun 12, 2026
72c70ca
Merge remote-tracking branch 'origin/master' into pu/sliding-window-c…
2witstudios Jun 12, 2026
63449be
feat(db): regenerate conversation_compactions migration as 0163
2witstudios Jun 12, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 29 additions & 0 deletions apps/web/src/app/api/ai/chat/messages/[messageId]/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ import { getActorInfo, logMessageActivity } from '@pagespace/lib/monitoring/acti
import { broadcastAiMessageEdited, broadcastAiMessageDeleted } from '@/lib/websocket/socket-utils';
import { resolveTriggeredBy } from '@/lib/websocket/broadcast-triggered-by';
import { convertDbMessageToUIMessage } from '@/lib/ai/core/message-utils';
import { getState, invalidate } from '@/lib/ai/core/compaction/compaction-repository';

const AUTH_OPTIONS = { allow: ['session', 'mcp'] as const, requireCSRF: true };

Expand Down Expand Up @@ -102,6 +103,20 @@ export async function PATCH(
// Update the message content and set editedAt
await chatMessageRepository.updateMessageContent(messageId, updatedContent);

// Invalidate compaction if the edited message was in the compacted range.
// Awaited so the stale summary cannot be read by a concurrent request before we return.
try {
const state = await getState(message.conversationId, { source: 'page', pageId: message.pageId });
// No row: a first compaction may be in flight — write the invalidation
// tombstone so its pending insert loses. Row present: invalidate only
// when the touched message is inside the compacted range.
if (!state || (state.compactedUpToCreatedAt && message.createdAt <= state.compactedUpToCreatedAt)) {
await invalidate(message.conversationId, { source: 'page', pageId: message.pageId });
}
} catch (err) {
loggers.api.error('Failed to invalidate compaction state after message edit', err as Error);
}

// Broadcast to remote viewers. Failure must never break the request.
void (async () => {
try {
Expand Down Expand Up @@ -222,6 +237,20 @@ export async function DELETE(
// Soft delete the message
await chatMessageRepository.softDeleteMessage(messageId);

// Invalidate compaction if the deleted message was in the compacted range.
// Awaited so the stale summary cannot be read by a concurrent request before we return.
try {
const state = await getState(message.conversationId, { source: 'page', pageId: message.pageId });
// No row: a first compaction may be in flight — write the invalidation
// tombstone so its pending insert loses. Row present: invalidate only
// when the touched message is inside the compacted range.
if (!state || (state.compactedUpToCreatedAt && message.createdAt <= state.compactedUpToCreatedAt)) {
await invalidate(message.conversationId, { source: 'page', pageId: message.pageId });
}
} catch (err) {
loggers.api.error('Failed to invalidate compaction state after message delete', err as Error);
}

// Broadcast to remote viewers. Failure must never break the request.
void (async () => {
try {
Expand Down
183 changes: 131 additions & 52 deletions apps/web/src/app/api/ai/chat/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import {
createUIMessageStreamResponse,
type TextUIPart,
type ToolSet,
type ModelMessage,
} from 'ai';
import { ONPREM_ALLOWED_PROVIDERS, DEFAULT_PROVIDER, DEFAULT_MODEL, ADMIN_ONLY_PROVIDERS, resolveProviderModel } from '@/lib/ai/core/ai-providers-config';
import { ALL_PROVIDER_NAMES } from '@/lib/ai/core/ai-utils';
Expand Down Expand Up @@ -56,6 +57,13 @@ import { getModelCapabilities } from '@/lib/ai/core/model-capabilities';
import { convertMCPToolsToAISDKSchemas, parseMCPToolName, sanitizeToolNamesForProvider } from '@/lib/ai/core/mcp-tool-converter';
import { getUserPersonalization } from '@/lib/ai/core/personalization-utils';
import { applyToolExposureMode } from '@/lib/ai/tools/tool-exposure';
import {
buildVolatileTurnContext,
appendTurnContextToLastUserMessage,
withCacheBreakpoints,
} from '@/lib/ai/core/prompt-assembly';
import { prepareHistoryForModel } from '@/lib/ai/core/context-assembly';
import { getAgentMemoryContext, buildAgentMemorySection } from '@/lib/ai/core/agent-memory';

// Runtime-toggled tools that must stay directly callable even in search mode.
const ALWAYS_UPFRONT_TOOLS = new Set(['web_search']);
Expand Down Expand Up @@ -571,7 +579,8 @@ export async function POST(request: Request) {
selectedModel,
};

const providerResult = await createAIProvider(userId, providerRequest);
// Thread the already-loaded user row through the factory so it skips redundant DB selects.
const providerResult = await createAIProvider(userId, providerRequest, { user: user ?? null });

if (isProviderError(providerResult)) {
return createProviderErrorResponse(providerResult);
Expand All @@ -588,8 +597,8 @@ export async function POST(request: Request) {
? makeOnStepFinishHandler(creditAbortController, availableBalanceCents, resolvedModelName ?? 'unknown')
: null;

// Update user's current provider/model if changed
await updateUserProviderSettings(userId, selectedProvider, selectedModel);
// Update user's current provider/model if changed (thread the loaded row to skip a DB select).
await updateUserProviderSettings(userId, selectedProvider, selectedModel, { user: user ?? null });

// Parse read-only mode (defaults to false for full access)
const readOnlyMode = isReadOnly === true;
Expand Down Expand Up @@ -680,8 +689,11 @@ export async function POST(request: Request) {

// Create execute functions that signal client-side execution
// The AI SDK will call these, but we throw a special error that the client intercepts
// Sort keys so tool array order is deterministic across requests (only real config
// changes — webSearch/readOnly/MCP/exposure-mode — may change the tool array).
const mcpToolsWithExecute: Record<string, unknown> = {};
for (const [toolName, toolSchema] of Object.entries(mcpToolSchemas)) {
for (const toolName of Object.keys(mcpToolSchemas).sort()) {
const toolSchema = mcpToolSchemas[toolName];
mcpToolsWithExecute[toolName] = {
...toolSchema,
execute: async (args: Record<string, unknown>) => {
Expand Down Expand Up @@ -776,49 +788,8 @@ export async function POST(request: Request) {
// Always inject the finish tool so the model can signal task completion
filteredTools = { ...filteredTools, ...finishTool } as ToolSet;

loggers.ai.debug('AI Chat API: Loading conversation history', {
pageId: chatId
});

const pageId = chatId as string;
const dbMessages = await db
.select()
.from(chatMessages)
.where(and(
eq(chatMessages.pageId, pageId),
eq(chatMessages.conversationId, conversationId),
eq(chatMessages.isActive, true)
))
.orderBy(chatMessages.createdAt);

const conversationHistory: UIMessage[] = dbMessages.map(msg =>
convertDbMessageToUIMessage({
id: msg.id,
pageId: msg.pageId,
userId: msg.userId,
role: msg.role,
content: msg.content,
toolCalls: msg.toolCalls,
toolResults: msg.toolResults,
createdAt: msg.createdAt,
isActive: msg.isActive,
editedAt: msg.editedAt,
messageType: msg.messageType === 'todo_list' ? 'todo_list' : 'standard',
})
);

loggers.ai.debug('AI Chat API: Loaded conversation from database', {
messageCount: conversationHistory.length,
pageId
});

// Convert UIMessages to ModelMessages for the AI model
// First sanitize messages to remove tool parts without results (prevents "input-available" state errors)
// NOTE: We use database-loaded messages, NOT messages from client
const sanitizedMessages = sanitizeMessagesForModel(conversationHistory);
const modelMessages = convertToModelMessages(sanitizedMessages, {
tools: filteredTools // Use original tools - no wrapping needed
});
// Build system prompt BEFORE history loading so its token estimate is
// available for prepareConversationContext's context-window budget math.

// Fetch user personalization for AI system prompt injection
const personalization = await getUserPersonalization(userId);
Expand Down Expand Up @@ -865,7 +836,7 @@ export async function POST(request: Request) {
personalization ?? undefined
);
}

// Build timestamp system prompt for temporal awareness
const userTimezone = user?.timezone ?? undefined;
const timestampSystemPrompt = buildTimestampSystemPrompt(userTimezone);
Expand All @@ -888,6 +859,83 @@ export async function POST(request: Request) {
}
}

// Build agent memory section (AI_CHAT pages only). Fetches the "Agent Memory"
// child page content — stable per request, only changes when the agent edits
// the page, so it lives in the STABLE system section (not the volatile block).
let agentMemoryPrompt = '';
if (page.type === 'AI_CHAT') {
const memoryContent = await getAgentMemoryContext(chatId, userId);
agentMemoryPrompt = buildAgentMemorySection(memoryContent);
}

loggers.ai.debug('AI Chat API: Loading conversation history', {
pageId: chatId
});

const pageId = chatId as string;
const dbMessages = await db
.select()
.from(chatMessages)
.where(and(
eq(chatMessages.pageId, pageId),
eq(chatMessages.conversationId, conversationId),
eq(chatMessages.isActive, true)
))
.orderBy(chatMessages.createdAt);

const conversationHistory: UIMessage[] = dbMessages.map(msg =>
convertDbMessageToUIMessage({
id: msg.id,
pageId: msg.pageId,
userId: msg.userId,
role: msg.role,
content: msg.content,
toolCalls: msg.toolCalls,
toolResults: msg.toolResults,
createdAt: msg.createdAt,
isActive: msg.isActive,
editedAt: msg.editedAt,
messageType: msg.messageType === 'todo_list' ? 'todo_list' : 'standard',
})
);

loggers.ai.debug('AI Chat API: Loaded conversation from database', {
messageCount: conversationHistory.length,
pageId
});

// Sanitize, compact, and elide — all in the unified seam.
// createUIMessageStream keeps the FULL conversationHistory for the UI;
// only the model-facing messages go through the seam.
const {
messages: preparedMessages,
summaryText,
stableBoundaryIndex,
scheduleCompaction,
} = await prepareHistoryForModel({
history: conversationHistory,
conversationId: conversationId!,
source: 'page',
pageId,
model: resolvedModelName ?? currentModel,
provider: resolvedProvider ?? currentProvider,
systemPrompt: systemPrompt + pageTreePrompt + agentMemoryPrompt + toolDiscoveryPrompt,
tools: filteredTools as Record<string, unknown>,
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
user: user ? { id: user.id, role: user.role } : null,
});

// Convert elided UIMessages to ModelMessages for streamText.
const tailModelMessages = convertToModelMessages(preparedMessages, { tools: filteredTools });
const modelMessages: ModelMessage[] = summaryText
? [{ role: 'user' as const, content: summaryText }, ...tailModelMessages]
: tailModelMessages;

// Intentional second sanitize (prepareHistoryForModel already sanitized once):
// createUIMessageStream must receive the FULL conversation history for the UI
// (originalMessages), not the compacted/elided model tail in preparedMessages —
// so conversationHistory is sanitized directly instead of reusing the seam output.
const sanitizedMessages = sanitizeMessagesForModel(conversationHistory);
Comment thread
2witstudios marked this conversation as resolved.
Resolved

loggers.ai.debug('AI Chat API: Tools configured for Page AI', { toolCount: Object.keys(filteredTools).length });
loggers.ai.info('AI Chat API: Starting streamText for Page AI', { model: currentModel, pageName: page.title });

Expand Down Expand Up @@ -959,10 +1007,29 @@ export async function POST(request: Request) {
maxSteps: AGENT_MAX_STEPS,
startTimeMs: startTime,
logger: loggers.ai,
buildStreamText: (messages) => streamText({
buildStreamText: (messages) => {
// Volatile per-turn data (timestamp/mention/command) is appended
// to the last user message so the system prefix stays byte-stable
// and provider prefix caches (Anthropic/OpenAI/Gemini) are not
// invalidated on every turn.
const turnContext = buildVolatileTurnContext({
timestampPrompt: timestampSystemPrompt,
mentionPrompt: mentionSystemPrompt,
commandPrompt: commandSystemPrompt,
});
const messagesWithContext = appendTurnContextToLastUserMessage(messages, turnContext);
// Apply cache breakpoints:
// A) last message — covers system+tools+history every step after step 1.
// B) stableBoundaryIndex — the first tail message after the compaction
// summary (index 1 when a summary exists, 0 = disabled otherwise).
// This cross-request breakpoint survives until the next recompaction.
const cachedMessages = withCacheBreakpoints(messagesWithContext, stableBoundaryIndex);
return streamText({
model,
system: systemPrompt + mentionSystemPrompt + commandSystemPrompt + timestampSystemPrompt + pageTreePrompt + toolDiscoveryPrompt,
messages,
// Stable system prompt — no volatile sections; stays byte-identical
// across turns so provider prefix caches survive per request.
system: systemPrompt + pageTreePrompt + agentMemoryPrompt + toolDiscoveryPrompt,
messages: cachedMessages,
tools: filteredTools,
stopWhen: [hasToolCall(FINISH_TOOL_NAME), stepCountIs(AGENT_MAX_STEPS)],
// abortSignal from the abort registry — only fires on explicit user stop, never on client disconnect
Expand Down Expand Up @@ -1021,7 +1088,15 @@ export async function POST(request: Request) {
});
lifecycle!.finish(true);
},
}),
// Re-mark breakpoints per step so mid-loop tool results are cached.
// stableBoundaryIndex stays fixed (the summary is always at position
// 0; its first tail neighbour at position 1 remains stable as new
// messages are appended to the END of the accumulating array).
prepareStep: ({ messages: stepMessages }) => ({
messages: withCacheBreakpoints(stepMessages, stableBoundaryIndex),
}),
})
},
});

// Billing reads the SUMMED usage / OpenRouter cost across every attempt
Expand Down Expand Up @@ -1202,6 +1277,10 @@ export async function POST(request: Request) {
// Don't fail the response - but the hold may remain for the reconcile sweep.
}

// Schedule compaction for the NEXT request (summarises old tail via after()).
// Runs after the response is fully sent; non-fatal if it fails.
scheduleCompaction();

// Reflect a user stop, including one that landed during inter-attempt backoff or
// raced in after the loop broke (onAbort only fires while a streamText is live).
// finish() is idempotent, so this is a no-op if onAbort already ran.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -104,12 +104,14 @@ const mockMessage = (overrides: Partial<{
content: string;
role: string;
isActive: boolean;
createdAt: Date;
}> = {}) => ({
id: overrides.id ?? mockMessageId,
conversationId: overrides.conversationId ?? mockConversationId,
content: overrides.content ?? 'Hello, AI!',
role: overrides.role ?? 'user',
isActive: overrides.isActive ?? true,
createdAt: overrides.createdAt ?? new Date('2024-01-01T00:00:10Z'),
});

const createContext = (id: string, messageId: string) => ({
Expand Down
31 changes: 31 additions & 0 deletions apps/web/src/app/api/ai/global/[id]/messages/[messageId]/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import { getActorInfo, logMessageActivity } from '@pagespace/lib/monitoring/acti
import { broadcastAiMessageEdited, broadcastAiMessageDeleted } from '@/lib/websocket/socket-utils';
import { resolveTriggeredBy } from '@/lib/websocket/broadcast-triggered-by';
import { globalChannelId } from '@pagespace/lib/ai/global-channel-id';
import { getState, invalidate } from '@/lib/ai/core/compaction/compaction-repository';

const AUTH_OPTIONS = { allow: ['session'] as const, requireCSRF: true };

Expand Down Expand Up @@ -64,6 +65,21 @@ export async function PATCH(
// Update the message content
await globalConversationRepository.updateMessageContent(messageId, updatedContent);

// Invalidate compaction when a message in this conversation is edited (stale summary guard).
// Awaited so the stale summary cannot be read by a concurrent request before we return.
try {
const state = await getState(conversationId, { source: 'global' });
// No row: a first compaction may be in flight — write the invalidation
// tombstone so its pending insert loses. Row present: invalidate only
// when the touched message is inside the compacted range (later edits
// leave the still-valid summary in place).
if (!state || (state.compactedUpToCreatedAt && message.createdAt <= state.compactedUpToCreatedAt)) {
await invalidate(conversationId, { source: 'global' });
}
} catch (err) {
loggers.api.error('Failed to invalidate compaction state after global message edit', err as Error);
}

// Broadcast to remote viewers (other tabs of this user). Failure must never break the request.
void (async () => {
try {
Expand Down Expand Up @@ -168,6 +184,21 @@ export async function DELETE(
// Soft delete the message
await globalConversationRepository.softDeleteMessage(messageId);

// Invalidate compaction when a message in this conversation is deleted (stale summary guard).
// Awaited so the stale summary cannot be read by a concurrent request before we return.
try {
const state = await getState(conversationId, { source: 'global' });
// No row: a first compaction may be in flight — write the invalidation
// tombstone so its pending insert loses. Row present: invalidate only
// when the touched message is inside the compacted range (later edits
// leave the still-valid summary in place).
if (!state || (state.compactedUpToCreatedAt && message.createdAt <= state.compactedUpToCreatedAt)) {
await invalidate(conversationId, { source: 'global' });
}
} catch (err) {
loggers.api.error('Failed to invalidate compaction state after global message delete', err as Error);
}

// Broadcast to remote viewers (other tabs of this user). Failure must never break the request.
void (async () => {
try {
Expand Down
Loading
Loading