18 KiB
18 KiB
| phase | plan | type | wave | depends_on | files_modified | autonomous | requirements | must_haves | ||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 22-agent-streaming | 01 | execute | 1 |
|
true |
|
|
Purpose: Establishes the entire server-side API surface that the UI plans (02/03) will consume. Every new endpoint is tested. Output: Working SSE stream endpoint (echo-stream mode), edit message endpoint, conversation agent update, migration SQL, tests.
NOTE -- Echo-stream scope (CHAT-01 partial): The SSE endpoint uses an echo-stream that replays the user's last message word-by-word. This fully exercises the streaming pipeline (SSE headers, token events, done event, abort detection, message persistence) so the UI can be built and tested against real streaming behavior. Real LLM integration (replacing the echo loop with an adapter call) is Phase 23 (Brainstormer agent). The echo-stream satisfies CHAT-01's "tokens appear as generated" contract at the transport level; Phase 23 provides semantic content.
<execution_context> @$HOME/.claude/get-shit-done/workflows/execute-plan.md @$HOME/.claude/get-shit-done/templates/summary.md </execution_context>
@.planning/PROJECT.md @.planning/ROADMAP.md @.planning/STATE.md @.planning/phases/22-agent-streaming/22-RESEARCH.mdFrom packages/shared/src/types/chat.ts:
export interface ChatConversation {
id: string;
companyId: string;
title: string | null;
agentId: string | null;
pinnedAt: string | null;
archivedAt: string | null;
deletedAt: string | null;
createdAt: string;
updatedAt: string;
}
export interface ChatMessage {
id: string;
conversationId: string;
role: "user" | "assistant" | "system";
content: string;
agentId: string | null;
createdAt: string;
}
From packages/shared/src/validators/chat.ts:
export const createConversationSchema = z.object({ title: z.string().max(200).optional() });
export const updateConversationSchema = z.object({ title: z.string().max(200).optional() });
export const createMessageSchema = z.object({
role: z.enum(["user", "assistant", "system"]),
content: z.string().min(1),
agentId: z.string().uuid().optional().nullable(),
});
From packages/db/src/schema/chat_messages.ts:
export const chatMessages = pgTable("chat_messages", {
id: uuid("id").primaryKey().defaultRandom(),
conversationId: uuid("conversation_id").notNull().references(() => chatConversations.id, { onDelete: "cascade" }),
role: text("role").notNull(),
content: text("content").notNull(),
agentId: uuid("agent_id"),
createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
}, ...);
From server/src/services/chat.ts:
export function chatService(db: Db) {
// Returns object with: listConversations, createConversation, getConversation,
// updateConversation, softDeleteConversation, archiveConversation, unarchiveConversation,
// pinConversation, unpinConversation, listMessages, addMessage
}
From server/src/routes/chat.ts:
export function chatRoutes(db: Db) {
// Mounts all routes on a Router. Key: PATCH /conversations/:id uses validate(updateConversationSchema)
}
SSE pattern from server/src/routes/plugins.ts:1146:
res.writeHead(200, {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
});
res.flushHeaders();
res.write(":ok\n\n");
2. **Shared types** -- Update `ChatMessage` interface in `packages/shared/src/types/chat.ts`:
- Add `editedContent: string | null;`
- Add `editedAt: string | null;`
3. **Validators** -- In `packages/shared/src/validators/chat.ts`:
- Update `updateConversationSchema` to include `agentId: z.string().uuid().optional().nullable()`
- Add `export const editMessageSchema = z.object({ content: z.string().min(1) });`
- Add `export const streamMessageSchema = z.object({ content: z.string().min(1), agentId: z.string().uuid().optional().nullable() });`
4. **Service methods** -- Add to `chatService` in `server/src/services/chat.ts`:
- `editMessage(messageId: string, data: { content: string })` -- sets `editedContent = data.content`, `editedAt = new Date()` on the message row, returns the updated row
- `getMessageHistory(conversationId: string)` -- selects all messages WHERE conversationId matches, ORDER BY createdAt ASC (ascending, for LLM context window). Returns `ChatMessage[]`. Use `editedContent ?? content` as the effective content field (alias as `effectiveContent` in the return).
- Update `updateConversation` to accept and persist `agentId` field: `set({ title: data.title, agentId: data.agentId, updatedAt: new Date() })`. Only set fields that are provided (check `data.agentId !== undefined` before including in set).
5. **Extend existing tests** in `server/src/__tests__/chat-routes.test.ts`:
- Add test: `PATCH /conversations/:id with agentId` -- create conversation, PATCH with `{ agentId: someAgentId }`, verify response has the agentId set. (Use a dummy UUID string for agentId if the test DB doesn't enforce FK -- check existing test patterns.)
- Add test: `PUT /conversations/:id/messages/:messageId` -- create conversation, add message, PUT with `{ content: "edited" }`, verify response has `editedContent: "edited"` and `editedAt` is not null.
1. **Edit message route** -- Add to `server/src/routes/chat.ts`:
```typescript
// PUT /conversations/:id/messages/:messageId
router.put("/conversations/:id/messages/:messageId", validate(editMessageSchema), async (req, res) => {
assertBoard(req);
const message = await svc.editMessage(req.params.messageId as string, req.body);
if (!message) {
res.status(404).json({ error: "Not found" });
return;
}
res.json(message);
});
```
2. **SSE stream endpoint** -- Add to `server/src/routes/chat.ts`:
```typescript
// GET /conversations/:id/stream
router.get("/conversations/:id/stream", async (req, res) => {
assertBoard(req);
const conversationId = req.params.id as string;
const triggerMessageId = req.query.triggerMessageId as string | undefined;
const conversation = await svc.getConversation(conversationId);
if (!conversation) {
res.status(404).json({ error: "Not found" });
return;
}
// Set SSE headers -- copied from plugins.ts:1146
res.writeHead(200, {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
});
res.flushHeaders();
res.write(":ok\n\n");
let aborted = false;
req.on("close", () => { aborted = true; });
// Resolve the agent for this conversation
const agentId = conversation.agentId;
// Get message history for LLM context
const history = await svc.getMessageHistory(conversationId);
// ECHO-STREAM PLACEHOLDER (Phase 22):
// Streams the user's last message back word-by-word to fully exercise the SSE
// pipeline. Phase 23 replaces this block with:
// const adapter = resolveAdapter(agentId);
// for await (const token of adapter.stream(history)) { ... }
const lastUserMsg = history.filter(m => m.role === "user").at(-1);
const echoContent = lastUserMsg
? `Echo from agent: ${lastUserMsg.content}`
: "No message to echo.";
const tokens = echoContent.split(/(\s+)/);
let accumulated = "";
for (const token of tokens) {
if (aborted) break;
accumulated += token;
res.write(`data: ${JSON.stringify({ type: "token", content: token })}\n\n`);
// Tiny yield to allow abort detection
await new Promise(resolve => setTimeout(resolve, 5));
}
// Persist assistant message only if stream completed (not aborted)
if (!aborted && accumulated.trim()) {
const assistantMsg = await svc.addMessage(conversationId, {
role: "assistant",
content: accumulated,
agentId,
});
res.write(`data: ${JSON.stringify({ type: "done", messageId: assistantMsg.id })}\n\n`);
} else if (aborted) {
// Do NOT persist partial messages per RESEARCH.md pitfall 4
}
res.end();
});
```
Import `editMessageSchema` and `streamMessageSchema` from `@paperclipai/shared` at the top of the routes file (alongside existing imports).
3. **Stream tests** -- Create `server/src/__tests__/chat-stream-routes.test.ts`:
- Use the same test DB setup pattern as `chat-routes.test.ts` (read that file for the pattern).
- Test: `GET /conversations/:id/stream?triggerMessageId=X` -- create conversation, add user message, open stream, collect all SSE data events, verify:
- Response status is 200
- Content-Type header contains "text/event-stream"
- X-Accel-Buffering header is "no"
- First received data is `:ok` comment (or first data event has type "token")
- Last data event has `type: "done"` with a `messageId` string
- Test: `GET /conversations/:id/stream` for non-existent conversation returns 404
- Test: After stream completes, a new assistant message exists in the DB (query via list messages)
- Test: `PUT /conversations/:id/messages/:messageId` with valid body returns 200 and editedContent matches
For SSE testing: use supertest's `.buffer(true).parse(...)` or collect the raw response body. Alternatively, make a raw HTTP request to the test server and read the stream. Follow whatever pattern the existing test file uses for HTTP calls.
4. Add the `editMessageSchema` and `streamMessageSchema` imports to the routes file's import block from `@paperclipai/shared`.
<success_criteria>
- New migration SQL exists and applies the editedContent + editedAt columns
- GET /conversations/:id/stream returns text/event-stream with token events then done event (echo-stream placeholder -- Phase 23 replaces with real LLM)
- PUT /conversations/:id/messages/:messageId updates editedContent and editedAt
- PATCH /conversations/:id with { agentId } persists the agent selection
- All server tests pass (both chat-routes and chat-stream-routes) </success_criteria>