Spaces:

zimejin
/

Job-Scorer

Sleeping

Job-Scorer / lib /agent-chat /session-transcript-store.ts

Improve job discovery strategy and result cards

c708e76 about 2 months ago

11 kB

/**
 * Lightweight durable chat transcript per agent session (Redis, 7d TTL).
 * Stores user/assistant text plus compact tool outcome digests — not full tool payloads.
 */
import { getRedis } from "@/lib/batch-jobs/redis";
import type { ScoreResult } from "@/lib/types";

export type TranscriptMessage = {
  id: string;
  role: "user" | "assistant";
  content: string;
  /** ISO timestamp */
  at: string;
  /** Optional compact digest of tool outcomes for this turn */
  toolDigest?: string;
  jobs?: Array<{
    url: string;
    title: string;
    company?: string;
    source?: string;
    verdict: string;
    scoresSnapshot?: ScoreResult;
    scoredFrom?: string;
  }>;
  searchRunId?: string;
  /** Durable SearchRun to resume from a timeout closer. */
  resumeSearchRunId?: string;
};

export type SessionTranscript = {
  sessionId: string;
  userId: string;
  updatedAt: number;
  messages: TranscriptMessage[];
  activeSearchRunId?: string;
  /** User-set rail title; when absent, title is derived from the first user message. */
  customTitle?: string;
};

export type SessionSummary = {
  sessionId: string;
  title: string;
  preview: string;
  updatedAt: number;
  messageCount: number;
};

const TTL_SEC = 60 * 60 * 24 * 7;
const MAX_MESSAGES = 80;
const MAX_SESSIONS_PER_USER = 40;

function transcriptKey(userId: string, sessionId: string): string {
  return `agent:session-transcript:${userId.trim()}:${sessionId.trim()}`;
}

function indexKey(userId: string): string {
  return `agent:session-index:${userId.trim()}`;
}

function summaryKey(userId: string, sessionId: string): string {
  return `agent:session-summary:${userId.trim()}:${sessionId.trim()}`;
}

const MAX_CUSTOM_TITLE = 80;

export function sanitizeSessionTitle(raw: string): string {
  return raw.trim().replace(/\s+/g, " ").slice(0, MAX_CUSTOM_TITLE);
}

export function deriveSessionTitle(messages: TranscriptMessage[]): string {
  const firstUser = messages.find((m) => m.role === "user" && m.content.trim());
  if (!firstUser) return "New chat";
  const t = firstUser.content.trim().replace(/\s+/g, " ");
  if (t.length <= 48) return t;
  return `${t.slice(0, 47).trimEnd()}…`;
}

export function deriveSessionPreview(messages: TranscriptMessage[]): string {
  for (let i = messages.length - 1; i >= 0; i--) {
    const c = messages[i]?.content?.trim();
    if (!c) continue;
    const t = c.replace(/\s+/g, " ");
    if (t.length <= 72) return t;
    return `${t.slice(0, 71).trimEnd()}…`;
  }
  return "";
}

export function summaryFromTranscript(transcript: SessionTranscript): SessionSummary {
  const custom = transcript.customTitle?.trim();
  return {
    sessionId: transcript.sessionId,
    title: custom || deriveSessionTitle(transcript.messages),
    preview: deriveSessionPreview(transcript.messages),
    updatedAt: transcript.updatedAt,
    messageCount: transcript.messages.length,
  };
}

function parseSummary(raw: unknown): SessionSummary | null {
  try {
    const parsed =
      typeof raw === "string" ? (JSON.parse(raw) as SessionSummary) : (raw as SessionSummary);
    if (!parsed || typeof parsed.sessionId !== "string" || !parsed.sessionId.trim()) return null;
    return {
      sessionId: parsed.sessionId.trim(),
      title: typeof parsed.title === "string" && parsed.title.trim() ? parsed.title : "New chat",
      preview: typeof parsed.preview === "string" ? parsed.preview : "",
      updatedAt: typeof parsed.updatedAt === "number" ? parsed.updatedAt : 0,
      messageCount: typeof parsed.messageCount === "number" ? parsed.messageCount : 0,
    };
  } catch {
    return null;
  }
}

