brother7
16f5c57f17
Require assistant natural language after the latest user message so a completed prompt cannot reuse historical replies when queuing the observation snapshot.
431 lines
15 KiB
TypeScript
431 lines
15 KiB
TypeScript
import { extractText } from '@/pages/Chat/message-utils';
|
|
import { useAuthStore } from '@/stores/auth';
|
|
import {
|
|
getProfileAccountKey,
|
|
isUserProfileComplete,
|
|
useUserProfileStore,
|
|
} from '@/stores/user-profile';
|
|
import type { RawMessage } from '@/types/chat';
|
|
import type { ConversationSnapshot } from '@/types/coding-conversation';
|
|
import type {
|
|
AgentSessionData,
|
|
AgentSessionMessage,
|
|
AgentSessionMessageRole,
|
|
} from '../../shared/agent-session';
|
|
|
|
const PENDING_AGENT_SESSION_SYNC_STORAGE_KEY = 'niancode-agent-session-sync-pending';
|
|
const SESSION_UPLOAD_DEBOUNCE_MS = 1_000;
|
|
const SESSION_UPLOAD_RETRY_DELAYS_MS = [5_000, 30_000, 120_000, 300_000];
|
|
|
|
type PendingUpload = {
|
|
userId: string;
|
|
data: AgentSessionData;
|
|
retryCount: number;
|
|
};
|
|
|
|
const pendingUploads = new Map<string, PendingUpload>();
|
|
const uploadTimers = new Map<string, ReturnType<typeof setTimeout>>();
|
|
const inFlightUploads = new Map<string, Promise<void>>();
|
|
let pendingUploadsHydrated = false;
|
|
|
|
const ANSI_ESCAPE_PATTERN = new RegExp(`${String.fromCharCode(27)}\\[[0-?]*[ -/]*[@-~]`, 'g');
|
|
const FENCED_CODE_PATTERN = /```[\s\S]*?(?:```|$)/g;
|
|
const TILDE_FENCED_CODE_PATTERN = /~~~[\s\S]*?(?:~~~|$)/g;
|
|
const INLINE_CODE_PATTERN = /`[^`\n]*`/g;
|
|
const MEDIA_MARKER_PATTERN = /\b(?:MEDIA|media):(?:\/|~\/)[^\s"'<>()[\]{}，。；：！？、]+/g;
|
|
const ARTIFACT_URL_PATTERN = /https?:\/\/[^\s"'<>()[\]{}，。；：！？、]+\.(?:png|jpe?g|gif|webp|bmp|avif|svg|pdf|docx?|xlsx?|pptx?|txt|csv|md|zip|tar|gz|rar|7z|mp3|wav|ogg|aac|flac|m4a|mp4|mov|avi|mkv|webm|m4v)(?:[?#][^\s"'<>()[\]{}，。；：！？、]+)?/gi;
|
|
const ABSOLUTE_PATH_PATTERN = /(?<![A-Za-z0-9_])(?:~\/|\/(?:Users|home|private|tmp|var|opt|mnt|Volumes|Documents|Desktop|Downloads|workspace|workspaces|repo|project|src|dist|build)(?:\/|[^\s"'`<>()[\]{}，。；：！？、])[^\s"'`<>()[\]{}，。；：！？、]*|[A-Za-z]:[\\/][^\s"'`<>()[\]{}，。；：！？、]*|\\\\[^\s"'`<>()[\]{}，。；：！？、]+)/g;
|
|
const RELATIVE_PATH_PATTERN = /(?<![A-Za-z0-9_])(?:\.\.?[\\/]|(?:src|dist|build|public|app|lib|components|pages|tests|electron|shared|resources)[\\/])[^\s"'`<>()[\]{}，。；：！？、]+/gi;
|
|
const SOURCE_FILE_NAME_PATTERN = /(?<![A-Za-z0-9_])[A-Za-z0-9_-]+\.(?:c|cc|cpp|cs|css|go|h|hpp|html?|java|js|json|jsx|kt|md|php|py|rb|rs|scss|sh|sql|swift|ts|tsx|vue|xml|yaml|yml)(?![A-Za-z0-9_])/gi;
|
|
const CODE_LIKE_LINE_PATTERN = /^\s*(?:#include\b|(?:const|let|var|function|class|interface|type|import|export|def|async\s+function)\b|(?:SELECT|INSERT\s+INTO|UPDATE\s+.+\s+SET|DELETE\s+FROM)\b)/;
|
|
|
|
function normalizeWhitespace(value: string): string {
|
|
return value
|
|
.replace(/[ \t]+\n/g, '\n')
|
|
.replace(/\n{3,}/g, '\n\n')
|
|
.trim();
|
|
}
|
|
|
|
function stripLogLines(value: string): string {
|
|
return value
|
|
.split('\n')
|
|
.filter((line) => {
|
|
const trimmed = line.trim();
|
|
if (!trimmed) return true;
|
|
return !(
|
|
/^\[(?:\d{4}-\d{2}-\d{2}[T ][^\]]+|(?:DEBUG|INFO|WARN|ERROR|TRACE)[^\]]*)\]/i.test(trimmed)
|
|
|| /^(?:DEBUG|INFO|WARN|ERROR|TRACE)\s*(?:\||:)/i.test(trimmed)
|
|
|| /^at\s+[^\s(]+\s*\([^)]*\)/.test(trimmed)
|
|
|| /^(?:GET|POST|PUT|PATCH|DELETE|OPTIONS|HEAD)\s+\//.test(trimmed)
|
|
);
|
|
})
|
|
.join('\n');
|
|
}
|
|
|
|
function stripCodeLikeLines(value: string): string {
|
|
return value
|
|
.split('\n')
|
|
.filter((line) => !CODE_LIKE_LINE_PATTERN.test(line))
|
|
.join('\n');
|
|
}
|
|
|
|
/**
|
|
* Keep only the natural-language part of a user/assistant message before it
|
|
* leaves the local workspace. Runtime metadata and tool parts are removed by
|
|
* the existing message projection; this second pass removes code, paths,
|
|
* logs, and artifact links from visible text as well.
|
|
*/
|
|
export function sanitizeAgentSessionText(value: string): string {
|
|
const withoutRuntimeArtifacts = value
|
|
.replace(ANSI_ESCAPE_PATTERN, '')
|
|
.replace(FENCED_CODE_PATTERN, '\n')
|
|
.replace(TILDE_FENCED_CODE_PATTERN, '\n')
|
|
.replace(INLINE_CODE_PATTERN, '')
|
|
.replace(MEDIA_MARKER_PATTERN, '')
|
|
.replace(ARTIFACT_URL_PATTERN, '')
|
|
.replace(ABSOLUTE_PATH_PATTERN, '')
|
|
.replace(RELATIVE_PATH_PATTERN, '')
|
|
.replace(SOURCE_FILE_NAME_PATTERN, '');
|
|
return normalizeWhitespace(stripCodeLikeLines(stripLogLines(withoutRuntimeArtifacts)));
|
|
}
|
|
|
|
function getMessageText(message: RawMessage): string {
|
|
// `extractText` keeps text blocks while ignoring thinking/tool/image parts;
|
|
// using the Codex final-answer projection here would incorrectly discard
|
|
// natural-language narration that appears before a tool call.
|
|
const visibleText = extractText(message);
|
|
return sanitizeAgentSessionText(visibleText);
|
|
}
|
|
|
|
function toIsoTimestamp(value: unknown): string | undefined {
|
|
if (typeof value === 'number' && Number.isFinite(value)) {
|
|
const milliseconds = Math.abs(value) < 1_000_000_000_000 ? value * 1_000 : value;
|
|
const date = new Date(milliseconds);
|
|
return Number.isNaN(date.getTime()) ? undefined : date.toISOString();
|
|
}
|
|
if (typeof value === 'string' && value.trim()) {
|
|
const date = new Date(value);
|
|
return Number.isNaN(date.getTime()) ? undefined : date.toISOString();
|
|
}
|
|
return undefined;
|
|
}
|
|
|
|
function getMessageTimestamp(message: RawMessage): string | undefined {
|
|
const record = message as RawMessage & {
|
|
createdAt?: unknown;
|
|
created_at?: unknown;
|
|
updatedAt?: unknown;
|
|
updated_at?: unknown;
|
|
};
|
|
return toIsoTimestamp(message.timestamp)
|
|
?? toIsoTimestamp(record.createdAt)
|
|
?? toIsoTimestamp(record.created_at)
|
|
?? toIsoTimestamp(record.updatedAt)
|
|
?? toIsoTimestamp(record.updated_at);
|
|
}
|
|
|
|
function getMessageId(message: RawMessage): string | undefined {
|
|
return typeof message.id === 'string' && message.id.trim()
|
|
? message.id.trim()
|
|
: undefined;
|
|
}
|
|
|
|
function isSyncableRole(role: RawMessage['role']): role is AgentSessionMessageRole {
|
|
return role === 'user' || role === 'assistant';
|
|
}
|
|
|
|
/** Build a wire-safe full snapshot for one locally completed session. */
|
|
export function buildAgentSessionData(
|
|
projectId: string,
|
|
sessionId: string,
|
|
messages: readonly RawMessage[],
|
|
updatedAt = new Date().toISOString(),
|
|
): AgentSessionData | null {
|
|
const normalizedProjectId = projectId.trim();
|
|
const normalizedSessionId = sessionId.trim();
|
|
if (!normalizedProjectId || !normalizedSessionId) return null;
|
|
|
|
const normalizedMessages: AgentSessionMessage[] = [];
|
|
const indexById = new Map<string, number>();
|
|
for (const message of messages) {
|
|
if (!isSyncableRole(message.role) || message.isError) continue;
|
|
const text = getMessageText(message);
|
|
if (!text) continue;
|
|
|
|
const next: AgentSessionMessage = {
|
|
...(getMessageId(message) ? { id: getMessageId(message) } : {}),
|
|
role: message.role,
|
|
...(getMessageTimestamp(message) ? { created_at: getMessageTimestamp(message) } : {}),
|
|
text,
|
|
};
|
|
const messageId = next.id;
|
|
if (messageId && indexById.has(messageId)) {
|
|
normalizedMessages[indexById.get(messageId)!] = next;
|
|
continue;
|
|
}
|
|
if (messageId) indexById.set(messageId, normalizedMessages.length);
|
|
normalizedMessages.push(next);
|
|
}
|
|
|
|
if (normalizedMessages.length === 0) return null;
|
|
return {
|
|
project_id: normalizedProjectId,
|
|
session_id: normalizedSessionId,
|
|
updated_at: updatedAt,
|
|
messages: normalizedMessages,
|
|
};
|
|
}
|
|
|
|
function getStorage(): Storage | null {
|
|
if (typeof window === 'undefined') return null;
|
|
try {
|
|
return window.localStorage;
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function isAgentSessionData(value: unknown): value is AgentSessionData {
|
|
if (!value || typeof value !== 'object' || Array.isArray(value)) return false;
|
|
const record = value as Record<string, unknown>;
|
|
if (
|
|
typeof record.project_id !== 'string'
|
|
|| typeof record.session_id !== 'string'
|
|
|| typeof record.updated_at !== 'string'
|
|
|| !Array.isArray(record.messages)
|
|
) return false;
|
|
return record.messages.every((message) => (
|
|
Boolean(message)
|
|
&& typeof message === 'object'
|
|
&& !Array.isArray(message)
|
|
&& ((message as Record<string, unknown>).role === 'user'
|
|
|| (message as Record<string, unknown>).role === 'assistant')
|
|
&& typeof (message as Record<string, unknown>).text === 'string'
|
|
));
|
|
}
|
|
|
|
function getCurrentAccountKey(): string | null {
|
|
const authUser = useAuthStore.getState().user;
|
|
return getProfileAccountKey(authUser?.userId ?? authUser?.username);
|
|
}
|
|
|
|
function getUploadKey(userId: string, data: AgentSessionData): string {
|
|
return `${userId}:${data.project_id}:${data.session_id}`;
|
|
}
|
|
|
|
function hydratePendingUploads(): void {
|
|
if (pendingUploadsHydrated) return;
|
|
pendingUploadsHydrated = true;
|
|
const storage = getStorage();
|
|
if (!storage) return;
|
|
|
|
try {
|
|
const parsed = JSON.parse(storage.getItem(PENDING_AGENT_SESSION_SYNC_STORAGE_KEY) ?? '[]') as unknown;
|
|
if (!Array.isArray(parsed)) return;
|
|
for (const value of parsed) {
|
|
const record = value && typeof value === 'object' && !Array.isArray(value)
|
|
? value as Record<string, unknown>
|
|
: null;
|
|
const userId = getProfileAccountKey(
|
|
typeof record?.userId === 'string' ? record.userId : undefined,
|
|
);
|
|
const data = record?.data;
|
|
// Queue entries are account-bound. Do not guess an owner for records
|
|
// written by an older build, otherwise a different login could receive
|
|
// another account's local conversation data.
|
|
if (!userId || !isAgentSessionData(data)) continue;
|
|
const retryCount = typeof record?.retryCount === 'number' && Number.isFinite(record.retryCount)
|
|
? Math.max(0, Math.floor(record.retryCount))
|
|
: 0;
|
|
pendingUploads.set(getUploadKey(userId, data), { userId, data, retryCount });
|
|
}
|
|
} catch {
|
|
// A malformed retry queue must not prevent the app from starting.
|
|
}
|
|
}
|
|
|
|
function persistPendingUploads(): void {
|
|
const storage = getStorage();
|
|
if (!storage) return;
|
|
try {
|
|
storage.setItem(
|
|
PENDING_AGENT_SESSION_SYNC_STORAGE_KEY,
|
|
JSON.stringify([...pendingUploads.values()]),
|
|
);
|
|
} catch {
|
|
// Quota/private-mode failures should not affect the active conversation.
|
|
}
|
|
}
|
|
|
|
function clearUploadTimer(key: string): void {
|
|
const timer = uploadTimers.get(key);
|
|
if (timer === undefined) return;
|
|
clearTimeout(timer);
|
|
uploadTimers.delete(key);
|
|
}
|
|
|
|
function scheduleUpload(key: string, delayMs: number): void {
|
|
clearUploadTimer(key);
|
|
uploadTimers.set(key, setTimeout(() => {
|
|
uploadTimers.delete(key);
|
|
void flushUpload(key);
|
|
}, delayMs));
|
|
}
|
|
|
|
function getRetryDelay(retryCount: number): number {
|
|
return SESSION_UPLOAD_RETRY_DELAYS_MS[
|
|
Math.min(Math.max(retryCount - 1, 0), SESSION_UPLOAD_RETRY_DELAYS_MS.length - 1)
|
|
];
|
|
}
|
|
|
|
async function uploadWithCurrentProfile(userId: string, data: AgentSessionData): Promise<void> {
|
|
if (getCurrentAccountKey() !== userId) {
|
|
throw new Error('Authenticated account changed during Agent session sync');
|
|
}
|
|
const accessToken = await useAuthStore.getState().getValidAccessToken();
|
|
if (!accessToken) throw new Error('No authenticated session for Agent session sync');
|
|
|
|
if (getCurrentAccountKey() !== userId) {
|
|
throw new Error('Authenticated account changed during Agent session sync');
|
|
}
|
|
|
|
const profile = useUserProfileStore.getState().profilesByUserId[userId];
|
|
// Session observation must never pull cloud profile data into the local
|
|
// workspace. The normal profile bootstrap owns that read; until it has
|
|
// produced a local profile, this upload remains queued for a retry.
|
|
if (!isUserProfileComplete(profile)) {
|
|
throw new Error('Agent profile is not complete');
|
|
}
|
|
|
|
await useUserProfileStore.getState().pushSessionData(userId, accessToken, data);
|
|
}
|
|
|
|
async function flushUpload(key: string): Promise<void> {
|
|
const existing = inFlightUploads.get(key);
|
|
if (existing) return existing;
|
|
|
|
const task = (async () => {
|
|
hydratePendingUploads();
|
|
const pending = pendingUploads.get(key);
|
|
if (!pending) return;
|
|
if (getCurrentAccountKey() !== pending.userId) return;
|
|
|
|
pendingUploads.delete(key);
|
|
persistPendingUploads();
|
|
try {
|
|
await uploadWithCurrentProfile(pending.userId, pending.data);
|
|
} catch {
|
|
if (!pendingUploads.has(key)) {
|
|
const retryCount = pending.retryCount + 1;
|
|
pendingUploads.set(key, { userId: pending.userId, data: pending.data, retryCount });
|
|
persistPendingUploads();
|
|
if (getCurrentAccountKey() === pending.userId) {
|
|
scheduleUpload(key, getRetryDelay(retryCount));
|
|
}
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (pendingUploads.has(key)) {
|
|
scheduleUpload(key, SESSION_UPLOAD_DEBOUNCE_MS);
|
|
}
|
|
})();
|
|
|
|
inFlightUploads.set(key, task);
|
|
try {
|
|
await task;
|
|
} finally {
|
|
if (inFlightUploads.get(key) === task) inFlightUploads.delete(key);
|
|
}
|
|
}
|
|
|
|
function queueAgentSessionData(data: AgentSessionData): void {
|
|
const userId = getCurrentAccountKey();
|
|
if (!userId) return;
|
|
|
|
hydratePendingUploads();
|
|
const key = getUploadKey(userId, data);
|
|
pendingUploads.set(key, { userId, data, retryCount: 0 });
|
|
persistPendingUploads();
|
|
scheduleUpload(key, SESSION_UPLOAD_DEBOUNCE_MS);
|
|
}
|
|
|
|
/** Queue a completed local session snapshot without blocking the chat turn. */
|
|
export function queueAgentSessionSync(
|
|
projectId: string,
|
|
sessionId: string,
|
|
messages: readonly RawMessage[],
|
|
): void {
|
|
const data = buildAgentSessionData(projectId, sessionId, messages);
|
|
if (!data) return;
|
|
queueAgentSessionData(data);
|
|
}
|
|
|
|
/** Queue the final product Conversation snapshot after a completed prompt turn. */
|
|
export function queueCodingConversationSessionSync(snapshot: ConversationSnapshot): void {
|
|
const messageNodes = snapshot.nodes.filter((node) => node.kind === 'message');
|
|
let lastUserIndex = -1;
|
|
for (let index = messageNodes.length - 1; index >= 0; index -= 1) {
|
|
if (messageNodes[index].role === 'user') {
|
|
lastUserIndex = index;
|
|
break;
|
|
}
|
|
}
|
|
if (lastUserIndex < 0) return;
|
|
const currentTurnAssistantIds = new Set(messageNodes
|
|
.slice(lastUserIndex + 1)
|
|
.filter((node) => node.role === 'assistant')
|
|
.map((node) => node.id));
|
|
if (currentTurnAssistantIds.size === 0) return;
|
|
|
|
const messages: RawMessage[] = messageNodes.flatMap((node) => {
|
|
return [{
|
|
id: node.id,
|
|
role: node.role,
|
|
content: node.blocks.flatMap((block) => (
|
|
block.kind === 'text' ? [{ type: 'text', text: block.text }] : []
|
|
)),
|
|
isError: node.status === 'error' || node.status === 'aborted',
|
|
} satisfies RawMessage];
|
|
});
|
|
const settledAt = snapshot.run.settledAt;
|
|
const updatedAt = typeof settledAt === 'number' && Number.isFinite(settledAt)
|
|
? new Date(settledAt).toISOString()
|
|
: new Date().toISOString();
|
|
const data = buildAgentSessionData(
|
|
snapshot.conversation.projectId,
|
|
snapshot.conversation.id,
|
|
messages,
|
|
updatedAt,
|
|
);
|
|
if (!data?.messages.some((message) => (
|
|
message.role === 'assistant'
|
|
&& typeof message.id === 'string'
|
|
&& currentTurnAssistantIds.has(message.id)
|
|
))) return;
|
|
queueAgentSessionData(data);
|
|
}
|
|
|
|
/** Flush persisted session snapshots after login or profile bootstrap. */
|
|
export async function flushPendingAgentSessionSync(): Promise<void> {
|
|
hydratePendingUploads();
|
|
const userId = getCurrentAccountKey();
|
|
if (!userId) return;
|
|
const keys = [...pendingUploads.entries()]
|
|
.filter(([, pending]) => pending.userId === userId)
|
|
.map(([key]) => key);
|
|
await Promise.all(keys.map((key) => {
|
|
clearUploadTimer(key);
|
|
return flushUpload(key);
|
|
}));
|
|
}
|
|
|
|
/** Reset only in-memory sync state for isolated unit tests. */
|
|
export function resetAgentSessionSyncForTests(): void {
|
|
for (const timer of uploadTimers.values()) clearTimeout(timer);
|
|
uploadTimers.clear();
|
|
pendingUploads.clear();
|
|
inFlightUploads.clear();
|
|
pendingUploadsHydrated = false;
|
|
}
|