We're replacing the current "flush after 10 messages or 8h stale" journal system with session-based summaries. A "session" = a window of chat activity. When the user goes silent for 2+ hours, the session closes, a summary blob is generated, and it appears in the journal with a time range timestamp (e.g. "4:00 PM – 6:30 PM"). Multiple sessions per day = multiple blobs. Users can edit blobs after generation.
backend/src/jobs/flush.ts— groups unflushed USER messages by calendar date, concatenates as[HH:MM] messagelines, upserts intojournal_entries(one row per user per day), then fires summary generation + mood extraction + supermemory ingestbackend/src/lib/summarize.ts— calls GPT-4o to rewrite the day's transcript into a first-person narrativebackend/src/lib/openai-extract.ts— extracts mood, energy, themes, moodReason from textbackend/src/lib/crypto.ts— encryptForUser / decryptForUser for at-rest encryption with per-user DEKsbackend/src/lib/supermemory.ts— addDocument / deleteSmDocument / fireAndLog for semantic memory ingestbackend/src/lib/time.ts— toLocalHHMM, toLocalDateString, toLocalDateUTC for timezone-aware formattingbackend/src/lib/usage.ts— recordUsage for tracking LLM token costsbackend/src/lib/analytics.ts— track() for event analyticsbackend/src/lib/style.ts— extractStyleSignals for writing style profiling (refreshed on flush)backend/src/lib/pi-runtime.ts— evictUserSessions for cache invalidationbackend/src/api/me-routes.ts— /api/me/journal and /api/me/journal/:date endpointsbackend/src/api/helpers.ts— sendJson, sendError, parseJsonBodyOrErrorbackend/src/lib/auth.ts— requireJwt for endpoint authbackend/prisma/schema.prisma— JournalEntry model (one entry per user per day), ChatMessage model, User model
File: backend/prisma/schema.prisma
model ChatSession {
id String @id @default(uuid())
userId String
user User @relation(fields: [userId], references: [id], onDelete: Cascade)
// when the first message in this session was sent
startedAt DateTime
// when the last message in this session was sent (updated on each new message)
endedAt DateTime
// null while session is open; set when the session closes and summary generates
closedAt DateTime?
// the generated first-person summary blob
// encrypted at rest with user's DEK (same as JournalEntry.summary)
summary String?
summaryLength Int?
// AI-extracted metadata (same fields as JournalEntry)
mood MoodLevel?
energy EnergyLevel?
moodReason String? // encrypted
moodReasonLength Int?
themes String[]
// which JournalEntry this session was rolled into (for the daily view)
journalEntryId String?
journalEntry JournalEntry? @relation(fields: [journalEntryId], references: [id])
// supermemory tracking (same pattern as JournalEntry)
smDocumentId String?
smStatus SmStatus @default(PENDING)
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
messages ChatMessage[]
@@index([userId, closedAt])
@@index([userId, endedAt(sort: Desc)])
@@map("chat_sessions")
}chatSessionId String?
chatSession ChatSession? @relation(fields: [chatSessionId], references: [id])sessions ChatSession[]chatSessions ChatSession[]npx prisma migrate dev --name add-chat-sessionsFile: backend/src/lib/sessions.ts — NEW FILE
// session-based journal grouping. replaces the old 10-message / 8h-stale flush
// model with activity-window detection. each USER message is assigned to an open
// session; when the user goes silent for SESSION_GAP_MS, the session closes and
// a summary is generated.
//
// two entry points:
// - assignMessageToSession(userId, messageId, createdAt): called on every USER message
// - closeStaleSessionsCron(): 15-min cron, finds and closes expired open sessions
//
// concurrency: per-user in-memory lock set (same pattern as flush.ts flushesInFlight)
import { db } from "./db.js";
import { log } from "./log.js";
import { toLocalHHMM, toLocalDateString, toLocalDateUTC } from "./time.js";
import { extractMetadata } from "./openai-extract.js";
import { generateSessionSummary } from "./summarize.js";
import { addDocument, fireAndLog } from "./supermemory.js";
import { recordUsage } from "./usage.js";
import { track } from "./analytics.js";
import { decryptForUser, encryptForUser } from "./crypto.js";
// how long a user must be silent before their session auto-closes
export const SESSION_GAP_MS = 2 * 60 * 60 * 1000; // 2 hours
// per-user in-memory lock. prevents two closeSession calls racing for the same session.
const closesInFlight = new Set<string>();/**
* Called on every USER message after it's persisted to chat_messages.
* Assigns the message to an existing open session or creates a new one.
* If the gap since the last message exceeds SESSION_GAP_MS, the old
* session is closed (fire-and-forget) and a new one is created.
*
* Returns the ChatSession id the message was assigned to.
*/
export async function assignMessageToSession(
userId: string,
messageId: string,
messageCreatedAt: Date,
): Promise<string> {
// Find the most recent open session for this user
const openSession = await db.chatSession.findFirst({
where: { userId, closedAt: null },
orderBy: { endedAt: "desc" },
});
if (openSession) {
const gapMs = messageCreatedAt.getTime() - openSession.endedAt.getTime();
if (gapMs < SESSION_GAP_MS) {
// Same session — extend it
await db.$transaction([
db.chatSession.update({
where: { id: openSession.id },
data: { endedAt: messageCreatedAt },
}),
db.chatMessage.update({
where: { id: messageId },
data: { chatSessionId: openSession.id },
}),
]);
return openSession.id;
}
// Gap exceeded — close the old session (fire-and-forget, don't block chat)
setImmediate(() => {
closeSession(openSession.id).catch((err) => {
log.error("closeSession failed (from assignMessage)", {
sessionId: openSession.id,
err: err instanceof Error ? err.message : String(err),
});
});
});
}
// Create a new session
const newSession = await db.chatSession.create({
data: {
userId,
startedAt: messageCreatedAt,
endedAt: messageCreatedAt,
},
});
await db.chatMessage.update({
where: { id: messageId },
data: { chatSessionId: newSession.id },
});
log.info("session: new session created", { userId, sessionId: newSession.id });
return newSession.id;
}/**
* Close a session: mark closedAt, generate summary, extract mood/themes,
* roll into the day's JournalEntry, ingest to supermemory.
*/
export async function closeSession(sessionId: string): Promise<void> {
if (closesInFlight.has(sessionId)) {
log.debug("session: close already in flight, skipping", { sessionId });
return;
}
closesInFlight.add(sessionId);
try {
const session = await db.chatSession.findUnique({
where: { id: sessionId },
include: {
user: { select: { id: true, timezone: true, displayName: true } },
messages: {
where: { role: "USER" },
orderBy: { createdAt: "asc" },
select: { id: true, content: true, createdAt: true },
},
},
});
if (!session || session.closedAt) return; // already closed or not found
if (session.messages.length === 0) {
await db.chatSession.update({
where: { id: sessionId },
data: { closedAt: new Date() },
});
return;
}
const userId = session.userId;
const timezone = session.user.timezone;
// Mark closed immediately so concurrent checks don't re-process
await db.chatSession.update({
where: { id: sessionId },
data: { closedAt: new Date() },
});
// Decrypt messages and build transcript
const formattedLines: string[] = [];
for (const m of session.messages) {
const plain = await decryptForUser(userId, m.content);
formattedLines.push(`[${toLocalHHMM(m.createdAt, timezone)}] ${plain.trim()}`);
}
const transcript = formattedLines.join("\n");
// Resolve display name
const displayNamePlain = session.user.displayName
? await decryptForUser(userId, session.user.displayName)
: "the user";
const startTime = toLocalHHMM(session.startedAt, timezone);
const endTime = toLocalHHMM(session.endedAt, timezone);
const dateStr = toLocalDateString(session.startedAt, timezone);
// Generate session summary
let summary: string | null = null;
try {
const summaryRes = await generateSessionSummary({
displayName: displayNamePlain,
dateStr,
startTime,
endTime,
transcript,
});
summary = summaryRes.summary;
if (summaryRes.usage) {
recordUsage({ userId, kind: "extract", usage: summaryRes.usage });
}
} catch (err) {
log.warn("session: summary generation failed (non-fatal)", {
sessionId,
err: err instanceof Error ? err.message : String(err),
});
}
// Extract mood/themes
let metadata = { mood: null, energy: null, moodReason: null, themes: [] } as {
mood: string | null;
energy: string | null;
moodReason: string | null;
themes: string[];
};
try {
const extractRes = await extractMetadata(transcript);
metadata = extractRes.metadata;
if (extractRes.usage) {
recordUsage({ userId, kind: "extract", usage: extractRes.usage });
}
} catch (err) {
log.warn("session: metadata extraction failed (non-fatal)", {
sessionId,
err: err instanceof Error ? err.message : String(err),
});
}
// Encrypt sensitive fields
const summaryCt = summary ? await encryptForUser(userId, summary) : null;
const moodReasonCt = metadata.moodReason
? await encryptForUser(userId, metadata.moodReason)
: null;
const transcriptCt = await encryptForUser(userId, transcript);
// Upsert into daily JournalEntry (same pattern as flush.ts flushGroup)
const entryDate = toLocalDateUTC(session.startedAt, timezone);
const existing = await db.journalEntry.findUnique({
where: { userId_entryDate: { userId, entryDate } },
});
const existingPlain = existing
? await decryptForUser(userId, existing.content)
: "";
const fullContent = existing
? `${existingPlain}\n${transcript}`
: transcript;
const fullContentCt = await encryptForUser(userId, fullContent);
const entry = await db.journalEntry.upsert({
where: { userId_entryDate: { userId, entryDate } },
create: {
userId,
content: fullContentCt,
contentLength: fullContent.length,
entryDate,
themes: metadata.themes,
mood: metadata.mood as any,
energy: metadata.energy as any,
smStatus: "PENDING",
},
update: {
content: fullContentCt,
contentLength: fullContent.length,
smStatus: "PENDING",
},
});
// Update session with summary + metadata + journal link
await db.chatSession.update({
where: { id: sessionId },
data: {
summary: summaryCt,
summaryLength: summary ? summary.length : null,
mood: metadata.mood as any,
energy: metadata.energy as any,
moodReason: moodReasonCt,
moodReasonLength: metadata.moodReason ? metadata.moodReason.length : null,
themes: metadata.themes,
journalEntryId: entry.id,
},
});
// Mark all USER messages in the session as flushed
const messageIds = session.messages.map((m) => m.id);
await db.chatMessage.updateMany({
where: { id: { in: messageIds } },
data: { flushed: true },
});
log.info("session: closed and summarized", {
userId,
sessionId,
messageCount: session.messages.length,
hasSummary: !!summary,
mood: metadata.mood,
themes: metadata.themes,
});
// Analytics
track("session_closed", {
message_count: session.messages.length,
duration_minutes: Math.round(
(session.endedAt.getTime() - session.startedAt.getTime()) / 60000,
),
has_summary: !!summary,
mood: metadata.mood,
themes_count: metadata.themes.length,
}, userId);
// Fire-and-forget: supermemory ingest
fireAndLog(
async () => {
const smRes = await addDocument({
userId,
content: fullContent,
customId: `session_${sessionId}`,
type: "chat_session",
metadata: {
sessionId,
date: dateStr,
startTime,
endTime,
mood: metadata.mood,
energy: metadata.energy,
themes: metadata.themes,
},
});
await db.chatSession.update({
where: { id: sessionId },
data: { smDocumentId: smRes.id, smStatus: "DONE" },
});
// Also update the journal entry's SM doc
const journalSmRes = await addDocument({
userId,
content: fullContent,
customId: `journal_entry_${entry.id}`,
type: "journal_entry",
metadata: {
date: dateStr,
mood: metadata.mood,
energy: metadata.energy,
themes: metadata.themes,
},
});
await db.journalEntry.update({
where: { id: entry.id },
data: { smDocumentId: journalSmRes.id, smStatus: "DONE" },
});
},
{ op: "session_ingest", userId, sessionId },
);
} catch (err) {
log.error("session: closeSession failed", {
sessionId,
err: err instanceof Error ? err.message : String(err),
});
} finally {
closesInFlight.delete(sessionId);
}
}/**
* Cron job: find all open sessions where endedAt is older than the gap
* threshold and close them. Run every 15 minutes.
*/
export async function closeStaleSessionsCron(): Promise<void> {
const staleCutoff = new Date(Date.now() - SESSION_GAP_MS);
const staleSessions = await db.chatSession.findMany({
where: {
closedAt: null,
endedAt: { lt: staleCutoff },
},
select: { id: true, userId: true },
});
log.info("closeStaleSessionsCron: scanning", { count: staleSessions.length });
for (const s of staleSessions) {
try {
await closeSession(s.id);
} catch (err) {
log.error("closeStaleSessionsCron: session failed", {
sessionId: s.id,
err: err instanceof Error ? err.message : String(err),
});
}
}
}File: backend/src/lib/summarize.ts
Add this alongside the existing generateJournalSummary function. Same structure, same MODEL, same error handling pattern.
Rewrite a chat session transcript as a first-person journal entry, written in the user's own voice. Use "I" statements. Preserve their actual words and phrasing where possible, don't over-literary or clean up their grammar. Keep it honest, not preachy or over-reflective.
Capture the ARC of the session — how the conversation started, what shifted, where it landed. If multiple topics came up, show how they connected (they usually do emotionally, even when they seem unrelated).
Pull 1-2 direct quotes from the transcript that capture the most raw/honest moments. Wrap them naturally into the prose.
Target 80-200 words depending on session length. A 5-minute session might only need 2 sentences. A 2-hour session might need a full paragraph.
Plain prose, no bullet points, no headers, no timestamps (the frontend renders the time range separately). Never mention an AI assistant, chatbot, Maya, Sage, or Luna — write as if this is the user's own journal.
NAME: {displayName}
DATE: {dateStr}
SESSION: {startTime} – {endTime}
CHAT TRANSCRIPT:
{transcript}
First-person summary:
export interface GenerateSessionSummaryInput {
displayName: string;
dateStr: string; // YYYY-MM-DD in user's local timezone
startTime: string; // HH:MM formatted in user's timezone
endTime: string; // HH:MM formatted in user's timezone
transcript: string; // "[HH:MM] message" lines
}
export async function generateSessionSummary(
opts: GenerateSessionSummaryInput,
): Promise<SummaryResult>Parameters: temperature 0.3, max_tokens 400. Apply stripEmDashes to the result. Build usage via buildExtractUsage. Same error handling as generateJournalSummary.
File: backend/src/index.ts
Add a new 15-minute cron job:
import { closeStaleSessionsCron } from "./lib/sessions.js";
// every 15 minutes: close stale chat sessions and generate summaries
cron.schedule("*/15 * * * *", () => {
closeStaleSessionsCron().catch((err) =>
log.error("stale session cron failed", {
err: err instanceof Error ? err.message : String(err),
}),
);
});Keep the existing hourly flushStaleBuffers cron running during migration. It will naturally stop finding work once all new messages flow through sessions.
File: wherever USER messages are persisted (look for unflushedCount increment and flushUserBuffer call — likely backend/src/api/chat.ts or backend/src/bot/handlers.ts)
After the USER message is saved to chat_messages:
import { assignMessageToSession } from "../lib/sessions.js";
// After message is persisted:
// message = the just-created ChatMessage row
assignMessageToSession(userId, message.id, message.createdAt).catch((err) => {
log.error("session assignment failed", {
userId,
messageId: message.id,
err: err instanceof Error ? err.message : String(err),
});
});This should NOT block the chat response. assignMessageToSession is fast (one read + one write). If it triggers closeSession on an old session, that runs fire-and-forget via setImmediate.
Keep the existing unflushedCount + flushUserBuffer logic for now — both systems run in parallel during migration.
File: backend/src/api/me-routes.ts
Returns all closed session blobs for a specific date, with formatted time ranges.
export async function handleMeJournalDaySessions(
req: IncomingMessage,
res: ServerResponse,
date: string,
): Promise<void> {
const auth = requireJwt(req, res);
if (!auth) return;
if (!/^\d{4}-\d{2}-\d{2}$/.test(date)) {
sendError(res, 400, "date must be YYYY-MM-DD");
return;
}
const user = await db.user.findUnique({
where: { id: auth.userId },
select: { timezone: true },
});
if (!user) {
sendError(res, 404, "user not found");
return;
}
const dayStart = toLocalDateUTC(new Date(`${date}T12:00:00.000Z`), user.timezone);
const dayEnd = new Date(dayStart.getTime() + 24 * 60 * 60 * 1000);
const sessions = await db.chatSession.findMany({
where: {
userId: auth.userId,
closedAt: { not: null },
startedAt: { gte: dayStart, lt: dayEnd },
},
orderBy: { startedAt: "asc" },
select: {
id: true,
startedAt: true,
endedAt: true,
summary: true,
summaryLength: true,
mood: true,
energy: true,
themes: true,
moodReason: true,
},
});
const decrypted = await Promise.all(
sessions.map(async (s) => ({
id: s.id,
startedAt: s.startedAt.toISOString(),
endedAt: s.endedAt.toISOString(),
startTime: s.startedAt.toLocaleTimeString("en-US", {
hour: "numeric",
minute: "2-digit",
timeZone: user.timezone,
}),
endTime: s.endedAt.toLocaleTimeString("en-US", {
hour: "numeric",
minute: "2-digit",
timeZone: user.timezone,
}),
summary: s.summary
? await decryptForUser(auth.userId, s.summary)
: null,
mood: s.mood,
energy: s.energy,
themes: s.themes,
moodReason: s.moodReason
? await decryptForUser(auth.userId, s.moodReason)
: null,
})),
);
sendJson(res, 200, { sessions: decrypted });
}In the matchMeRoute function, add handling for /api/me/journal/:date/sessions:
// Inside matchMeRoute, in the journal section:
if (parts[2] === "journal") {
if (parts.length === 3) return { kind: "journal" };
if (parts.length === 4 && parts[3]) return { kind: "journalDay", date: parts[3] };
// NEW: /api/me/journal/:date/sessions
if (parts.length === 5 && parts[3] && parts[4] === "sessions") {
return { kind: "journalDaySessions", date: parts[3] };
}
}Add the corresponding handler dispatch in server.ts or wherever routes are dispatched.
In the existing /api/me/journal handler, after fetching journal entries, also fetch sessions per entry and include them in the response:
// After building the entries array, for each entry fetch its sessions:
const entriesWithSessions = await Promise.all(
entries.map(async (entry) => {
const sessions = await db.chatSession.findMany({
where: {
journalEntryId: entry.id,
closedAt: { not: null },
},
orderBy: { startedAt: "asc" },
select: {
id: true,
startedAt: true,
endedAt: true,
summary: true,
mood: true,
energy: true,
themes: true,
moodReason: true,
},
});
const decryptedSessions = await Promise.all(
sessions.map(async (s) => ({
id: s.id,
startedAt: s.startedAt.toISOString(),
endedAt: s.endedAt.toISOString(),
startTime: s.startedAt.toLocaleTimeString("en-US", {
hour: "numeric",
minute: "2-digit",
timeZone: user.timezone,
}),
endTime: s.endedAt.toLocaleTimeString("en-US", {
hour: "numeric",
minute: "2-digit",
timeZone: user.timezone,
}),
summary: s.summary
? await decryptForUser(auth.userId, s.summary)
: null,
mood: s.mood,
energy: s.energy,
themes: s.themes,
moodReason: s.moodReason
? await decryptForUser(auth.userId, s.moodReason)
: null,
})),
);
return {
...entry, // existing entry fields
sessions: decryptedSessions,
};
}),
);Old entries with no sessions will have sessions: [].
| File | Action |
|---|---|
backend/prisma/schema.prisma |
Add ChatSession model, add chatSessionId to ChatMessage, add sessions to JournalEntry and User |
backend/src/lib/sessions.ts |
NEW — assignMessageToSession, closeSession, closeStaleSessionsCron |
backend/src/lib/summarize.ts |
Add generateSessionSummary with session-specific prompt |
backend/src/index.ts |
Add 15-min cron for closeStaleSessionsCron |
backend/src/api/chat.ts or backend/src/bot/handlers.ts |
Wire assignMessageToSession after USER message persist |
backend/src/api/me-routes.ts |
Add handleMeJournalDaySessions, update matchMeRoute, update handleMeJournal to include sessions |
backend/src/api/server.ts |
Add route dispatch for /api/me/journal/:date/sessions |
| Parameter | Default | Location | Notes |
|---|---|---|---|
SESSION_GAP_MS |
2 hours (7,200,000ms) | sessions.ts |
How long silence = session end |
| Cron interval | 15 min | index.ts |
How often to check for stale sessions |
| Summary max_tokens | 400 | summarize.ts |
Scales with session length |
| Summary temperature | 0.3 | summarize.ts |
Low for consistency |
- ALL text that touches the DB must go through
encryptForUser/decryptForUser - Use
toLocalHHMM,toLocalDateString,toLocalDateUTCfromlib/time.tsfor timezone handling - Use the same
fireAndLogpattern asflush.tsfor fire-and-forget async operations - Use
recordUsagefor ALL LLM calls (summary + metadata extraction) - Use
track()for analytics events - The in-memory
closesInFlightset prevents concurrent closes of the same session assignMessageToSessionmust NOT block the chat response- Keep existing flush system running in parallel during migration