async function indexSessionFromTranscript(transcript: SessionTranscript): Promise<void> {
  const r = getRedis();
  if (!r) return;
  const uid = transcript.userId.trim();
  const sid = transcript.sessionId.trim();
  if (!uid || !sid) return;
  const summary = summaryFromTranscript(transcript);
  try {
    const ik = indexKey(uid);
    await r.zadd(ik, { score: summary.updatedAt, member: sid });
    await r.set(summaryKey(uid, sid), JSON.stringify(summary), { ex: TTL_SEC });
    await r.expire(ik, TTL_SEC);
    const count = await r.zcard(ik);
    if (typeof count === "number" && count > MAX_SESSIONS_PER_USER) {
      const remove = count - MAX_SESSIONS_PER_USER;
      await r.zremrangebyrank(ik, 0, remove - 1);
    }
  } catch (e) {
    console.warn("[session-transcript] index failed", e instanceof Error ? e.message : e);
  }
}

/** Remove a chat transcript and drop it from the user's history index. */
export async function deleteUserSession(userId: string, sessionId: string): Promise<boolean> {
  const r = getRedis();
  const uid = userId.trim();
  const sid = sessionId.trim();
  if (!r || !uid || !sid) return false;
  try {
    await r.del(transcriptKey(uid, sid));
    await r.del(summaryKey(uid, sid));
    await r.zrem(indexKey(uid), sid);
    return true;
  } catch (e) {
    console.warn("[session-transcript] delete failed", e instanceof Error ? e.message : e);
    return false;
  }
}

/** Persist a custom rail title. Empty title clears the override. */
export async function renameUserSession(
  userId: string,
  sessionId: string,
  title: string,
): Promise<SessionSummary | null> {
  const existing = await getSessionTranscript(userId, sessionId);
  if (!existing) return null;
  const next = sanitizeSessionTitle(title);
  if (next) existing.customTitle = next;
  else delete existing.customTitle;
  await saveSessionTranscript(existing, { touch: false });
  return summaryFromTranscript(existing);
}

export async function getSessionTranscript(
  userId: string,
  sessionId: string,
): Promise<SessionTranscript | null> {
  const r = getRedis();
  if (!r) return null;
  const uid = userId.trim();
  const sid = sessionId.trim();
  if (!uid || !sid) return null;
  try {
    const raw = (await r.get(transcriptKey(uid, sid))) as unknown;
    if (raw == null) return null;
    const parsed =
      typeof raw === "string" ? (JSON.parse(raw) as SessionTranscript) : (raw as SessionTranscript);
    if (!parsed || !Array.isArray(parsed.messages)) return null;
    return parsed;
  } catch {
    return null;
  }
}

export async function saveSessionTranscript(
  transcript: SessionTranscript,
  opts?: { touch?: boolean },
): Promise<void> {
  const r = getRedis();
  if (!r) return;
  const uid = transcript.userId.trim();
  const sid = transcript.sessionId.trim();
  if (!uid || !sid) return;
  const trimmed: SessionTranscript = {
    ...transcript,
    userId: uid,
    sessionId: sid,
    updatedAt: opts?.touch === false ? transcript.updatedAt : Date.now(),
    messages: transcript.messages.slice(-MAX_MESSAGES),
  };
  try {
    await r.set(transcriptKey(uid, sid), JSON.stringify(trimmed), { ex: TTL_SEC });
    await indexSessionFromTranscript(trimmed);
  } catch (e) {
    console.warn("[session-transcript] save failed", e instanceof Error ? e.message : e);
  }
}

function mergeTranscriptJobs(
  prev?: TranscriptMessage["jobs"],
  incoming?: TranscriptMessage["jobs"],
): TranscriptMessage["jobs"] | undefined {
  if (!incoming?.length) return prev;
  if (!prev?.length) return incoming;
  const byUrl = new Map(prev.map((j) => [j.url, j]));
  for (const j of incoming) {
    const existing = byUrl.get(j.url);
    if (!existing) {
      byUrl.set(j.url, j);
      continue;
    }
    byUrl.set(j.url, {
      ...existing,
      ...j,
      title: j.title?.trim() || existing.title,
      company: j.company?.trim() || existing.company,
    });
  }
  return [...byUrl.values()];
}

