ilia/Jobber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rename code

Browse Source
This commit is contained in:
DaKheera47 2026-02-15 18:39:57 +00:00
parent 672eb3d2b9
commit f9abd6ff74
21 changed files with 220 additions and 2856 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
orchestrator/src/client/components/ghostwriter/Composer.tsx
View File

@ -1,5 +1,4 @@
import { getMetaShortcutLabel, isMetaKeyPressed } from "@client/lib/meta-key";
import { RefreshCcw, Send, Square } from "lucide-react";
import { Send } from "lucide-react";
import type React from "react";
import { useState } from "react";
import { Button } from "@/components/ui/button";
@ -7,21 +6,10 @@ import { Textarea } from "@/components/ui/textarea";
type ComposerProps = {
disabled?: boolean;
isStreaming: boolean;
canRegenerate: boolean;
onRegenerate: () => Promise<void>;
onStop: () => Promise<void>;
onSend: (content: string) => Promise<void>;
};
export const Composer: React.FC<ComposerProps> = ({
disabled,
isStreaming,
canRegenerate,
onRegenerate,
onStop,
onSend,
}) => {
export const Composer: React.FC<ComposerProps> = ({ disabled, onSend }) => {
const [value, setValue] = useState("");
const submit = async () => {
@ -39,7 +27,7 @@ export const Composer: React.FC<ComposerProps> = ({
onChange={(event) => setValue(event.target.value)}
disabled={disabled}
onKeyDown={(event) => {
if (isMetaKeyPressed(event) && event.key === "Enter") {
if ((event.metaKey || event.ctrlKey) && event.key === "Enter") {
event.preventDefault();
void submit();
}
@ -48,42 +36,16 @@ export const Composer: React.FC<ComposerProps> = ({
/>
<div className="flex items-center justify-between">
<div className="text-[10px] text-muted-foreground">
{getMetaShortcutLabel("Enter")} to send
</div>
<div className="flex items-center gap-1">
{isStreaming ? (
<Button
size="icon"
variant="outline"
onClick={() => void onStop()}
aria-label="Stop generating"
title="Stop generating"
>
<Square className="h-3.5 w-3.5" />
</Button>
) : (
<Button
size="icon"
variant="outline"
onClick={() => void onRegenerate()}
disabled={disabled || !canRegenerate}
aria-label="Regenerate response"
title="Regenerate response"
>
<RefreshCcw className="h-3.5 w-3.5" />
</Button>
)}
<Button
size="icon"
onClick={() => void submit()}
disabled={disabled || !value.trim()}
aria-label="Send message"
title="Send message"
>
<Send className="h-3.5 w-3.5" />
</Button>
Cmd/Ctrl+Enter to send
</div>
<Button
size="sm"
onClick={() => void submit()}
disabled={disabled || !value.trim()}
>
<Send className="mr-1 h-3.5 w-3.5" />
Send
</Button>
</div>
</div>
);

177
orchestrator/src/client/components/ghostwriter/GhostwriterPanel.tsx
View File

@ -1,16 +1,27 @@
import type { Job, JobChatMessage, JobChatStreamEvent } from "@shared/types";
import type {
Job,
JobChatMessage,
JobChatStreamEvent,
JobChatThread,
} from "@shared/types";
import { MessageSquare } from "lucide-react";
import type React from "react";
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import { toast } from "sonner";
import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
import * as api from "../../api";
import { Composer } from "./Composer";
import { MessageList } from "./MessageList";
import { RunControls } from "./RunControls";
import { ThreadList } from "./ThreadList";
type GhostwriterPanelProps = {
job: Job;
};
export const GhostwriterPanel: React.FC<GhostwriterPanelProps> = ({ job }) => {
const [threads, setThreads] = useState<JobChatThread[]>([]);
const [activeThreadId, setActiveThreadId] = useState<string | null>(null);
const [messages, setMessages] = useState<JobChatMessage[]>([]);
const [isLoading, setIsLoading] = useState(true);
const [isStreaming, setIsStreaming] = useState(false);
@ -18,13 +29,16 @@ export const GhostwriterPanel: React.FC<GhostwriterPanelProps> = ({ job }) => {
null,
);
const [activeRunId, setActiveRunId] = useState<string | null>(null);
const messageListRef = useRef<HTMLDivElement | null>(null);
const streamAbortRef = useRef<AbortController | null>(null);
const messageCount = messages.length;
const activeThreadIdRef = useRef<string | null>(null);
useEffect(() => {
activeThreadIdRef.current = activeThreadId;
}, [activeThreadId]);
useEffect(() => {
if (messageCount === 0) return;
const container = messageListRef.current;
if (!container) return;
const distanceToBottom =
@ -32,19 +46,45 @@ export const GhostwriterPanel: React.FC<GhostwriterPanelProps> = ({ job }) => {
if (distanceToBottom < 120 || isStreaming) {
container.scrollTop = container.scrollHeight;
}
}, [messageCount, isStreaming]);
});
const loadMessages = useCallback(async () => {
const data = await api.listJobGhostwriterMessages(job.id, {
limit: 300,
const loadThreadMessages = useCallback(
async (threadId: string) => {
const data = await api.listJobChatMessages(job.id, threadId, {
limit: 300,
});
setMessages(data.messages);
},
[job.id],
);
const createThread = useCallback(async () => {
const created = await api.createJobChatThread(job.id, {
title: `${job.title} @ ${job.employer}`,
});
setMessages(data.messages);
}, [job.id]);
setThreads((current) => [created.thread, ...current]);
setActiveThreadId(created.thread.id);
setMessages([]);
return created.thread;
}, [job.id, job.title, job.employer]);
const load = useCallback(async () => {
setIsLoading(true);
try {
await loadMessages();
const data = await api.listJobChatThreads(job.id);
const nextThreads = data.threads;
setThreads(nextThreads);
let threadId = nextThreads[0]?.id ?? null;
if (!threadId) {
const created = await createThread();
threadId = created.id;
}
setActiveThreadId(threadId);
if (threadId) {
await loadThreadMessages(threadId);
}
} catch (error) {
const message =
error instanceof Error ? error.message : "Failed to load Ghostwriter";
@ -52,7 +92,7 @@ export const GhostwriterPanel: React.FC<GhostwriterPanelProps> = ({ job }) => {
} finally {
setIsLoading(false);
}
}, [loadMessages]);
}, [job.id, createThread, loadThreadMessages]);
useEffect(() => {
void load();
@ -68,9 +108,8 @@ export const GhostwriterPanel: React.FC<GhostwriterPanelProps> = ({ job }) => {
setActiveRunId(event.runId);
setStreamingMessageId(event.messageId);
setMessages((current) => {
if (current.some((message) => message.id === event.messageId)) {
if (current.some((message) => message.id === event.messageId))
return current;
}
return [
...current,
{
@ -135,11 +174,12 @@ export const GhostwriterPanel: React.FC<GhostwriterPanelProps> = ({ job }) => {
const sendMessage = useCallback(
async (content: string) => {
if (isStreaming) return;
if (!activeThreadIdRef.current || isStreaming) return;
const threadId = activeThreadIdRef.current;
const optimisticUser: JobChatMessage = {
id: `tmp-user-${Date.now()}`,
threadId: messages[messages.length - 1]?.threadId || "pending-thread",
threadId,
jobId: job.id,
role: "user",
content,
@ -159,13 +199,14 @@ export const GhostwriterPanel: React.FC<GhostwriterPanelProps> = ({ job }) => {
streamAbortRef.current = controller;
try {
await api.streamJobGhostwriterMessage(
await api.streamJobChatMessage(
job.id,
threadId,
{ content, signal: controller.signal },
{ onEvent: onStreamEvent },
);
await loadMessages();
await loadThreadMessages(threadId);
} catch (error) {
if (error instanceof Error && error.name === "AbortError") {
return;
@ -179,30 +220,25 @@ export const GhostwriterPanel: React.FC<GhostwriterPanelProps> = ({ job }) => {
setIsStreaming(false);
}
},
[isStreaming, job.id, loadMessages, messages, onStreamEvent],
[isStreaming, job.id, loadThreadMessages, onStreamEvent],
);
const stopStreaming = useCallback(async () => {
streamAbortRef.current?.abort();
streamAbortRef.current = null;
setIsStreaming(false);
setStreamingMessageId(null);
const runId = activeRunId;
setActiveRunId(null);
if (!runId) {
return;
}
if (!activeThreadId || !activeRunId) return;
try {
await api.cancelJobGhostwriterRun(job.id, runId);
await loadMessages();
await api.cancelJobChatRun(job.id, activeThreadId, activeRunId);
streamAbortRef.current?.abort();
streamAbortRef.current = null;
setIsStreaming(false);
setActiveRunId(null);
setStreamingMessageId(null);
await loadThreadMessages(activeThreadId);
} catch (error) {
const message =
error instanceof Error ? error.message : "Failed to stop run";
toast.error(message);
}
}, [activeRunId, job.id, loadMessages]);
}, [activeThreadId, activeRunId, job.id, loadThreadMessages]);
const canRegenerate = useMemo(() => {
if (isStreaming || messages.length === 0) return false;
@ -211,7 +247,7 @@ export const GhostwriterPanel: React.FC<GhostwriterPanelProps> = ({ job }) => {
}, [isStreaming, messages]);
const regenerate = useCallback(async () => {
if (isStreaming || messages.length === 0) return;
if (!activeThreadId || isStreaming || messages.length === 0) return;
const last = messages[messages.length - 1];
if (last.role !== "assistant") return;
@ -220,13 +256,14 @@ export const GhostwriterPanel: React.FC<GhostwriterPanelProps> = ({ job }) => {
streamAbortRef.current = controller;
try {
await api.streamRegenerateJobGhostwriterMessage(
await api.streamRegenerateJobChatMessage(
job.id,
activeThreadId,
last.id,
{ signal: controller.signal },
{ onEvent: onStreamEvent },
);
await loadMessages();
await loadThreadMessages(activeThreadId);
} catch (error) {
if (error instanceof Error && error.name === "AbortError") {
return;
@ -240,44 +277,60 @@ export const GhostwriterPanel: React.FC<GhostwriterPanelProps> = ({ job }) => {
streamAbortRef.current = null;
setIsStreaming(false);
}
}, [isStreaming, job.id, loadMessages, messages, onStreamEvent]);
}, [
activeThreadId,
isStreaming,
job.id,
loadThreadMessages,
messages,
onStreamEvent,
]);
return (
<div className="flex h-full min-h-0 flex-1 flex-col">
<div
ref={messageListRef}
className="min-h-0 flex-1 overflow-y-auto border-b border-border/50 pb-3 pr-1"
>
{messages.length === 0 && !isLoading ? (
<div className="flex h-full min-h-[260px] justify-center px-3 flex-col text-left">
<h4 className="font-medium">
{job.title} at {job.employer}
</h4>
<p className="mt-2 text-sm leading-relaxed text-muted-foreground">
Ghostwriter already has this job description, your resume and your
writing style preferences. Ask for tailored response drafts, or
concise role-fit talking points.
</p>
</div>
) : (
<Card className="border-border/50">
<CardHeader>
<CardTitle className="flex items-center gap-2 text-base">
<MessageSquare className="h-4 w-4" />
Ghostwriter
</CardTitle>
</CardHeader>
<CardContent className="space-y-4">
<ThreadList
threads={threads}
activeThreadId={activeThreadId}
onSelectThread={(threadId) => {
setActiveThreadId(threadId);
void loadThreadMessages(threadId);
}}
onCreateThread={() => {
void createThread();
}}
disabled={isLoading || isStreaming}
/>
<div
ref={messageListRef}
className="max-h-[420px] overflow-y-auto rounded-md border border-border/50 p-3"
>
<MessageList
messages={messages}
isStreaming={isStreaming}
streamingMessageId={streamingMessageId}
/>
)}
</div>
</div>
<div className="mt-4">
<Composer
disabled={isLoading || isStreaming}
<RunControls
isStreaming={isStreaming}
canRegenerate={canRegenerate}
onRegenerate={regenerate}
onStop={stopStreaming}
onRegenerate={regenerate}
/>
<Composer
disabled={isLoading || isStreaming || !activeThreadId}
onSend={sendMessage}
/>
</div>
</div>
</CardContent>
</Card>
);
};

23
orchestrator/src/client/components/ghostwriter/MessageList.tsx
View File

@ -1,7 +1,5 @@
import type { JobChatMessage } from "@shared/types";
import type React from "react";
import ReactMarkdown from "react-markdown";
import remarkGfm from "remark-gfm";
import { StreamingMessage } from "./StreamingMessage";
type MessageListProps = {
@ -17,7 +15,12 @@ export const MessageList: React.FC<MessageListProps> = ({
}) => {
return (
<div className="space-y-3">
{messages.length > 0 &&
{messages.length === 0 ? (
<div className="rounded-md border border-dashed border-border/60 p-3 text-xs text-muted-foreground">
Ask for interview prep, response drafts, or application strategy for
this job.
</div>
) : (
messages.map((message) => {
const isUser = message.role === "user";
const isActiveStreaming =
@ -37,24 +40,20 @@ export const MessageList: React.FC<MessageListProps> = ({
<div className="mb-1 text-[10px] uppercase tracking-wide text-muted-foreground">
{isUser
? "You"
: `Ghostwriter${message.version > 1 ? ` v${message.version}` : ""}`}
: `Copilot${message.version > 1 ? ` v${message.version}` : ""}`}
</div>
{isActiveStreaming ? (
<StreamingMessage content={message.content} />
) : message.role === "assistant" ? (
<div className="text-sm leading-relaxed text-foreground [&_a]:text-primary [&_a]:underline [&_blockquote]:border-l [&_blockquote]:border-border [&_blockquote]:pl-3 [&_code]:rounded [&_code]:bg-muted/40 [&_code]:px-1 [&_h1]:mt-4 [&_h1]:text-base [&_h1]:font-semibold [&_h2]:mt-4 [&_h2]:text-sm [&_h2]:font-semibold [&_h3]:mt-3 [&_h3]:text-sm [&_h3]:font-semibold [&_li]:my-1 [&_ol]:my-2 [&_ol]:list-decimal [&_ol]:pl-5 [&_p]:my-2 [&_pre]:my-3 [&_pre]:overflow-x-auto [&_pre]:rounded [&_pre]:bg-muted/40 [&_pre]:p-3 [&_ul]:my-2 [&_ul]:list-disc [&_ul]:pl-5">
<ReactMarkdown remarkPlugins={[remarkGfm]}>
{message.content || "..."}
</ReactMarkdown>
</div>
) : (
<div className="whitespace-pre-wrap text-sm leading-relaxed text-foreground">
{message.content}
{message.content ||
(message.role === "assistant" ? "..." : "")}
</div>
)}
</div>
);
})}
})
)}
</div>
);
};

118
orchestrator/src/client/components/ghostwriter/ThreadList.tsx
View File

@ -1,72 +1,27 @@
import type { Job, JobChatThread } from "@shared/types";
import type { JobChatThread } from "@shared/types";
import type React from "react";
import { Button } from "@/components/ui/button";
import { cn } from "@/lib/utils";
type ThreadListProps = {
job: Job;
threads: JobChatThread[];
previews: Record<string, string>;
activeThreadId: string | null;
onSelectThread: (threadId: string) => void;
onCreateThread: () => void;
disabled?: boolean;
};
function formatRelativeTime(value: string | null): string {
if (!value) return "Updated just now";
const date = new Date(value);
if (Number.isNaN(date.getTime())) return "Updated recently";
const diffMs = date.getTime() - Date.now();
const absMs = Math.abs(diffMs);
const minute = 60 * 1000;
const hour = 60 * minute;
const day = 24 * hour;
const rtf = new Intl.RelativeTimeFormat("en", { numeric: "auto" });
if (absMs < hour) {
const magnitudeMinutes = Math.max(1, Math.round(absMs / minute));
const minutes = Math.sign(diffMs) * magnitudeMinutes;
return `Updated ${rtf.format(minutes, "minute")}`;
}
if (absMs < day) {
const hours = Math.round(diffMs / hour);
return `Updated ${rtf.format(hours, "hour")}`;
}
const days = Math.round(diffMs / day);
return `Updated ${rtf.format(days, "day")}`;
}
function normalizeThreadTitle(input: string | null, fallback: string): string {
const value = input?.trim();
return value && value.length > 0 ? value : fallback;
}
export const ThreadList: React.FC<ThreadListProps> = ({
job,
threads,
previews,
activeThreadId,
onSelectThread,
onCreateThread,
disabled,
}) => {
const titleCounts = new Map<string, number>();
threads.forEach((thread) => {
const normalized = normalizeThreadTitle(
thread.title,
`${job.title} @ ${job.employer}`,
);
titleCounts.set(normalized, (titleCounts.get(normalized) ?? 0) + 1);
});
const seenTitles = new Map<string, number>();
return (
<aside className="min-h-0 space-y-3 pr-0 md:pr-4">
<div className="space-y-2">
<div className="flex items-center justify-between">
<div className="text-[11px] font-medium uppercase tracking-wide text-muted-foreground">
<div className="text-xs uppercase tracking-wide text-muted-foreground">
Threads
</div>
<Button
@ -74,67 +29,34 @@ export const ThreadList: React.FC<ThreadListProps> = ({
variant="outline"
onClick={onCreateThread}
disabled={disabled}
className="h-8 px-2.5 text-xs"
>
New
</Button>
</div>
<div className="max-h-[13rem] space-y-1 overflow-auto pr-1">
<div className="max-h-40 space-y-1 overflow-auto rounded-md border border-border/50 p-1">
{threads.length === 0 ? (
<div className="p-2 text-xs text-muted-foreground">
No threads yet
</div>
) : (
threads.map((thread) => {
const rawTitle = normalizeThreadTitle(
thread.title,
`${job.title} @ ${job.employer}`,
);
const seenCount = (seenTitles.get(rawTitle) ?? 0) + 1;
seenTitles.set(rawTitle, seenCount);
const hasDuplicates = (titleCounts.get(rawTitle) ?? 0) > 1;
const title = hasDuplicates
? `${rawTitle} (${seenCount})`
: rawTitle;
const preview = previews[thread.id]?.trim() || "No messages yet";
const isActive = activeThreadId === thread.id;
return (
<button
key={thread.id}
type="button"
onClick={() => onSelectThread(thread.id)}
disabled={disabled}
className={cn(
"relative w-full rounded-md border px-3 py-2.5 text-left transition-colors",
isActive
? "border-foreground/25 bg-accent text-accent-foreground"
: "border-transparent hover:border-border/50 hover:bg-muted/40",
)}
>
{isActive && (
<span className="absolute bottom-2 left-0 top-2 w-0.5 rounded-r bg-foreground" />
)}
<div
className={cn(
"truncate pr-2 text-xs",
isActive ? "font-semibold" : "font-medium text-foreground",
)}
>
{title}
</div>
<div className="mt-1 text-[10px] text-muted-foreground">
{formatRelativeTime(thread.lastMessageAt ?? thread.updatedAt)}
</div>
<div className="mt-1 truncate text-[11px] text-muted-foreground/90">
{preview}
</div>
</button>
);
})
threads.map((thread) => (
<button
key={thread.id}
type="button"
onClick={() => onSelectThread(thread.id)}
className={cn(
"w-full rounded px-2 py-1.5 text-left text-xs transition-colors",
activeThreadId === thread.id
? "bg-primary/10 text-primary"
: "hover:bg-muted",
)}
>
{thread.title || "Untitled thread"}
</button>
))
)}
</div>
</aside>
</div>
);
};

52
orchestrator/src/client/components/job-chat/Composer.tsx
View File

@ -1,52 +0,0 @@
import { Send } from "lucide-react";
import type React from "react";
import { useState } from "react";
import { Button } from "@/components/ui/button";
import { Textarea } from "@/components/ui/textarea";
type ComposerProps = {
disabled?: boolean;
onSend: (content: string) => Promise<void>;
};
export const Composer: React.FC<ComposerProps> = ({ disabled, onSend }) => {
const [value, setValue] = useState("");
const submit = async () => {
const content = value.trim();
if (!content || disabled) return;
setValue("");
await onSend(content);
};
return (
<div className="space-y-2">
<Textarea
placeholder="Ask anything about this job..."
value={value}
onChange={(event) => setValue(event.target.value)}
disabled={disabled}
onKeyDown={(event) => {
if ((event.metaKey || event.ctrlKey) && event.key === "Enter") {
event.preventDefault();
void submit();
}
}}
className="min-h-[84px]"
/>
<div className="flex items-center justify-between">
<div className="text-[10px] text-muted-foreground">
Cmd/Ctrl+Enter to send
</div>
<Button
size="sm"
onClick={() => void submit()}
disabled={disabled || !value.trim()}
>
<Send className="mr-1 h-3.5 w-3.5" />
Send
</Button>
</div>
</div>
);
};

336
orchestrator/src/client/components/job-chat/JobChatPanel.tsx
View File

@ -1,336 +0,0 @@
import type {
Job,
JobChatMessage,
JobChatStreamEvent,
JobChatThread,
} from "@shared/types";
import { MessageSquare } from "lucide-react";
import type React from "react";
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import { toast } from "sonner";
import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
import * as api from "../../api";
import { Composer } from "./Composer";
import { MessageList } from "./MessageList";
import { RunControls } from "./RunControls";
import { ThreadList } from "./ThreadList";
type JobChatPanelProps = {
job: Job;
};
export const JobChatPanel: React.FC<JobChatPanelProps> = ({ job }) => {
const [threads, setThreads] = useState<JobChatThread[]>([]);
const [activeThreadId, setActiveThreadId] = useState<string | null>(null);
const [messages, setMessages] = useState<JobChatMessage[]>([]);
const [isLoading, setIsLoading] = useState(true);
const [isStreaming, setIsStreaming] = useState(false);
const [streamingMessageId, setStreamingMessageId] = useState<string | null>(
null,
);
const [activeRunId, setActiveRunId] = useState<string | null>(null);
const messageListRef = useRef<HTMLDivElement | null>(null);
const streamAbortRef = useRef<AbortController | null>(null);
const activeThreadIdRef = useRef<string | null>(null);
useEffect(() => {
activeThreadIdRef.current = activeThreadId;
}, [activeThreadId]);
useEffect(() => {
const container = messageListRef.current;
if (!container) return;
const distanceToBottom =
container.scrollHeight - container.scrollTop - container.clientHeight;
if (distanceToBottom < 120 || isStreaming) {
container.scrollTop = container.scrollHeight;
}
});
const loadThreadMessages = useCallback(
async (threadId: string) => {
const data = await api.listJobChatMessages(job.id, threadId, {
limit: 300,
});
setMessages(data.messages);
},
[job.id],
);
const createThread = useCallback(async () => {
const created = await api.createJobChatThread(job.id, {
title: `${job.title} @ ${job.employer}`,
});
setThreads((current) => [created.thread, ...current]);
setActiveThreadId(created.thread.id);
setMessages([]);
return created.thread;
}, [job.id, job.title, job.employer]);
const load = useCallback(async () => {
setIsLoading(true);
try {
const data = await api.listJobChatThreads(job.id);
const nextThreads = data.threads;
setThreads(nextThreads);
let threadId = nextThreads[0]?.id ?? null;
if (!threadId) {
const created = await createThread();
threadId = created.id;
}
setActiveThreadId(threadId);
if (threadId) {
await loadThreadMessages(threadId);
}
} catch (error) {
const message =
error instanceof Error ? error.message : "Failed to load Ghostwriter";
toast.error(message);
} finally {
setIsLoading(false);
}
}, [job.id, createThread, loadThreadMessages]);
useEffect(() => {
void load();
return () => {
streamAbortRef.current?.abort();
streamAbortRef.current = null;
};
}, [load]);
const onStreamEvent = useCallback(
(event: JobChatStreamEvent) => {
if (event.type === "ready") {
setActiveRunId(event.runId);
setStreamingMessageId(event.messageId);
setMessages((current) => {
if (current.some((message) => message.id === event.messageId))
return current;
return [
...current,
{
id: event.messageId,
threadId: event.threadId,
jobId: job.id,
role: "assistant",
content: "",
status: "partial",
tokensIn: null,
tokensOut: null,
version: 1,
replacesMessageId: null,
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
},
];
});
return;
}
if (event.type === "delta") {
setMessages((current) =>
current.map((message) =>
message.id === event.messageId
? {
...message,
content: `${message.content}${event.delta}`,
status: "partial",
updatedAt: new Date().toISOString(),
}
: message,
),
);
return;
}
if (event.type === "completed" || event.type === "cancelled") {
setMessages((current) => {
const next = current.filter(
(message) => message.id !== event.message.id,
);
return [...next, event.message].sort((a, b) =>
a.createdAt.localeCompare(b.createdAt),
);
});
setStreamingMessageId(null);
setActiveRunId(null);
setIsStreaming(false);
return;
}
if (event.type === "error") {
toast.error(event.message);
setStreamingMessageId(null);
setActiveRunId(null);
setIsStreaming(false);
}
},
[job.id],
);
const sendMessage = useCallback(
async (content: string) => {
if (!activeThreadIdRef.current || isStreaming) return;
const threadId = activeThreadIdRef.current;
const optimisticUser: JobChatMessage = {
id: `tmp-user-${Date.now()}`,
threadId,
jobId: job.id,
role: "user",
content,
status: "complete",
tokensIn: null,
tokensOut: null,
version: 1,
replacesMessageId: null,
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
};
setMessages((current) => [...current, optimisticUser]);
setIsStreaming(true);
const controller = new AbortController();
streamAbortRef.current = controller;
try {
await api.streamJobChatMessage(
job.id,
threadId,
{ content, signal: controller.signal },
{ onEvent: onStreamEvent },
);
await loadThreadMessages(threadId);
} catch (error) {
if (error instanceof Error && error.name === "AbortError") {
return;
}
const message =
error instanceof Error ? error.message : "Failed to send message";
toast.error(message);
} finally {
streamAbortRef.current = null;
setIsStreaming(false);
}
},
[isStreaming, job.id, loadThreadMessages, onStreamEvent],
);
const stopStreaming = useCallback(async () => {
if (!activeThreadId || !activeRunId) return;
try {
await api.cancelJobChatRun(job.id, activeThreadId, activeRunId);
streamAbortRef.current?.abort();
streamAbortRef.current = null;
setIsStreaming(false);
setActiveRunId(null);
setStreamingMessageId(null);
await loadThreadMessages(activeThreadId);
} catch (error) {
const message =
error instanceof Error ? error.message : "Failed to stop run";
toast.error(message);
}
}, [activeThreadId, activeRunId, job.id, loadThreadMessages]);
const canRegenerate = useMemo(() => {
if (isStreaming || messages.length === 0) return false;
const last = messages[messages.length - 1];
return last.role === "assistant";
}, [isStreaming, messages]);
const regenerate = useCallback(async () => {
if (!activeThreadId || isStreaming || messages.length === 0) return;
const last = messages[messages.length - 1];
if (last.role !== "assistant") return;
setIsStreaming(true);
const controller = new AbortController();
streamAbortRef.current = controller;
try {
await api.streamRegenerateJobChatMessage(
job.id,
activeThreadId,
last.id,
{ signal: controller.signal },
{ onEvent: onStreamEvent },
);
await loadThreadMessages(activeThreadId);
} catch (error) {
if (error instanceof Error && error.name === "AbortError") {
return;
}
const message =
error instanceof Error
? error.message
: "Failed to regenerate response";
toast.error(message);
} finally {
streamAbortRef.current = null;
setIsStreaming(false);
}
}, [
activeThreadId,
isStreaming,
job.id,
loadThreadMessages,
messages,
onStreamEvent,
]);
return (
<Card className="border-border/50">
<CardHeader>
<CardTitle className="flex items-center gap-2 text-base">
<MessageSquare className="h-4 w-4" />
Ghostwriter
</CardTitle>
</CardHeader>
<CardContent className="space-y-4">
<ThreadList
threads={threads}
activeThreadId={activeThreadId}
onSelectThread={(threadId) => {
setActiveThreadId(threadId);
void loadThreadMessages(threadId);
}}
onCreateThread={() => {
void createThread();
}}
disabled={isLoading || isStreaming}
/>
<div
ref={messageListRef}
className="max-h-[420px] overflow-y-auto rounded-md border border-border/50 p-3"
>
<MessageList
messages={messages}
isStreaming={isStreaming}
streamingMessageId={streamingMessageId}
/>
</div>
<RunControls
isStreaming={isStreaming}
canRegenerate={canRegenerate}
onStop={stopStreaming}
onRegenerate={regenerate}
/>
<Composer
disabled={isLoading || isStreaming || !activeThreadId}
onSend={sendMessage}
/>
</CardContent>
</Card>
);
};

59
orchestrator/src/client/components/job-chat/MessageList.tsx
View File

@ -1,59 +0,0 @@
import type { JobChatMessage } from "@shared/types";
import type React from "react";
import { StreamingMessage } from "./StreamingMessage";
type MessageListProps = {
messages: JobChatMessage[];
isStreaming: boolean;
streamingMessageId: string | null;
};
export const MessageList: React.FC<MessageListProps> = ({
messages,
isStreaming,
streamingMessageId,
}) => {
return (
<div className="space-y-3">
{messages.length === 0 ? (
<div className="rounded-md border border-dashed border-border/60 p-3 text-xs text-muted-foreground">
Ask for interview prep, response drafts, or application strategy for
this job.
</div>
) : (
messages.map((message) => {
const isUser = message.role === "user";
const isActiveStreaming =
isStreaming &&
message.role === "assistant" &&
streamingMessageId === message.id;
return (
<div
key={message.id}
className={`rounded-lg border p-3 ${
isUser
? "border-primary/30 bg-primary/5"
: "border-border/60 bg-background"
}`}
>
<div className="mb-1 text-[10px] uppercase tracking-wide text-muted-foreground">
{isUser
? "You"
: `Copilot${message.version > 1 ? ` v${message.version}` : ""}`}
</div>
{isActiveStreaming ? (
<StreamingMessage content={message.content} />
) : (
<div className="whitespace-pre-wrap text-sm leading-relaxed text-foreground">
{message.content ||
(message.role === "assistant" ? "..." : "")}
</div>
)}
</div>
);
})
)}
</div>
);
};

45
orchestrator/src/client/components/job-chat/RunControls.tsx
View File

@ -1,45 +0,0 @@
import { Loader2, RefreshCcw, Square } from "lucide-react";
import type React from "react";
import { Button } from "@/components/ui/button";
type RunControlsProps = {
isStreaming: boolean;
canRegenerate: boolean;
onStop: () => Promise<void>;
onRegenerate: () => Promise<void>;
};
export const RunControls: React.FC<RunControlsProps> = ({
isStreaming,
canRegenerate,
onStop,
onRegenerate,
}) => {
return (
<div className="flex items-center justify-end gap-2">
{isStreaming ? (
<Button size="sm" variant="outline" onClick={() => void onStop()}>
<Square className="mr-1 h-3.5 w-3.5" />
Stop
</Button>
) : (
<Button
size="sm"
variant="outline"
onClick={() => void onRegenerate()}
disabled={!canRegenerate}
>
<RefreshCcw className="mr-1 h-3.5 w-3.5" />
Regenerate
</Button>
)}
{isStreaming && (
<div className="inline-flex items-center gap-1 text-xs text-muted-foreground">
<Loader2 className="h-3.5 w-3.5 animate-spin" />
Generating
</div>
)}
</div>
);
};

16
orchestrator/src/client/components/job-chat/StreamingMessage.tsx
View File

@ -1,16 +0,0 @@
import type React from "react";
type StreamingMessageProps = {
content: string;
};
export const StreamingMessage: React.FC<StreamingMessageProps> = ({
content,
}) => {
return (
<div className="whitespace-pre-wrap text-sm leading-relaxed text-foreground">
{content}
<span className="ml-1 inline-block h-4 w-2 animate-pulse rounded bg-primary/60 align-middle" />
</div>
);
};

62
orchestrator/src/client/components/job-chat/ThreadList.tsx
View File

@ -1,62 +0,0 @@
import type { JobChatThread } from "@shared/types";
import type React from "react";
import { Button } from "@/components/ui/button";
import { cn } from "@/lib/utils";
type ThreadListProps = {
threads: JobChatThread[];
activeThreadId: string | null;
onSelectThread: (threadId: string) => void;
onCreateThread: () => void;
disabled?: boolean;
};
export const ThreadList: React.FC<ThreadListProps> = ({
threads,
activeThreadId,
onSelectThread,
onCreateThread,
disabled,
}) => {
return (
<div className="space-y-2">
<div className="flex items-center justify-between">
<div className="text-xs uppercase tracking-wide text-muted-foreground">
Threads
</div>
<Button
size="sm"
variant="outline"
onClick={onCreateThread}
disabled={disabled}
>
New
</Button>
</div>
<div className="max-h-40 space-y-1 overflow-auto rounded-md border border-border/50 p-1">
{threads.length === 0 ? (
<div className="p-2 text-xs text-muted-foreground">
No threads yet
</div>
) : (
threads.map((thread) => (
<button
key={thread.id}
type="button"
onClick={() => onSelectThread(thread.id)}
className={cn(
"w-full rounded px-2 py-1.5 text-left text-xs transition-colors",
activeThreadId === thread.id
? "bg-primary/10 text-primary"
: "hover:bg-muted",
)}
>
{thread.title || "Untitled thread"}
</button>
))
)}
</div>
</div>
);
};

5
orchestrator/src/client/pages/JobPage.tsx
View File

@ -23,9 +23,8 @@ import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
import { formatTimestamp } from "@/lib/utils";
import * as api from "../api";
import { ConfirmDelete } from "../components/ConfirmDelete";
import { GhostwriterDrawer } from "../components/ghostwriter/GhostwriterDrawer";
import { GhostwriterPanel } from "../components/ghostwriter/GhostwriterPanel";
import { JobHeader } from "../components/JobHeader";
import { JobChatPanel } from "../components/job-chat/JobChatPanel";
import {
type LogEventFormValues,
LogEventModal,
@ -354,7 +353,7 @@ export const JobPage: React.FC = () => {
</Card>
)}
{job && <JobChatPanel job={job} />}
{job && <GhostwriterPanel job={job} />}
</div>
</div>

4
orchestrator/src/server/api/routes.ts
View File

@ -6,7 +6,7 @@ import { Router } from "express";
import { backupRouter } from "./routes/backup";
import { databaseRouter } from "./routes/database";
import { demoRouter } from "./routes/demo";
import { jobChatRouter } from "./routes/job-chat";
import { ghostwriterRouter } from "./routes/ghostwriter";
import { jobsRouter } from "./routes/jobs";
import { manualJobsRouter } from "./routes/manual-jobs";
import { onboardingRouter } from "./routes/onboarding";
@ -21,7 +21,7 @@ import { webhookRouter } from "./routes/webhook";
export const apiRouter = Router();
apiRouter.use("/jobs", jobsRouter);
apiRouter.use("/jobs/:id/chat", jobChatRouter);
apiRouter.use("/jobs/:id/chat", ghostwriterRouter);
apiRouter.use("/demo", demoRouter);
apiRouter.use("/settings", settingsRouter);
apiRouter.use("/pipeline", pipelineRouter);

75
orchestrator/src/server/api/routes/ghostwriter.test.ts
View File

@ -39,22 +39,6 @@ vi.mock("../../services/ghostwriter", () => ({
updatedAt: new Date().toISOString(),
},
]),
listMessagesForJob: vi.fn(async () => [
{
id: "message-1",
threadId: "thread-1",
jobId: "job-1",
role: "user",
content: "hello",
status: "complete",
tokensIn: 1,
tokensOut: null,
version: 1,
replacesMessageId: null,
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
},
]),
sendMessage: vi.fn(async () => ({
userMessage: {
id: "user-1",
@ -86,37 +70,6 @@ vi.mock("../../services/ghostwriter", () => ({
},
runId: "run-1",
})),
sendMessageForJob: vi.fn(async () => ({
userMessage: {
id: "user-1",
threadId: "thread-1",
jobId: "job-1",
role: "user",
content: "hello",
status: "complete",
tokensIn: 1,
tokensOut: null,
version: 1,
replacesMessageId: null,
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
},
assistantMessage: {
id: "assistant-1",
threadId: "thread-1",
jobId: "job-1",
role: "assistant",
content: "hi",
status: "complete",
tokensIn: 1,
tokensOut: 1,
version: 1,
replacesMessageId: null,
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
},
runId: "run-1",
})),
cancelRun: vi.fn(async () => ({ cancelled: true, alreadyFinished: false })),
regenerateMessage: vi.fn(async () => ({
runId: "run-2",
@ -151,8 +104,8 @@ describe.sequential("Ghostwriter API", () => {
await stopServer({ server, closeDb, tempDir });
});
it("lists messages with request id metadata", async () => {
const res = await fetch(`${baseUrl}/api/jobs/job-1/chat/messages`, {
it("lists threads with request id metadata", async () => {
const res = await fetch(`${baseUrl}/api/jobs/job-1/chat/threads`, {
headers: {
"x-request-id": "chat-req-1",
},
@ -162,18 +115,34 @@ describe.sequential("Ghostwriter API", () => {
expect(res.status).toBe(200);
expect(res.headers.get("x-request-id")).toBe("chat-req-1");
expect(body.ok).toBe(true);
expect(body.data.messages.length).toBe(1);
expect(body.data.threads.length).toBe(1);
expect(body.meta.requestId).toBe("chat-req-1");
});
it("sends a message in the per-job conversation", async () => {
const messageRes = await fetch(`${baseUrl}/api/jobs/job-1/chat/messages`, {
it("creates thread and sends a message", async () => {
const threadRes = await fetch(`${baseUrl}/api/jobs/job-1/chat/threads`, {
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({ content: "hello" }),
body: JSON.stringify({ title: "My thread" }),
});
const threadBody = await threadRes.json();
expect(threadRes.status).toBe(201);
expect(threadBody.ok).toBe(true);
expect(threadBody.data.thread.id).toBe("thread-created");
const messageRes = await fetch(
`${baseUrl}/api/jobs/job-1/chat/threads/thread-1/messages`,
{
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({ content: "hello" }),
},
);
const messageBody = await messageRes.json();
expect(messageRes.status).toBe(200);

352
orchestrator/src/server/api/routes/ghostwriter.ts
View File

@ -1,5 +1,4 @@
import { asyncRoute, fail, ok } from "@infra/http";
import { logger } from "@infra/logger";
import { runWithRequestContext } from "@infra/request-context";
import { badRequest, toAppError } from "@server/infra/errors";
import { type Request, type Response, Router } from "express";
@ -35,312 +34,9 @@ function getJobId(req: Request): string {
}
function writeSse(res: Response, event: unknown): void {
if (res.writableEnded || res.destroyed) return;
res.write(`data: ${JSON.stringify(event)}\n\n`);
}
function setupSseStream(
req: Request,
res: Response,
onDisconnectRun?: (runId: string) => Promise<void>,
): {
setRunId: (runId: string) => void;
isClosed: () => boolean;
cleanup: () => void;
} {
res.status(200);
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache, no-transform");
res.setHeader("Connection", "keep-alive");
res.setHeader("X-Accel-Buffering", "no");
res.flushHeaders?.();
let closed = false;
let activeRunId: string | null = null;
let disconnectHandled = false;
const handleDisconnect = () => {
if (disconnectHandled) return;
disconnectHandled = true;
if (!activeRunId || !onDisconnectRun) return;
void onDisconnectRun(activeRunId).catch((error) => {
logger.warn("Ghostwriter stream disconnect cancellation failed", {
runId: activeRunId,
error,
});
});
};
const heartbeat = setInterval(() => {
if (res.writableEnded || res.destroyed) return;
res.write(": heartbeat\n\n");
}, 30000);
const onClose = () => {
closed = true;
clearInterval(heartbeat);
handleDisconnect();
};
req.on("close", onClose);
return {
setRunId: (runId: string) => {
activeRunId = runId;
if (closed) {
handleDisconnect();
}
},
isClosed: () => closed,
cleanup: () => {
clearInterval(heartbeat);
req.off("close", onClose);
},
};
}
ghostwriterRouter.get(
"/messages",
asyncRoute(async (req, res) => {
const jobId = getJobId(req);
const parsed = listMessagesQuerySchema.safeParse(req.query);
if (!parsed.success) {
return fail(
res,
badRequest(parsed.error.message, parsed.error.flatten()),
);
}
await runWithRequestContext({ jobId }, async () => {
const messages = await ghostwriterService.listMessagesForJob({
jobId,
limit: parsed.data.limit,
offset: parsed.data.offset,
});
ok(res, { messages });
});
}),
);
ghostwriterRouter.post(
"/messages",
asyncRoute(async (req, res) => {
const jobId = getJobId(req);
const parsed = sendMessageSchema.safeParse(req.body);
if (!parsed.success) {
return fail(
res,
badRequest(parsed.error.message, parsed.error.flatten()),
);
}
await runWithRequestContext({ jobId }, async () => {
if (parsed.data.stream) {
const sse = setupSseStream(req, res, async (runId: string) => {
await ghostwriterService.cancelRunForJob({
jobId,
runId,
});
});
try {
await ghostwriterService.sendMessageForJob({
jobId,
content: parsed.data.content,
stream: {
onReady: ({ runId, threadId, messageId, requestId }) => {
sse.setRunId(runId);
if (sse.isClosed()) return;
writeSse(res, {
type: "ready",
runId,
threadId,
messageId,
requestId,
});
},
onDelta: ({ runId, messageId, delta }) =>
writeSse(res, {
type: "delta",
runId,
messageId,
delta,
}),
onCompleted: ({ runId, message }) =>
writeSse(res, {
type: "completed",
runId,
message,
}),
onCancelled: ({ runId, message }) =>
writeSse(res, {
type: "cancelled",
runId,
message,
}),
onError: ({ runId, code, message, requestId }) =>
writeSse(res, {
type: "error",
runId,
code,
message,
requestId,
}),
},
});
} catch (error) {
const appError = toAppError(error);
writeSse(res, {
type: "error",
code: appError.code,
message: appError.message,
requestId: res.getHeader("x-request-id") || "unknown",
});
} finally {
sse.cleanup();
if (!res.writableEnded) {
res.end();
}
}
return;
}
const result = await ghostwriterService.sendMessageForJob({
jobId,
content: parsed.data.content,
});
ok(res, {
userMessage: result.userMessage,
assistantMessage: result.assistantMessage,
runId: result.runId,
});
});
}),
);
ghostwriterRouter.post(
"/runs/:runId/cancel",
asyncRoute(async (req, res) => {
const jobId = getJobId(req);
const runId = req.params.runId;
if (!runId) {
return fail(res, badRequest("Missing run id"));
}
await runWithRequestContext({ jobId }, async () => {
const result = await ghostwriterService.cancelRunForJob({
jobId,
runId,
});
ok(res, result);
});
}),
);
ghostwriterRouter.post(
"/messages/:assistantMessageId/regenerate",
asyncRoute(async (req, res) => {
const jobId = getJobId(req);
const assistantMessageId = req.params.assistantMessageId;
if (!assistantMessageId) {
return fail(res, badRequest("Missing message id"));
}
const parsed = regenerateSchema.safeParse(req.body ?? {});
if (!parsed.success) {
return fail(
res,
badRequest(parsed.error.message, parsed.error.flatten()),
);
}
await runWithRequestContext({ jobId }, async () => {
if (parsed.data.stream) {
const sse = setupSseStream(req, res, async (runId: string) => {
await ghostwriterService.cancelRunForJob({
jobId,
runId,
});
});
try {
await ghostwriterService.regenerateMessageForJob({
jobId,
assistantMessageId,
stream: {
onReady: ({ runId, threadId, messageId, requestId }) => {
sse.setRunId(runId);
if (sse.isClosed()) return;
writeSse(res, {
type: "ready",
runId,
threadId,
messageId,
requestId,
});
},
onDelta: ({ runId, messageId, delta }) =>
writeSse(res, {
type: "delta",
runId,
messageId,
delta,
}),
onCompleted: ({ runId, message }) =>
writeSse(res, {
type: "completed",
runId,
message,
}),
onCancelled: ({ runId, message }) =>
writeSse(res, {
type: "cancelled",
runId,
message,
}),
onError: ({ runId, code, message, requestId }) =>
writeSse(res, {
type: "error",
runId,
code,
message,
requestId,
}),
},
});
} catch (error) {
const appError = toAppError(error);
writeSse(res, {
type: "error",
code: appError.code,
message: appError.message,
requestId: res.getHeader("x-request-id") || "unknown",
});
} finally {
sse.cleanup();
if (!res.writableEnded) {
res.end();
}
}
return;
}
const result = await ghostwriterService.regenerateMessageForJob({
jobId,
assistantMessageId,
});
ok(res, result);
});
}),
);
ghostwriterRouter.get(
"/threads",
asyncRoute(async (req, res) => {
@ -423,13 +119,11 @@ ghostwriterRouter.post(
await runWithRequestContext({ jobId }, async () => {
if (parsed.data.stream) {
const sse = setupSseStream(req, res, async (runId: string) => {
await ghostwriterService.cancelRun({
jobId,
threadId,
runId,
});
});
res.status(200);
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache, no-transform");
res.setHeader("Connection", "keep-alive");
res.flushHeaders?.();
try {
await ghostwriterService.sendMessage({
@ -437,17 +131,14 @@ ghostwriterRouter.post(
threadId,
content: parsed.data.content,
stream: {
onReady: ({ runId, messageId, requestId }) => {
sse.setRunId(runId);
if (sse.isClosed()) return;
onReady: ({ runId, messageId, requestId }) =>
writeSse(res, {
type: "ready",
runId,
threadId,
messageId,
requestId,
});
},
}),
onDelta: ({ runId, messageId, delta }) =>
writeSse(res, {
type: "delta",
@ -486,10 +177,7 @@ ghostwriterRouter.post(
requestId: res.getHeader("x-request-id") || "unknown",
});
} finally {
sse.cleanup();
if (!res.writableEnded) {
res.end();
}
res.end();
}
return;
@ -554,13 +242,11 @@ ghostwriterRouter.post(
await runWithRequestContext({ jobId }, async () => {
if (parsed.data.stream) {
const sse = setupSseStream(req, res, async (runId: string) => {
await ghostwriterService.cancelRun({
jobId,
threadId,
runId,
});
});
res.status(200);
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache, no-transform");
res.setHeader("Connection", "keep-alive");
res.flushHeaders?.();
try {
await ghostwriterService.regenerateMessage({
@ -568,17 +254,14 @@ ghostwriterRouter.post(
threadId,
assistantMessageId,
stream: {
onReady: ({ runId, messageId, requestId }) => {
sse.setRunId(runId);
if (sse.isClosed()) return;
onReady: ({ runId, messageId, requestId }) =>
writeSse(res, {
type: "ready",
runId,
threadId,
messageId,
requestId,
});
},
}),
onDelta: ({ runId, messageId, delta }) =>
writeSse(res, {
type: "delta",
@ -617,10 +300,7 @@ ghostwriterRouter.post(
requestId: res.getHeader("x-request-id") || "unknown",
});
} finally {
sse.cleanup();
if (!res.writableEnded) {
res.end();
}
res.end();
}
return;

154
orchestrator/src/server/api/routes/job-chat.test.ts
View File

@ -1,154 +0,0 @@
import type { Server } from "node:http";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { startServer, stopServer } from "./test-utils";
vi.mock("../../services/job-chat", () => ({
listThreads: vi.fn(async () => [
{
id: "thread-1",
jobId: "job-1",
title: "Thread",
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
lastMessageAt: new Date().toISOString(),
},
]),
createThread: vi.fn(
async (input: { jobId: string; title?: string | null }) => ({
id: "thread-created",
jobId: input.jobId,
title: input.title ?? null,
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
lastMessageAt: null,
}),
),
listMessages: vi.fn(async () => [
{
id: "message-1",
threadId: "thread-1",
jobId: "job-1",
role: "user",
content: "hello",
status: "complete",
tokensIn: 1,
tokensOut: null,
version: 1,
replacesMessageId: null,
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
},
]),
sendMessage: vi.fn(async () => ({
userMessage: {
id: "user-1",
threadId: "thread-1",
jobId: "job-1",
role: "user",
content: "hello",
status: "complete",
tokensIn: 1,
tokensOut: null,
version: 1,
replacesMessageId: null,
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
},
assistantMessage: {
id: "assistant-1",
threadId: "thread-1",
jobId: "job-1",
role: "assistant",
content: "hi",
status: "complete",
tokensIn: 1,
tokensOut: 1,
version: 1,
replacesMessageId: null,
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
},
runId: "run-1",
})),
cancelRun: vi.fn(async () => ({ cancelled: true, alreadyFinished: false })),
regenerateMessage: vi.fn(async () => ({
runId: "run-2",
assistantMessage: {
id: "assistant-2",
threadId: "thread-1",
jobId: "job-1",
role: "assistant",
content: "updated",
status: "complete",
tokensIn: 1,
tokensOut: 1,
version: 2,
replacesMessageId: "assistant-1",
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
},
})),
}));
describe.sequential("Ghostwriter API", () => {
let server: Server;
let baseUrl: string;
let closeDb: () => void;
let tempDir: string;
beforeEach(async () => {
({ server, baseUrl, closeDb, tempDir } = await startServer());
});
afterEach(async () => {
await stopServer({ server, closeDb, tempDir });
});
it("lists threads with request id metadata", async () => {
const res = await fetch(`${baseUrl}/api/jobs/job-1/chat/threads`, {
headers: {
"x-request-id": "chat-req-1",
},
});
const body = await res.json();
expect(res.status).toBe(200);
expect(res.headers.get("x-request-id")).toBe("chat-req-1");
expect(body.ok).toBe(true);
expect(body.data.threads.length).toBe(1);
expect(body.meta.requestId).toBe("chat-req-1");
});
it("creates thread and sends a message", async () => {
const threadRes = await fetch(`${baseUrl}/api/jobs/job-1/chat/threads`, {
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({ title: "My thread" }),
});
const threadBody = await threadRes.json();
expect(threadRes.status).toBe(201);
expect(threadBody.ok).toBe(true);
expect(threadBody.data.thread.id).toBe("thread-created");
const messageRes = await fetch(
`${baseUrl}/api/jobs/job-1/chat/threads/thread-1/messages`,
{
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({ content: "hello" }),
},
);
const messageBody = await messageRes.json();
expect(messageRes.status).toBe(200);
expect(messageBody.ok).toBe(true);
expect(messageBody.data.runId).toBe("run-1");
expect(messageBody.data.assistantMessage.role).toBe("assistant");
expect(typeof messageBody.meta.requestId).toBe("string");
});
});

318
orchestrator/src/server/api/routes/job-chat.ts
View File

@ -1,318 +0,0 @@
import { asyncRoute, fail, ok } from "@infra/http";
import { runWithRequestContext } from "@infra/request-context";
import { badRequest, toAppError } from "@server/infra/errors";
import { type Request, type Response, Router } from "express";
import { z } from "zod";
import * as jobChatService from "../../services/job-chat";
export const jobChatRouter = Router({ mergeParams: true });
const createThreadSchema = z.object({
title: z.string().trim().max(200).nullable().optional(),
});
const listMessagesQuerySchema = z.object({
limit: z.coerce.number().int().min(1).max(500).optional(),
offset: z.coerce.number().int().min(0).max(10000).optional(),
});
const sendMessageSchema = z.object({
content: z.string().trim().min(1).max(20000),
stream: z.boolean().optional(),
});
const regenerateSchema = z.object({
stream: z.boolean().optional(),
});
function getJobId(req: Request): string {
const jobId = req.params.id;
if (!jobId) {
throw badRequest("Missing job id");
}
return jobId;
}
function writeSse(res: Response, event: unknown): void {
res.write(`data: ${JSON.stringify(event)}\n\n`);
}
jobChatRouter.get(
"/threads",
asyncRoute(async (req, res) => {
const jobId = getJobId(req);
await runWithRequestContext({ jobId }, async () => {
const threads = await jobChatService.listThreads(jobId);
ok(res, { threads });
});
}),
);
jobChatRouter.post(
"/threads",
asyncRoute(async (req, res) => {
const jobId = getJobId(req);
const parsed = createThreadSchema.safeParse(req.body);
if (!parsed.success) {
return fail(
res,
badRequest(parsed.error.message, parsed.error.flatten()),
);
}
await runWithRequestContext({ jobId }, async () => {
const thread = await jobChatService.createThread({
jobId,
title: parsed.data.title,
});
ok(res, { thread }, 201);
});
}),
);
jobChatRouter.get(
"/threads/:threadId/messages",
asyncRoute(async (req, res) => {
const jobId = getJobId(req);
const threadId = req.params.threadId;
if (!threadId) {
return fail(res, badRequest("Missing thread id"));
}
const parsed = listMessagesQuerySchema.safeParse(req.query);
if (!parsed.success) {
return fail(
res,
badRequest(parsed.error.message, parsed.error.flatten()),
);
}
await runWithRequestContext({ jobId }, async () => {
const messages = await jobChatService.listMessages({
jobId,
threadId,
limit: parsed.data.limit,
offset: parsed.data.offset,
});
ok(res, { messages });
});
}),
);
jobChatRouter.post(
"/threads/:threadId/messages",
asyncRoute(async (req, res) => {
const jobId = getJobId(req);
const threadId = req.params.threadId;
if (!threadId) {
return fail(res, badRequest("Missing thread id"));
}
const parsed = sendMessageSchema.safeParse(req.body);
if (!parsed.success) {
return fail(
res,
badRequest(parsed.error.message, parsed.error.flatten()),
);
}
await runWithRequestContext({ jobId }, async () => {
if (parsed.data.stream) {
res.status(200);
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache, no-transform");
res.setHeader("Connection", "keep-alive");
res.flushHeaders?.();
try {
await jobChatService.sendMessage({
jobId,
threadId,
content: parsed.data.content,
stream: {
onReady: ({ runId, messageId, requestId }) =>
writeSse(res, {
type: "ready",
runId,
threadId,
messageId,
requestId,
}),
onDelta: ({ runId, messageId, delta }) =>
writeSse(res, {
type: "delta",
runId,
messageId,
delta,
}),
onCompleted: ({ runId, message }) =>
writeSse(res, {
type: "completed",
runId,
message,
}),
onCancelled: ({ runId, message }) =>
writeSse(res, {
type: "cancelled",
runId,
message,
}),
onError: ({ runId, code, message, requestId }) =>
writeSse(res, {
type: "error",
runId,
code,
message,
requestId,
}),
},
});
} catch (error) {
const appError = toAppError(error);
writeSse(res, {
type: "error",
code: appError.code,
message: appError.message,
requestId: res.getHeader("x-request-id") || "unknown",
});
} finally {
res.end();
}
return;
}
const result = await jobChatService.sendMessage({
jobId,
threadId,
content: parsed.data.content,
});
ok(res, {
userMessage: result.userMessage,
assistantMessage: result.assistantMessage,
runId: result.runId,
});
});
}),
);
jobChatRouter.post(
"/threads/:threadId/runs/:runId/cancel",
asyncRoute(async (req, res) => {
const jobId = getJobId(req);
const threadId = req.params.threadId;
const runId = req.params.runId;
if (!threadId || !runId) {
return fail(res, badRequest("Missing thread id or run id"));
}
await runWithRequestContext({ jobId }, async () => {
const result = await jobChatService.cancelRun({
jobId,
threadId,
runId,
});
ok(res, result);
});
}),
);
jobChatRouter.post(
"/threads/:threadId/messages/:assistantMessageId/regenerate",
asyncRoute(async (req, res) => {
const jobId = getJobId(req);
const threadId = req.params.threadId;
const assistantMessageId = req.params.assistantMessageId;
if (!threadId || !assistantMessageId) {
return fail(res, badRequest("Missing thread id or message id"));
}
const parsed = regenerateSchema.safeParse(req.body ?? {});
if (!parsed.success) {
return fail(
res,
badRequest(parsed.error.message, parsed.error.flatten()),
);
}
await runWithRequestContext({ jobId }, async () => {
if (parsed.data.stream) {
res.status(200);
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache, no-transform");
res.setHeader("Connection", "keep-alive");
res.flushHeaders?.();
try {
await jobChatService.regenerateMessage({
jobId,
threadId,
assistantMessageId,
stream: {
onReady: ({ runId, messageId, requestId }) =>
writeSse(res, {
type: "ready",
runId,
threadId,
messageId,
requestId,
}),
onDelta: ({ runId, messageId, delta }) =>
writeSse(res, {
type: "delta",
runId,
messageId,
delta,
}),
onCompleted: ({ runId, message }) =>
writeSse(res, {
type: "completed",
runId,
message,
}),
onCancelled: ({ runId, message }) =>
writeSse(res, {
type: "cancelled",
runId,
message,
}),
onError: ({ runId, code, message, requestId }) =>
writeSse(res, {
type: "error",
runId,
code,
message,
requestId,
}),
},
});
} catch (error) {
const appError = toAppError(error);
writeSse(res, {
type: "error",
code: appError.code,
message: appError.message,
requestId: res.getHeader("x-request-id") || "unknown",
});
} finally {
res.end();
}
return;
}
const result = await jobChatService.regenerateMessage({
jobId,
threadId,
assistantMessageId,
});
ok(res, result);
});
}),
);

39
orchestrator/src/server/repositories/ghostwriter.ts
View File

@ -70,20 +70,6 @@ export async function listThreadsForJob(
return rows.map(mapThread);
}
export async function getOrCreateThreadForJob(input: {
jobId: string;
title?: string | null;
}): Promise<JobChatThread> {
const existing = await listThreadsForJob(input.jobId);
if (existing.length > 0) {
return existing[0];
}
return createThread({
jobId: input.jobId,
title: input.title ?? null,
});
}
export async function getThreadById(
threadId: string,
): Promise<JobChatThread | null> {
@ -341,28 +327,3 @@ export async function completeRun(
return getRunById(runId);
}
export async function completeRunIfRunning(
runId: string,
input: {
status: Exclude<JobChatRunStatus, "running">;
errorCode?: string | null;
errorMessage?: string | null;
},
): Promise<JobChatRun | null> {
const nowEpoch = Date.now();
const nowIso = new Date(nowEpoch).toISOString();
await db
.update(jobChatRuns)
.set({
status: input.status,
completedAt: nowEpoch,
errorCode: input.errorCode ?? null,
errorMessage: input.errorMessage ?? null,
updatedAt: nowIso,
})
.where(and(eq(jobChatRuns.id, runId), eq(jobChatRuns.status, "running")));
return getRunById(runId);
}

329
orchestrator/src/server/repositories/job-chat.ts
View File

@ -1,329 +0,0 @@
import { randomUUID } from "node:crypto";
import type {
JobChatMessage,
JobChatMessageRole,
JobChatMessageStatus,
JobChatRun,
JobChatRunStatus,
JobChatThread,
} from "@shared/types";
import { and, desc, eq } from "drizzle-orm";
import { db, schema } from "../db";
const { jobChatMessages, jobChatRuns, jobChatThreads } = schema;
function mapThread(row: typeof jobChatThreads.$inferSelect): JobChatThread {
return {
id: row.id,
jobId: row.jobId,
title: row.title,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
lastMessageAt: row.lastMessageAt,
};
}
function mapMessage(row: typeof jobChatMessages.$inferSelect): JobChatMessage {
return {
id: row.id,
threadId: row.threadId,
jobId: row.jobId,
role: row.role as JobChatMessageRole,
content: row.content,
status: row.status as JobChatMessageStatus,
tokensIn: row.tokensIn,
tokensOut: row.tokensOut,
version: row.version,
replacesMessageId: row.replacesMessageId,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
};
}
function mapRun(row: typeof jobChatRuns.$inferSelect): JobChatRun {
return {
id: row.id,
threadId: row.threadId,
jobId: row.jobId,
status: row.status as JobChatRunStatus,
model: row.model,
provider: row.provider,
errorCode: row.errorCode,
errorMessage: row.errorMessage,
startedAt: row.startedAt,
completedAt: row.completedAt,
requestId: row.requestId,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
};
}
export async function listThreadsForJob(
jobId: string,
): Promise<JobChatThread[]> {
const rows = await db
.select()
.from(jobChatThreads)
.where(eq(jobChatThreads.jobId, jobId))
.orderBy(desc(jobChatThreads.updatedAt));
return rows.map(mapThread);
}
export async function getThreadById(
threadId: string,
): Promise<JobChatThread | null> {
const [row] = await db
.select()
.from(jobChatThreads)
.where(eq(jobChatThreads.id, threadId));
return row ? mapThread(row) : null;
}
export async function getThreadForJob(
jobId: string,
threadId: string,
): Promise<JobChatThread | null> {
const [row] = await db
.select()
.from(jobChatThreads)
.where(
and(eq(jobChatThreads.id, threadId), eq(jobChatThreads.jobId, jobId)),
);
return row ? mapThread(row) : null;
}
export async function createThread(input: {
jobId: string;
title?: string | null;
}): Promise<JobChatThread> {
const id = randomUUID();
const now = new Date().toISOString();
await db.insert(jobChatThreads).values({
id,
jobId: input.jobId,
title: input.title ?? null,
createdAt: now,
updatedAt: now,
lastMessageAt: null,
});
const thread = await getThreadById(id);
if (!thread) {
throw new Error(`Failed to load created chat thread ${id}.`);
}
return thread;
}
export async function touchThread(
threadId: string,
lastMessageAt?: string,
): Promise<void> {
const now = new Date().toISOString();
await db
.update(jobChatThreads)
.set({
updatedAt: now,
...(lastMessageAt !== undefined ? { lastMessageAt } : {}),
})
.where(eq(jobChatThreads.id, threadId));
}
export async function listMessagesForThread(
threadId: string,
options?: { limit?: number; offset?: number },
): Promise<JobChatMessage[]> {
const limit = options?.limit ?? 200;
const offset = options?.offset ?? 0;
const rows = await db
.select()
.from(jobChatMessages)
.where(eq(jobChatMessages.threadId, threadId))
.orderBy(jobChatMessages.createdAt)
.limit(limit)
.offset(offset);
return rows.map(mapMessage);
}
export async function getMessageById(
messageId: string,
): Promise<JobChatMessage | null> {
const [row] = await db
.select()
.from(jobChatMessages)
.where(eq(jobChatMessages.id, messageId));
return row ? mapMessage(row) : null;
}
export async function createMessage(input: {
threadId: string;
jobId: string;
role: JobChatMessageRole;
content: string;
status?: JobChatMessageStatus;
tokensIn?: number | null;
tokensOut?: number | null;
version?: number;
replacesMessageId?: string | null;
}): Promise<JobChatMessage> {
const id = randomUUID();
const now = new Date().toISOString();
await db.insert(jobChatMessages).values({
id,
threadId: input.threadId,
jobId: input.jobId,
role: input.role,
content: input.content,
status: input.status ?? "partial",
tokensIn: input.tokensIn ?? null,
tokensOut: input.tokensOut ?? null,
version: input.version ?? 1,
replacesMessageId: input.replacesMessageId ?? null,
createdAt: now,
updatedAt: now,
});
await touchThread(input.threadId, now);
const created = await getMessageById(id);
if (!created) {
throw new Error(`Failed to load created chat message ${id}.`);
}
return created;
}
export async function updateMessage(
messageId: string,
input: {
content?: string;
status?: JobChatMessageStatus;
tokensIn?: number | null;
tokensOut?: number | null;
},
): Promise<JobChatMessage | null> {
const now = new Date().toISOString();
await db
.update(jobChatMessages)
.set({
...(input.content !== undefined ? { content: input.content } : {}),
...(input.status !== undefined ? { status: input.status } : {}),
...(input.tokensIn !== undefined ? { tokensIn: input.tokensIn } : {}),
...(input.tokensOut !== undefined ? { tokensOut: input.tokensOut } : {}),
updatedAt: now,
})
.where(eq(jobChatMessages.id, messageId));
const message = await getMessageById(messageId);
if (message) {
await touchThread(message.threadId, now);
}
return message;
}
export async function getLatestAssistantMessage(
threadId: string,
): Promise<JobChatMessage | null> {
const [row] = await db
.select()
.from(jobChatMessages)
.where(
and(
eq(jobChatMessages.threadId, threadId),
eq(jobChatMessages.role, "assistant"),
),
)
.orderBy(desc(jobChatMessages.createdAt))
.limit(1);
return row ? mapMessage(row) : null;
}
export async function createRun(input: {
threadId: string;
jobId: string;
model: string | null;
provider: string | null;
requestId?: string | null;
}): Promise<JobChatRun> {
const id = randomUUID();
const startedAt = Date.now();
const now = new Date(startedAt).toISOString();
await db.insert(jobChatRuns).values({
id,
threadId: input.threadId,
jobId: input.jobId,
status: "running",
model: input.model,
provider: input.provider,
errorCode: null,
errorMessage: null,
startedAt,
completedAt: null,
requestId: input.requestId ?? null,
createdAt: now,
updatedAt: now,
});
const run = await getRunById(id);
if (!run) {
throw new Error(`Failed to load created chat run ${id}.`);
}
return run;
}
export async function getRunById(runId: string): Promise<JobChatRun | null> {
const [row] = await db
.select()
.from(jobChatRuns)
.where(eq(jobChatRuns.id, runId));
return row ? mapRun(row) : null;
}
export async function getActiveRunForThread(
threadId: string,
): Promise<JobChatRun | null> {
const [row] = await db
.select()
.from(jobChatRuns)
.where(
and(
eq(jobChatRuns.threadId, threadId),
eq(jobChatRuns.status, "running"),
),
)
.orderBy(desc(jobChatRuns.startedAt))
.limit(1);
return row ? mapRun(row) : null;
}
export async function completeRun(
runId: string,
input: {
status: Exclude<JobChatRunStatus, "running">;
errorCode?: string | null;
errorMessage?: string | null;
},
): Promise<JobChatRun | null> {
const nowEpoch = Date.now();
const nowIso = new Date(nowEpoch).toISOString();
await db
.update(jobChatRuns)
.set({
status: input.status,
completedAt: nowEpoch,
errorCode: input.errorCode ?? null,
errorMessage: input.errorMessage ?? null,
updatedAt: nowIso,
})
.where(eq(jobChatRuns.id, runId));
return getRunById(runId);
}

159
orchestrator/src/server/services/ghostwriter.ts
View File

@ -1,6 +1,5 @@
import { logger } from "@infra/logger";
import { getRequestId } from "@infra/request-context";
import type { JobChatMessage, JobChatRun } from "@shared/types";
import {
badRequest,
conflict,
@ -53,14 +52,6 @@ function chunkText(value: string, maxChunk = 60): string[] {
return chunks;
}
function isRunningRunUniqueConstraintError(error: unknown): boolean {
const message = error instanceof Error ? error.message : String(error);
return (
message.includes("idx_job_chat_runs_thread_running_unique") ||
message.includes("UNIQUE constraint failed: job_chat_runs.thread_id")
);
}
async function resolveLlmRuntimeSettings(): Promise<LlmRuntimeSettings> {
const overrides = await settingsRepo.getAllSettings();
@ -113,7 +104,6 @@ type GenerateReplyOptions = {
stream?: {
onReady: (payload: {
runId: string;
threadId: string;
messageId: string;
requestId: string;
}) => void;
@ -139,23 +129,15 @@ type GenerateReplyOptions = {
};
};
async function ensureJobThread(jobId: string, title?: string | null) {
return jobChatRepo.getOrCreateThreadForJob({
jobId,
title: title ?? null,
});
}
export async function createThread(input: {
jobId: string;
title?: string | null;
}) {
return ensureJobThread(input.jobId, input.title);
return jobChatRepo.createThread(input);
}
export async function listThreads(jobId: string) {
const thread = await ensureJobThread(jobId);
return [thread];
return jobChatRepo.listThreadsForJob(jobId);
}
export async function listMessages(input: {
@ -175,18 +157,6 @@ export async function listMessages(input: {
});
}
export async function listMessagesForJob(input: {
jobId: string;
limit?: number;
offset?: number;
}) {
const thread = await ensureJobThread(input.jobId);
return jobChatRepo.listMessagesForThread(thread.id, {
limit: input.limit,
offset: input.offset,
});
}
async function runAssistantReply(
options: GenerateReplyOptions,
): Promise<{ runId: string; messageId: string; message: string }> {
@ -211,47 +181,28 @@ async function runAssistantReply(
const requestId = getRequestId() ?? "unknown";
let run: JobChatRun;
try {
run = await jobChatRepo.createRun({
threadId: options.threadId,
jobId: options.jobId,
model: llmConfig.model,
provider: llmConfig.provider,
requestId,
});
} catch (error) {
if (isRunningRunUniqueConstraintError(error)) {
throw conflict("A chat generation is already running for this thread");
}
throw error;
}
const assistantMessage = await jobChatRepo.createMessage({
threadId: options.threadId,
jobId: options.jobId,
role: "assistant",
content: "",
status: "partial",
version: options.version ?? 1,
replacesMessageId: options.replaceMessageId ?? null,
});
let assistantMessage: JobChatMessage;
try {
assistantMessage = await jobChatRepo.createMessage({
threadId: options.threadId,
jobId: options.jobId,
role: "assistant",
content: "",
status: "partial",
version: options.version ?? 1,
replacesMessageId: options.replaceMessageId ?? null,
});
} catch (error) {
await jobChatRepo.completeRun(run.id, {
status: "failed",
errorCode: "INTERNAL_ERROR",
errorMessage: "Failed to create assistant message",
});
throw error;
}
const run = await jobChatRepo.createRun({
threadId: options.threadId,
jobId: options.jobId,
model: llmConfig.model,
provider: llmConfig.provider,
requestId,
});
const controller = new AbortController();
abortControllers.set(run.id, controller);
options.stream?.onReady({
runId: run.id,
threadId: options.threadId,
messageId: assistantMessage.id,
requestId,
});
@ -344,33 +295,10 @@ async function runAssistantReply(
},
);
const runAfterComplete = await jobChatRepo.completeRunIfRunning(run.id, {
await jobChatRepo.completeRun(run.id, {
status: "completed",
});
if (!runAfterComplete || runAfterComplete.status !== "completed") {
if (runAfterComplete?.status === "cancelled") {
const cancelledMessage = await jobChatRepo.updateMessage(
assistantMessage.id,
{
content: accumulated,
status: "cancelled",
tokensIn: estimateTokenCount(options.prompt),
tokensOut: estimateTokenCount(accumulated),
},
);
options.stream?.onCancelled({
runId: run.id,
message: cancelledMessage,
});
}
return {
runId: run.id,
messageId: assistantMessage.id,
message: accumulated,
};
}
options.stream?.onCompleted({
runId: run.id,
message: completedMessage,
@ -472,20 +400,6 @@ export async function sendMessage(input: {
};
}
export async function sendMessageForJob(input: {
jobId: string;
content: string;
stream?: GenerateReplyOptions["stream"];
}) {
const thread = await ensureJobThread(input.jobId);
return sendMessage({
jobId: input.jobId,
threadId: thread.id,
content: input.content,
stream: input.stream,
});
}
export async function regenerateMessage(input: {
jobId: string;
threadId: string;
@ -549,20 +463,6 @@ export async function regenerateMessage(input: {
};
}
export async function regenerateMessageForJob(input: {
jobId: string;
assistantMessageId: string;
stream?: GenerateReplyOptions["stream"];
}) {
const thread = await ensureJobThread(input.jobId);
return regenerateMessage({
jobId: input.jobId,
threadId: thread.id,
assistantMessageId: input.assistantMessageId,
stream: input.stream,
});
}
export async function cancelRun(input: {
jobId: string;
threadId: string;
@ -585,33 +485,14 @@ export async function cancelRun(input: {
controller.abort();
}
const runAfterCancel = await jobChatRepo.completeRunIfRunning(input.runId, {
await jobChatRepo.completeRun(input.runId, {
status: "cancelled",
errorCode: "REQUEST_TIMEOUT",
errorMessage: "Generation cancelled by user",
});
if (!runAfterCancel || runAfterCancel.status !== "cancelled") {
return {
cancelled: false,
alreadyFinished: true,
};
}
return {
cancelled: true,
alreadyFinished: false,
};
}
export async function cancelRunForJob(input: {
jobId: string;
runId: string;
}): Promise<{ cancelled: boolean; alreadyFinished: boolean }> {
const thread = await ensureJobThread(input.jobId);
return cancelRun({
jobId: input.jobId,
threadId: thread.id,
runId: input.runId,
});
}

193
orchestrator/src/server/services/job-chat-context.ts
View File

@ -1,193 +0,0 @@
import { logger } from "@infra/logger";
import { sanitizeUnknown } from "@infra/sanitize";
import type { Job, ResumeProfile } from "@shared/types";
import { badRequest, notFound } from "../infra/errors";
import * as jobsRepo from "../repositories/jobs";
import * as settingsRepo from "../repositories/settings";
import { getProfile } from "./profile";
import { resolveSettingValue } from "./settings-conversion";
type JobChatStyle = {
tone: string;
formality: string;
constraints: string;
doNotUse: string;
};
export type JobChatPromptContext = {
job: Job;
style: JobChatStyle;
systemPrompt: string;
jobSnapshot: string;
profileSnapshot: string;
};
const MAX_JOB_DESCRIPTION = 4000;
const MAX_PROFILE_SUMMARY = 1200;
const MAX_SKILLS = 18;
const MAX_PROJECTS = 6;
const MAX_EXPERIENCE = 5;
const MAX_ITEM_TEXT = 320;
function truncate(value: string | null | undefined, max: number): string {
if (!value) return "";
const trimmed = value.trim();
if (trimmed.length <= max) return trimmed;
return `${trimmed.slice(0, max)}...`;
}
function compactJoin(parts: Array<string | null | undefined>): string {
return parts.filter(Boolean).join("\n");
}
function buildJobSnapshot(job: Job): string {
const snapshot = {
event: "job.completed",
sentAt: new Date().toISOString(),
job: {
id: job.id,
source: job.source,
title: job.title,
employer: job.employer,
location: job.location,
salary: job.salary,
status: job.status,
jobUrl: job.jobUrl,
applicationLink: job.applicationLink,
suitabilityScore: job.suitabilityScore,
suitabilityReason: truncate(job.suitabilityReason, 600),
tailoredSummary: truncate(job.tailoredSummary, 1200),
tailoredHeadline: truncate(job.tailoredHeadline, 300),
tailoredSkills: truncate(job.tailoredSkills, 1200),
jobDescription: truncate(job.jobDescription, MAX_JOB_DESCRIPTION),
},
};
return JSON.stringify(snapshot, null, 2);
}
function buildProfileSnapshot(profile: ResumeProfile): string {
const summary =
truncate(profile?.sections?.summary?.content, MAX_PROFILE_SUMMARY) ||
truncate(profile?.basics?.summary, MAX_PROFILE_SUMMARY);
const skills = (profile?.sections?.skills?.items ?? [])
.slice(0, MAX_SKILLS)
.map((item) => {
const keywords = (item.keywords ?? []).slice(0, 8).join(", ");
return `${item.name}${keywords ? `: ${keywords}` : ""}`;
});
const projects = (profile?.sections?.projects?.items ?? [])
.filter((item) => item.visible !== false)
.slice(0, MAX_PROJECTS)
.map(
(item) =>
`${item.name} (${item.date || "n/a"}): ${truncate(item.summary, MAX_ITEM_TEXT)}`,
);
const experience = (profile?.sections?.experience?.items ?? [])
.filter((item) => item.visible !== false)
.slice(0, MAX_EXPERIENCE)
.map(
(item) =>
`${item.position} @ ${item.company} (${item.date || "n/a"}): ${truncate(item.summary, MAX_ITEM_TEXT)}`,
);
return compactJoin([
`Name: ${profile?.basics?.name || "Unknown"}`,
`Headline: ${truncate(profile?.basics?.headline || profile?.basics?.label, 200) || ""}`,
summary ? `Summary:\n${summary}` : null,
skills.length > 0 ? `Skills:\n- ${skills.join("\n- ")}` : null,
projects.length > 0 ? `Projects:\n- ${projects.join("\n- ")}` : null,
experience.length > 0 ? `Experience:\n- ${experience.join("\n- ")}` : null,
]);
}
function buildSystemPrompt(style: JobChatStyle): string {
return compactJoin([
"You are Ghostwriter, a job-application writing assistant for a single job.",
"Use only the provided job and profile context unless the user gives extra details.",
"Do not claim actions were executed. You are read-only and advisory.",
"If details are missing, say what is missing before making assumptions.",
"Avoid exposing private profile details that are unrelated to the user request.",
`Writing style tone: ${style.tone}.`,
`Writing style formality: ${style.formality}.`,
style.constraints ? `Writing constraints: ${style.constraints}` : null,
style.doNotUse ? `Avoid these terms: ${style.doNotUse}` : null,
]);
}
async function resolveStyle(): Promise<JobChatStyle> {
const overrides = await settingsRepo.getAllSettings();
const tone = resolveSettingValue(
"chatStyleTone",
overrides.chatStyleTone,
).value;
const formality = resolveSettingValue(
"chatStyleFormality",
overrides.chatStyleFormality,
).value;
const constraints = resolveSettingValue(
"chatStyleConstraints",
overrides.chatStyleConstraints,
).value;
const doNotUse = resolveSettingValue(
"chatStyleDoNotUse",
overrides.chatStyleDoNotUse,
).value;
return {
tone,
formality,
constraints,
doNotUse,
};
}
export async function buildJobChatPromptContext(
jobId: string,
): Promise<JobChatPromptContext> {
const job = await jobsRepo.getJobById(jobId);
if (!job) {
throw notFound("Job not found");
}
const style = await resolveStyle();
let profile: ResumeProfile = {};
try {
profile = await getProfile();
} catch (error) {
logger.warn("Failed to load profile for job chat context", {
jobId,
error: sanitizeUnknown(error),
});
}
const systemPrompt = buildSystemPrompt(style);
const jobSnapshot = buildJobSnapshot(job);
const profileSnapshot = buildProfileSnapshot(profile);
if (!jobSnapshot.trim()) {
throw badRequest("Unable to build job context");
}
logger.info("Built job chat context", {
jobId,
includesProfile: Boolean(profileSnapshot),
contextStats: sanitizeUnknown({
systemChars: systemPrompt.length,
jobChars: jobSnapshot.length,
profileChars: profileSnapshot.length,
}),
});
return {
job,
style,
systemPrompt,
jobSnapshot,
profileSnapshot,
};
}

498
orchestrator/src/server/services/job-chat.ts
View File

@ -1,498 +0,0 @@
import { logger } from "@infra/logger";
import { getRequestId } from "@infra/request-context";
import {
badRequest,
conflict,
notFound,
requestTimeout,
upstreamError,
} from "../infra/errors";
import * as jobChatRepo from "../repositories/job-chat";
import * as settingsRepo from "../repositories/settings";
import { buildJobChatPromptContext } from "./job-chat-context";
import { LlmService } from "./llm/service";
import type { JsonSchemaDefinition } from "./llm/types";
type LlmRuntimeSettings = {
model: string;
provider: string | null;
baseUrl: string | null;
apiKey: string | null;
};
const abortControllers = new Map<string, AbortController>();
const CHAT_RESPONSE_SCHEMA: JsonSchemaDefinition = {
name: "job_chat_response",
schema: {
type: "object",
properties: {
response: {
type: "string",
},
},
required: ["response"],
additionalProperties: false,
},
};
function estimateTokenCount(value: string): number {
if (!value) return 0;
return Math.ceil(value.length / 4);
}
function chunkText(value: string, maxChunk = 60): string[] {
if (!value) return [];
const chunks: string[] = [];
let cursor = 0;
while (cursor < value.length) {
chunks.push(value.slice(cursor, cursor + maxChunk));
cursor += maxChunk;
}
return chunks;
}
async function resolveLlmRuntimeSettings(): Promise<LlmRuntimeSettings> {
const overrides = await settingsRepo.getAllSettings();
const model =
overrides.modelTailoring ||
overrides.model ||
process.env.MODEL ||
"google/gemini-3-flash-preview";
const provider =
overrides.llmProvider || process.env.LLM_PROVIDER || "openrouter";
const baseUrl = overrides.llmBaseUrl || process.env.LLM_BASE_URL || null;
const apiKey = overrides.llmApiKey || process.env.LLM_API_KEY || null;
return {
model,
provider,
baseUrl,
apiKey,
};
}
async function buildConversationMessages(
threadId: string,
): Promise<Array<{ role: "user" | "assistant"; content: string }>> {
const messages = await jobChatRepo.listMessagesForThread(threadId, {
limit: 40,
});
return messages
.filter(
(message): message is typeof message & { role: "user" | "assistant" } =>
message.role === "user" || message.role === "assistant",
)
.filter((message) => message.status !== "failed")
.map((message) => ({
role: message.role,
content: message.content,
}));
}
type GenerateReplyOptions = {
jobId: string;
threadId: string;
prompt: string;
replaceMessageId?: string;
version?: number;
stream?: {
onReady: (payload: {
runId: string;
messageId: string;
requestId: string;
}) => void;
onDelta: (payload: {
runId: string;
messageId: string;
delta: string;
}) => void;
onCompleted: (payload: {
runId: string;
message: Awaited<ReturnType<typeof jobChatRepo.getMessageById>>;
}) => void;
onCancelled: (payload: {
runId: string;
message: Awaited<ReturnType<typeof jobChatRepo.getMessageById>>;
}) => void;
onError: (payload: {
runId: string;
code: string;
message: string;
requestId: string;
}) => void;
};
};
export async function createThread(input: {
jobId: string;
title?: string | null;
}) {
return jobChatRepo.createThread(input);
}
export async function listThreads(jobId: string) {
return jobChatRepo.listThreadsForJob(jobId);
}
export async function listMessages(input: {
jobId: string;
threadId: string;
limit?: number;
offset?: number;
}) {
const thread = await jobChatRepo.getThreadForJob(input.jobId, input.threadId);
if (!thread) {
throw notFound("Thread not found for this job");
}
return jobChatRepo.listMessagesForThread(input.threadId, {
limit: input.limit,
offset: input.offset,
});
}
async function runAssistantReply(
options: GenerateReplyOptions,
): Promise<{ runId: string; messageId: string; message: string }> {
const thread = await jobChatRepo.getThreadForJob(
options.jobId,
options.threadId,
);
if (!thread) {
throw notFound("Thread not found for this job");
}
const activeRun = await jobChatRepo.getActiveRunForThread(options.threadId);
if (activeRun) {
throw conflict("A chat generation is already running for this thread");
}
const [context, llmConfig, history] = await Promise.all([
buildJobChatPromptContext(options.jobId),
resolveLlmRuntimeSettings(),
buildConversationMessages(options.threadId),
]);
const requestId = getRequestId() ?? "unknown";
const assistantMessage = await jobChatRepo.createMessage({
threadId: options.threadId,
jobId: options.jobId,
role: "assistant",
content: "",
status: "partial",
version: options.version ?? 1,
replacesMessageId: options.replaceMessageId ?? null,
});
const run = await jobChatRepo.createRun({
threadId: options.threadId,
jobId: options.jobId,
model: llmConfig.model,
provider: llmConfig.provider,
requestId,
});
const controller = new AbortController();
abortControllers.set(run.id, controller);
options.stream?.onReady({
runId: run.id,
messageId: assistantMessage.id,
requestId,
});
let accumulated = "";
try {
const llm = new LlmService({
provider: llmConfig.provider,
baseUrl: llmConfig.baseUrl,
apiKey: llmConfig.apiKey,
});
const llmResult = await llm.callJson<{ response: string }>({
model: llmConfig.model,
messages: [
{
role: "system",
content: context.systemPrompt,
},
{
role: "system",
content: `Job Context (JSON):\n${context.jobSnapshot}`,
},
{
role: "system",
content: `Profile Context:\n${context.profileSnapshot || "No profile context available."}`,
},
...history,
{
role: "user",
content: options.prompt,
},
],
jsonSchema: CHAT_RESPONSE_SCHEMA,
maxRetries: 1,
retryDelayMs: 300,
jobId: options.jobId,
signal: controller.signal,
});
if (!llmResult.success) {
if (controller.signal.aborted) {
throw requestTimeout("Chat generation was cancelled");
}
throw upstreamError("LLM generation failed", {
reason: llmResult.error,
});
}
const finalText = (llmResult.data.response || "").trim();
const chunks = chunkText(finalText);
for (const chunk of chunks) {
if (controller.signal.aborted) {
const cancelled = await jobChatRepo.updateMessage(assistantMessage.id, {
content: accumulated,
status: "cancelled",
tokensIn: estimateTokenCount(options.prompt),
tokensOut: estimateTokenCount(accumulated),
});
await jobChatRepo.completeRun(run.id, {
status: "cancelled",
errorCode: "REQUEST_TIMEOUT",
errorMessage: "Generation cancelled by user",
});
options.stream?.onCancelled({ runId: run.id, message: cancelled });
return {
runId: run.id,
messageId: assistantMessage.id,
message: accumulated,
};
}
accumulated += chunk;
options.stream?.onDelta({
runId: run.id,
messageId: assistantMessage.id,
delta: chunk,
});
}
const completedMessage = await jobChatRepo.updateMessage(
assistantMessage.id,
{
content: accumulated,
status: "complete",
tokensIn: estimateTokenCount(options.prompt),
tokensOut: estimateTokenCount(accumulated),
},
);
await jobChatRepo.completeRun(run.id, {
status: "completed",
});
options.stream?.onCompleted({
runId: run.id,
message: completedMessage,
});
return {
runId: run.id,
messageId: assistantMessage.id,
message: accumulated,
};
} catch (error) {
const appError = error instanceof Error ? error : new Error(String(error));
const isCancelled =
controller.signal.aborted || appError.name === "AbortError";
const status = isCancelled ? "cancelled" : "failed";
const code = isCancelled ? "REQUEST_TIMEOUT" : "UPSTREAM_ERROR";
const message = isCancelled
? "Generation cancelled by user"
: appError.message || "Generation failed";
const failedMessage = await jobChatRepo.updateMessage(assistantMessage.id, {
content: accumulated,
status: isCancelled ? "cancelled" : "failed",
tokensIn: estimateTokenCount(options.prompt),
tokensOut: estimateTokenCount(accumulated),
});
await jobChatRepo.completeRun(run.id, {
status,
errorCode: code,
errorMessage: message,
});
if (isCancelled) {
options.stream?.onCancelled({ runId: run.id, message: failedMessage });
return {
runId: run.id,
messageId: assistantMessage.id,
message: accumulated,
};
}
options.stream?.onError({
runId: run.id,
code,
message,
requestId,
});
throw upstreamError(message, { runId: run.id });
} finally {
abortControllers.delete(run.id);
logger.info("Job chat run finished", {
jobId: options.jobId,
threadId: options.threadId,
runId: run.id,
});
}
}
export async function sendMessage(input: {
jobId: string;
threadId: string;
content: string;
stream?: GenerateReplyOptions["stream"];
}) {
const content = input.content.trim();
if (!content) {
throw badRequest("Message content is required");
}
const thread = await jobChatRepo.getThreadForJob(input.jobId, input.threadId);
if (!thread) {
throw notFound("Thread not found for this job");
}
const userMessage = await jobChatRepo.createMessage({
threadId: input.threadId,
jobId: input.jobId,
role: "user",
content,
status: "complete",
tokensIn: estimateTokenCount(content),
tokensOut: null,
});
const result = await runAssistantReply({
jobId: input.jobId,
threadId: input.threadId,
prompt: content,
stream: input.stream,
});
const assistantMessage = await jobChatRepo.getMessageById(result.messageId);
return {
userMessage,
assistantMessage,
runId: result.runId,
};
}
export async function regenerateMessage(input: {
jobId: string;
threadId: string;
assistantMessageId: string;
stream?: GenerateReplyOptions["stream"];
}) {
const thread = await jobChatRepo.getThreadForJob(input.jobId, input.threadId);
if (!thread) {
throw notFound("Thread not found for this job");
}
const target = await jobChatRepo.getMessageById(input.assistantMessageId);
if (
!target ||
target.threadId !== input.threadId ||
target.jobId !== input.jobId
) {
throw notFound("Assistant message not found for this thread");
}
if (target.role !== "assistant") {
throw badRequest("Only assistant messages can be regenerated");
}
const latestAssistant = await jobChatRepo.getLatestAssistantMessage(
input.threadId,
);
if (!latestAssistant || latestAssistant.id !== target.id) {
throw badRequest("Only the latest assistant message can be regenerated");
}
const messages = await jobChatRepo.listMessagesForThread(input.threadId, {
limit: 200,
});
const targetIndex = messages.findIndex((message) => message.id === target.id);
const priorUser =
targetIndex > 0
? [...messages.slice(0, targetIndex)]
.reverse()
.find((message) => message.role === "user")
: null;
if (!priorUser) {
throw badRequest("Could not find a user message to regenerate from");
}
const result = await runAssistantReply({
jobId: input.jobId,
threadId: input.threadId,
prompt: priorUser.content,
replaceMessageId: target.id,
version: (target.version || 1) + 1,
stream: input.stream,
});
const assistantMessage = await jobChatRepo.getMessageById(result.messageId);
return {
runId: result.runId,
assistantMessage,
};
}
export async function cancelRun(input: {
jobId: string;
threadId: string;
runId: string;
}): Promise<{ cancelled: boolean; alreadyFinished: boolean }> {
const run = await jobChatRepo.getRunById(input.runId);
if (!run || run.threadId !== input.threadId || run.jobId !== input.jobId) {
throw notFound("Run not found for this thread");
}
if (run.status !== "running") {
return {
cancelled: false,
alreadyFinished: true,
};
}
const controller = abortControllers.get(input.runId);
if (controller) {
controller.abort();
}
await jobChatRepo.completeRun(input.runId, {
status: "cancelled",
errorCode: "REQUEST_TIMEOUT",
errorMessage: "Generation cancelled by user",
});
return {
cancelled: true,
alreadyFinished: false,
};
}