Archived
@@ -0,0 +1,5 @@
|
|||||||
|
---
|
||||||
|
"@jmfederico/pi-web": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
Prevent malformed session prompt API calls from crashing the session daemon.
|
||||||
@@ -366,6 +366,20 @@ describe("PiSessionService", () => {
|
|||||||
await service.dispose();
|
await service.dispose();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("rejects malformed prompt text before opening the runtime", async () => {
|
||||||
|
const fake = fakeRuntime("prompt-session");
|
||||||
|
const service = new PiSessionService(new CapturingSessionEventHub(), {
|
||||||
|
createAgentRuntime: runtimeCreator(fake.runtime),
|
||||||
|
sessionManager: sessionGateway([sessionRecord("prompt-session")]),
|
||||||
|
heartbeatIntervalMs: 60_000,
|
||||||
|
});
|
||||||
|
|
||||||
|
await expect(service.prompt("prompt-session", undefined)).rejects.toThrow("Prompt text is required");
|
||||||
|
|
||||||
|
expect(fake.calls.prompt).toEqual([]);
|
||||||
|
await service.dispose();
|
||||||
|
});
|
||||||
|
|
||||||
it("includes queued message details in session status", async () => {
|
it("includes queued message details in session status", async () => {
|
||||||
const fake = fakeRuntime("status-session", {
|
const fake = fakeRuntime("status-session", {
|
||||||
messages: [{ role: "user", content: "hello" }, { role: "assistant", content: "hi" }],
|
messages: [{ role: "user", content: "hello" }, { role: "assistant", content: "hi" }],
|
||||||
|
|||||||
@@ -41,6 +41,17 @@ interface QueuedPrompt {
|
|||||||
text: string;
|
text: string;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function requirePromptText(value: unknown): string {
|
||||||
|
if (typeof value !== "string") throw new Error("Prompt text is required");
|
||||||
|
return value;
|
||||||
|
}
|
||||||
|
|
||||||
|
function parsePromptStreamingBehavior(value: unknown): QueuedPromptKind | undefined {
|
||||||
|
if (value === undefined) return undefined;
|
||||||
|
if (value === "steer" || value === "followUp") return value;
|
||||||
|
throw new Error('Prompt streamingBehavior must be "steer" or "followUp"');
|
||||||
|
}
|
||||||
|
|
||||||
type SessionArchiveRepository = Pick<SessionArchiveStore, "list" | "get" | "archive" | "restore" | "isArchived">;
|
type SessionArchiveRepository = Pick<SessionArchiveStore, "list" | "get" | "archive" | "restore" | "isArchived">;
|
||||||
interface PiSessionListEntry {
|
interface PiSessionListEntry {
|
||||||
id: string;
|
id: string;
|
||||||
@@ -360,22 +371,24 @@ export class PiSessionService {
|
|||||||
return commands.sort((a, b) => a.name.localeCompare(b.name));
|
return commands.sort((a, b) => a.name.localeCompare(b.name));
|
||||||
}
|
}
|
||||||
|
|
||||||
async prompt(sessionId: string, text: string, streamingBehavior?: "steer" | "followUp"): Promise<void> {
|
async prompt(sessionId: string, text: unknown, streamingBehavior?: unknown): Promise<void> {
|
||||||
|
const promptText = requirePromptText(text);
|
||||||
|
const requestedBehavior = parsePromptStreamingBehavior(streamingBehavior);
|
||||||
await this.assertWritable(sessionId);
|
await this.assertWritable(sessionId);
|
||||||
const session = await this.getOrOpen(sessionId);
|
const session = await this.getOrOpen(sessionId);
|
||||||
this.maybeGenerateSessionName(session, text);
|
this.maybeGenerateSessionName(session, promptText);
|
||||||
const isQueued = session.isStreaming || session.isCompacting;
|
const isQueued = session.isStreaming || session.isCompacting;
|
||||||
const behavior = isQueued ? streamingBehavior ?? "followUp" : undefined;
|
const behavior = isQueued ? requestedBehavior ?? "followUp" : undefined;
|
||||||
if (isQueued && this.hasQueuedMessageText(session, text)) {
|
if (isQueued && this.hasQueuedMessageText(session, promptText)) {
|
||||||
this.publishActivity(session, "duplicate queued message ignored", "active");
|
this.publishActivity(session, "duplicate queued message ignored", "active");
|
||||||
this.publishStatus(session);
|
this.publishStatus(session);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
if (session.isCompacting) {
|
if (session.isCompacting) {
|
||||||
this.enqueuePromptDuringCompaction(session, text, behavior ?? "followUp");
|
this.enqueuePromptDuringCompaction(session, promptText, behavior ?? "followUp");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
void this.submitPrompt(session, text, behavior);
|
void this.submitPrompt(session, promptText, behavior);
|
||||||
}
|
}
|
||||||
|
|
||||||
private submitPrompt(session: PiAgentSession, text: string, behavior: QueuedPromptKind | undefined): Promise<void> {
|
private submitPrompt(session: PiAgentSession, text: string, behavior: QueuedPromptKind | undefined): Promise<void> {
|
||||||
|
|||||||
@@ -15,4 +15,8 @@ describe("sessionNameGenerator", () => {
|
|||||||
expect(fallbackSessionName('<skill name="x" location="/x">\nDo x\n</skill>\n\nCheck the UI now'))
|
expect(fallbackSessionName('<skill name="x" location="/x">\nDo x\n</skill>\n\nCheck the UI now'))
|
||||||
.toBe("Check the UI now");
|
.toBe("Check the UI now");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("skips fallback names when the first request is missing", () => {
|
||||||
|
expect(fallbackSessionName(undefined)).toBeUndefined();
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -43,7 +43,9 @@ export async function generateShortSessionName<TApi extends Api>(modelRegistry:
|
|||||||
return cleanSessionName(finalMessage === undefined ? streamedText : textFromAssistant(finalMessage));
|
return cleanSessionName(finalMessage === undefined ? streamedText : textFromAssistant(finalMessage));
|
||||||
}
|
}
|
||||||
|
|
||||||
export function fallbackSessionName(firstMessage: string): string | undefined {
|
export function fallbackSessionName(firstMessage: unknown): string | undefined {
|
||||||
|
if (typeof firstMessage !== "string") return undefined;
|
||||||
|
|
||||||
return cleanSessionName(firstMessage
|
return cleanSessionName(firstMessage
|
||||||
.replace(/<skill name="[^"]+" location="[^"]+">[\s\S]*?<\/skill>/g, "")
|
.replace(/<skill name="[^"]+" location="[^"]+">[\s\S]*?<\/skill>/g, "")
|
||||||
.replace(/```[\s\S]*?```/g, " ")
|
.replace(/```[\s\S]*?```/g, " ")
|
||||||
|
|||||||
@@ -0,0 +1,58 @@
|
|||||||
|
import Fastify, { type FastifyInstance } from "fastify";
|
||||||
|
import fastifyWebsocket from "@fastify/websocket";
|
||||||
|
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||||
|
import { SessionEventHub } from "../realtime/sessionEventHub.js";
|
||||||
|
import { PiSessionService, type PiSessionManagerGateway } from "./piSessionService.js";
|
||||||
|
import { registerSessionRoutes } from "./sessionRoutes.js";
|
||||||
|
|
||||||
|
let app: FastifyInstance;
|
||||||
|
let service: PiSessionService;
|
||||||
|
let sessionManager: RejectingSessionManager;
|
||||||
|
|
||||||
|
beforeEach(async () => {
|
||||||
|
app = Fastify({ logger: false });
|
||||||
|
await app.register(fastifyWebsocket);
|
||||||
|
sessionManager = new RejectingSessionManager();
|
||||||
|
const eventHub = new SessionEventHub();
|
||||||
|
service = new PiSessionService(eventHub, { sessionManager, heartbeatIntervalMs: 60_000 });
|
||||||
|
registerSessionRoutes(app, service, eventHub);
|
||||||
|
});
|
||||||
|
|
||||||
|
afterEach(async () => {
|
||||||
|
await service.dispose();
|
||||||
|
await app.close();
|
||||||
|
});
|
||||||
|
|
||||||
|
describe("session routes", () => {
|
||||||
|
it("rejects prompt payloads that omit text without opening a session", async () => {
|
||||||
|
const response = await app.inject({ method: "POST", url: "/sessions/session-1/prompt", payload: { body: "Build the thing" } });
|
||||||
|
|
||||||
|
expect(response.statusCode).toBe(400);
|
||||||
|
expect(response.json()).toEqual({ error: "Prompt text is required" });
|
||||||
|
expect(sessionManager.calls).toEqual({ create: 0, list: 0, listAll: 0, open: 0 });
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
class RejectingSessionManager implements PiSessionManagerGateway {
|
||||||
|
readonly calls = { create: 0, list: 0, listAll: 0, open: 0 };
|
||||||
|
|
||||||
|
list() {
|
||||||
|
this.calls.list += 1;
|
||||||
|
return Promise.resolve([]);
|
||||||
|
}
|
||||||
|
|
||||||
|
create(): never {
|
||||||
|
this.calls.create += 1;
|
||||||
|
throw new Error("Session manager should not create sessions for invalid prompt payloads");
|
||||||
|
}
|
||||||
|
|
||||||
|
listAll() {
|
||||||
|
this.calls.listAll += 1;
|
||||||
|
return Promise.resolve([]);
|
||||||
|
}
|
||||||
|
|
||||||
|
open(): never {
|
||||||
|
this.calls.open += 1;
|
||||||
|
throw new Error("Session manager should not open sessions for invalid prompt payloads");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -2,6 +2,11 @@ import type { FastifyInstance } from "fastify";
|
|||||||
import type { SessionEventHub } from "../realtime/sessionEventHub.js";
|
import type { SessionEventHub } from "../realtime/sessionEventHub.js";
|
||||||
import type { PiSessionService } from "./piSessionService.js";
|
import type { PiSessionService } from "./piSessionService.js";
|
||||||
|
|
||||||
|
interface PromptRequestBody {
|
||||||
|
text?: unknown;
|
||||||
|
streamingBehavior?: unknown;
|
||||||
|
}
|
||||||
|
|
||||||
export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionService, eventHub: SessionEventHub, prefix = ""): void {
|
export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionService, eventHub: SessionEventHub, prefix = ""): void {
|
||||||
app.get<{ Querystring: { cwd?: string } }>(`${prefix}/sessions`, async (request, reply) => {
|
app.get<{ Querystring: { cwd?: string } }>(`${prefix}/sessions`, async (request, reply) => {
|
||||||
if (request.query.cwd === undefined || request.query.cwd === "") return reply.code(400).send({ error: "cwd query parameter is required" });
|
if (request.query.cwd === undefined || request.query.cwd === "") return reply.code(400).send({ error: "cwd query parameter is required" });
|
||||||
@@ -89,9 +94,9 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionS
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
app.post<{ Params: { sessionId: string }; Body: { text: string; streamingBehavior?: "steer" | "followUp" } }>(`${prefix}/sessions/:sessionId/prompt`, async (request, reply) => {
|
app.post<{ Params: { sessionId: string }; Body: PromptRequestBody | undefined }>(`${prefix}/sessions/:sessionId/prompt`, async (request, reply) => {
|
||||||
try {
|
try {
|
||||||
await sessions.prompt(request.params.sessionId, request.body.text, request.body.streamingBehavior);
|
await sessions.prompt(request.params.sessionId, request.body?.text, request.body?.streamingBehavior);
|
||||||
return { accepted: true };
|
return { accepted: true };
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) });
|
return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) });
|
||||||
|
|||||||
Reference in New Issue
Block a user