export function mergeTranscriptMessage(
  prev: TranscriptMessage,
  incoming: TranscriptMessage,
): TranscriptMessage {
  const jobs = mergeTranscriptJobs(prev.jobs, incoming.jobs);
  const content = incoming.content.trim() ? incoming.content : prev.content;
  return {
    ...prev,
    ...incoming,
    content,
    toolDigest: incoming.toolDigest ?? prev.toolDigest,
    searchRunId: incoming.searchRunId ?? prev.searchRunId,
    resumeSearchRunId: incoming.resumeSearchRunId ?? prev.resumeSearchRunId,
    ...(jobs?.length ? { jobs } : prev.jobs?.length ? { jobs: prev.jobs } : {}),
  };
}

export async function appendTranscriptMessages(opts: {
  userId: string;
  sessionId: string;
  messages: TranscriptMessage[];
  activeSearchRunId?: string | null;
}): Promise<SessionTranscript | null> {
  const existing =
    (await getSessionTranscript(opts.userId, opts.sessionId)) ??
    ({
      sessionId: opts.sessionId.trim(),
      userId: opts.userId.trim(),
      updatedAt: Date.now(),
      messages: [],
    } satisfies SessionTranscript);

  const byId = new Map(existing.messages.map((m, i) => [m.id, i]));
  for (const m of opts.messages) {
    if (!m.id) continue;
    const idx = byId.get(m.id);
    if (idx === undefined) {
      byId.set(m.id, existing.messages.length);
      existing.messages.push(m);
      continue;
    }
    existing.messages[idx] = mergeTranscriptMessage(existing.messages[idx]!, m);
  }
  if (opts.activeSearchRunId === null) {
    delete existing.activeSearchRunId;
  } else if (typeof opts.activeSearchRunId === "string" && opts.activeSearchRunId.trim()) {
    existing.activeSearchRunId = opts.activeSearchRunId.trim();
  }
  await saveSessionTranscript(existing);
  return existing;
}

export async function listUserSessions(
  userId: string,
  limit = 30,
): Promise<SessionSummary[]> {
  const r = getRedis();
  const uid = userId.trim();
  if (!r || !uid) return [];
  const lim = Math.min(Math.max(1, limit), MAX_SESSIONS_PER_USER);
  try {
    const ids = await r.zrange(indexKey(uid), 0, lim - 1, { rev: true });
    if (!Array.isArray(ids) || ids.length === 0) return [];
    const sessionIds = ids.filter((id): id is string => typeof id === "string" && id.trim().length > 0);
    if (!sessionIds.length) return [];
    const keys = sessionIds.map((id) => summaryKey(uid, id));
    const raws = (await r.mget(...keys)) as unknown[];
    const out: SessionSummary[] = [];
    for (let i = 0; i < sessionIds.length; i++) {
      const parsed = parseSummary(raws[i]);
      if (parsed) {
        out.push(parsed);
        continue;
      }
      const transcript = await getSessionTranscript(uid, sessionIds[i]);
      if (transcript) out.push(summaryFromTranscript(transcript));
    }
    return out;
  } catch (e) {
    console.warn("[session-transcript] list failed", e instanceof Error ? e.message : e);
    return [];
  }
}

export async function createUserSession(userId: string): Promise<SessionSummary | null> {
  const uid = userId.trim();
  if (!uid) return null;
  const sessionId =
    typeof crypto !== "undefined" && "randomUUID" in crypto ?
      crypto.randomUUID()
    : `sess-${Date.now()}-${Math.random().toString(36).slice(2, 9)}`;
  const transcript: SessionTranscript = {
    sessionId,
    userId: uid,
    updatedAt: Date.now(),
    messages: [],
  };
  await saveSessionTranscript(transcript);
  return summaryFromTranscript(transcript);
}

	/**
	* Lightweight durable chat transcript per agent session (Redis, 7d TTL).
	* Stores user/assistant text plus compact tool outcome digests — not full tool payloads.
	*/
	import { getRedis } from "@/lib/batch-jobs/redis";
	import type { ScoreResult } from "@/lib/types";

	export type TranscriptMessage = {
	id: string;
	role: "user" \| "assistant";
	content: string;
	/** ISO timestamp */
	at: string;
	/** Optional compact digest of tool outcomes for this turn */
	toolDigest?: string;
	jobs?: Array<{
	url: string;
	title: string;
	company?: string;
	source?: string;
	verdict: string;
	scoresSnapshot?: ScoreResult;
	scoredFrom?: string;
	}>;
	searchRunId?: string;
	/** Durable SearchRun to resume from a timeout closer. */
	resumeSearchRunId?: string;
	};

	export type SessionTranscript = {
	sessionId: string;
	userId: string;
	updatedAt: number;
	messages: TranscriptMessage[];
	activeSearchRunId?: string;
	/** User-set rail title; when absent, title is derived from the first user message. */
	customTitle?: string;
	};

	export type SessionSummary = {
	sessionId: string;
	title: string;
	preview: string;
	updatedAt: number;
	messageCount: number;
	};

	const TTL_SEC = 60 * 60 * 24 * 7;
	const MAX_MESSAGES = 80;
	const MAX_SESSIONS_PER_USER = 40;

	function transcriptKey(userId: string, sessionId: string): string {
	return `agent:session-transcript:${userId.trim()}:${sessionId.trim()}`;
	}

	function indexKey(userId: string): string {
	return `agent:session-index:${userId.trim()}`;
	}

	function summaryKey(userId: string, sessionId: string): string {
	return `agent:session-summary:${userId.trim()}:${sessionId.trim()}`;
	}

	const MAX_CUSTOM_TITLE = 80;

	export function sanitizeSessionTitle(raw: string): string {
	return raw.trim().replace(/\s+/g, " ").slice(0, MAX_CUSTOM_TITLE);
	}

	export function deriveSessionTitle(messages: TranscriptMessage[]): string {
	const firstUser = messages.find((m) => m.role === "user" && m.content.trim());
	if (!firstUser) return "New chat";
	const t = firstUser.content.trim().replace(/\s+/g, " ");
	if (t.length <= 48) return t;
	return `${t.slice(0, 47).trimEnd()}…`;
	}

	export function deriveSessionPreview(messages: TranscriptMessage[]): string {
	for (let i = messages.length - 1; i >= 0; i--) {
	const c = messages[i]?.content?.trim();
	if (!c) continue;
	const t = c.replace(/\s+/g, " ");
	if (t.length <= 72) return t;
	return `${t.slice(0, 71).trimEnd()}…`;
	}
	return "";
	}

	export function summaryFromTranscript(transcript: SessionTranscript): SessionSummary {
	const custom = transcript.customTitle?.trim();
	return {
	sessionId: transcript.sessionId,
	title: custom \|\| deriveSessionTitle(transcript.messages),
	preview: deriveSessionPreview(transcript.messages),
	updatedAt: transcript.updatedAt,
	messageCount: transcript.messages.length,
	};
	}

	function parseSummary(raw: unknown): SessionSummary \| null {
	try {
	const parsed =
	typeof raw === "string" ? (JSON.parse(raw) as SessionSummary) : (raw as SessionSummary);
	if (!parsed \|\| typeof parsed.sessionId !== "string" \|\| !parsed.sessionId.trim()) return null;
	return {
	sessionId: parsed.sessionId.trim(),
	title: typeof parsed.title === "string" && parsed.title.trim() ? parsed.title : "New chat",
	preview: typeof parsed.preview === "string" ? parsed.preview : "",
	updatedAt: typeof parsed.updatedAt === "number" ? parsed.updatedAt : 0,
	messageCount: typeof parsed.messageCount === "number" ? parsed.messageCount : 0,
	};
	} catch {
	return null;
	}
	}

	async function indexSessionFromTranscript(transcript: SessionTranscript): Promise<void> {
	const r = getRedis();
	if (!r) return;
	const uid = transcript.userId.trim();
	const sid = transcript.sessionId.trim();
	if (!uid \|\| !sid) return;
	const summary = summaryFromTranscript(transcript);
	try {
	const ik = indexKey(uid);
	await r.zadd(ik, { score: summary.updatedAt, member: sid });
	await r.set(summaryKey(uid, sid), JSON.stringify(summary), { ex: TTL_SEC });
	await r.expire(ik, TTL_SEC);
	const count = await r.zcard(ik);
	if (typeof count === "number" && count > MAX_SESSIONS_PER_USER) {
	const remove = count - MAX_SESSIONS_PER_USER;
	await r.zremrangebyrank(ik, 0, remove - 1);
	}
	} catch (e) {
	console.warn("[session-transcript] index failed", e instanceof Error ? e.message : e);
	}
	}

	/** Remove a chat transcript and drop it from the user's history index. */
	export async function deleteUserSession(userId: string, sessionId: string): Promise<boolean> {
	const r = getRedis();
	const uid = userId.trim();
	const sid = sessionId.trim();
	if (!r \|\| !uid \|\| !sid) return false;
	try {
	await r.del(transcriptKey(uid, sid));
	await r.del(summaryKey(uid, sid));
	await r.zrem(indexKey(uid), sid);
	return true;
	} catch (e) {
	console.warn("[session-transcript] delete failed", e instanceof Error ? e.message : e);
	return false;
	}
	}

	/** Persist a custom rail title. Empty title clears the override. */
	export async function renameUserSession(
	userId: string,
	sessionId: string,
	title: string,
	): Promise<SessionSummary \| null> {
	const existing = await getSessionTranscript(userId, sessionId);
	if (!existing) return null;
	const next = sanitizeSessionTitle(title);
	if (next) existing.customTitle = next;
	else delete existing.customTitle;
	await saveSessionTranscript(existing, { touch: false });
	return summaryFromTranscript(existing);
	}

	export async function getSessionTranscript(
	userId: string,
	sessionId: string,
	): Promise<SessionTranscript \| null> {
	const r = getRedis();
	if (!r) return null;
	const uid = userId.trim();
	const sid = sessionId.trim();
	if (!uid \|\| !sid) return null;
	try {
	const raw = (await r.get(transcriptKey(uid, sid))) as unknown;
	if (raw == null) return null;
	const parsed =
	typeof raw === "string" ? (JSON.parse(raw) as SessionTranscript) : (raw as SessionTranscript);
	if (!parsed \|\| !Array.isArray(parsed.messages)) return null;
	return parsed;
	} catch {
	return null;
	}
	}

	export async function saveSessionTranscript(
	transcript: SessionTranscript,
	opts?: { touch?: boolean },
	): Promise<void> {
	const r = getRedis();
	if (!r) return;
	const uid = transcript.userId.trim();
	const sid = transcript.sessionId.trim();
	if (!uid \|\| !sid) return;
	const trimmed: SessionTranscript = {
	...transcript,
	userId: uid,
	sessionId: sid,
	updatedAt: opts?.touch === false ? transcript.updatedAt : Date.now(),
	messages: transcript.messages.slice(-MAX_MESSAGES),
	};
	try {
	await r.set(transcriptKey(uid, sid), JSON.stringify(trimmed), { ex: TTL_SEC });
	await indexSessionFromTranscript(trimmed);
	} catch (e) {
	console.warn("[session-transcript] save failed", e instanceof Error ? e.message : e);
	}
	}

	function mergeTranscriptJobs(
	prev?: TranscriptMessage["jobs"],
	incoming?: TranscriptMessage["jobs"],
	): TranscriptMessage["jobs"] \| undefined {
	if (!incoming?.length) return prev;
	if (!prev?.length) return incoming;
	const byUrl = new Map(prev.map((j) => [j.url, j]));
	for (const j of incoming) {
	const existing = byUrl.get(j.url);
	if (!existing) {
	byUrl.set(j.url, j);
	continue;
	}
	byUrl.set(j.url, {
	...existing,
	...j,
	title: j.title?.trim() \|\| existing.title,
	company: j.company?.trim() \|\| existing.company,
	});
	}
	return [...byUrl.values()];
	}

	export function mergeTranscriptMessage(
	prev: TranscriptMessage,
	incoming: TranscriptMessage,
	): TranscriptMessage {
	const jobs = mergeTranscriptJobs(prev.jobs, incoming.jobs);
	const content = incoming.content.trim() ? incoming.content : prev.content;
	return {
	...prev,
	...incoming,
	content,
	toolDigest: incoming.toolDigest ?? prev.toolDigest,
	searchRunId: incoming.searchRunId ?? prev.searchRunId,
	resumeSearchRunId: incoming.resumeSearchRunId ?? prev.resumeSearchRunId,
	...(jobs?.length ? { jobs } : prev.jobs?.length ? { jobs: prev.jobs } : {}),
	};
	}

	export async function appendTranscriptMessages(opts: {
	userId: string;
	sessionId: string;
	messages: TranscriptMessage[];
	activeSearchRunId?: string \| null;
	}): Promise<SessionTranscript \| null> {
	const existing =
	(await getSessionTranscript(opts.userId, opts.sessionId)) ??
	({
	sessionId: opts.sessionId.trim(),
	userId: opts.userId.trim(),
	updatedAt: Date.now(),
	messages: [],
	} satisfies SessionTranscript);

	const byId = new Map(existing.messages.map((m, i) => [m.id, i]));
	for (const m of opts.messages) {
	if (!m.id) continue;
	const idx = byId.get(m.id);
	if (idx === undefined) {
	byId.set(m.id, existing.messages.length);
	existing.messages.push(m);
	continue;
	}
	existing.messages[idx] = mergeTranscriptMessage(existing.messages[idx]!, m);
	}
	if (opts.activeSearchRunId === null) {
	delete existing.activeSearchRunId;
	} else if (typeof opts.activeSearchRunId === "string" && opts.activeSearchRunId.trim()) {
	existing.activeSearchRunId = opts.activeSearchRunId.trim();
	}
	await saveSessionTranscript(existing);
	return existing;
	}

	export async function listUserSessions(
	userId: string,
	limit = 30,
	): Promise<SessionSummary[]> {
	const r = getRedis();
	const uid = userId.trim();
	if (!r \|\| !uid) return [];
	const lim = Math.min(Math.max(1, limit), MAX_SESSIONS_PER_USER);
	try {
	const ids = await r.zrange(indexKey(uid), 0, lim - 1, { rev: true });
	if (!Array.isArray(ids) \|\| ids.length === 0) return [];
	const sessionIds = ids.filter((id): id is string => typeof id === "string" && id.trim().length > 0);
	if (!sessionIds.length) return [];
	const keys = sessionIds.map((id) => summaryKey(uid, id));
	const raws = (await r.mget(...keys)) as unknown[];
	const out: SessionSummary[] = [];
	for (let i = 0; i < sessionIds.length; i++) {
	const parsed = parseSummary(raws[i]);
	if (parsed) {
	out.push(parsed);
	continue;
	}
	const transcript = await getSessionTranscript(uid, sessionIds[i]);
	if (transcript) out.push(summaryFromTranscript(transcript));
	}
	return out;
	} catch (e) {
	console.warn("[session-transcript] list failed", e instanceof Error ? e.message : e);
	return [];
	}
	}

	export async function createUserSession(userId: string): Promise<SessionSummary \| null> {
	const uid = userId.trim();
	if (!uid) return null;
	const sessionId =
	typeof crypto !== "undefined" && "randomUUID" in crypto ?
	crypto.randomUUID()
	: `sess-${Date.now()}-${Math.random().toString(36).slice(2, 9)}`;
	const transcript: SessionTranscript = {
	sessionId,
	userId: uid,
	updatedAt: Date.now(),
	messages: [],
	};
	await saveSessionTranscript(transcript);
	return summaryFromTranscript(transcript);
	}