jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: update sidebar agent workflow and host filter UI

Browse Source
This commit is contained in:
Mohamed Boudra
2026-02-15 08:59:17 +07:00
parent 9da5b26db0
commit 41d1fa6d97
31 changed files with 2482 additions and 1133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/client/daemon-client.test.ts
+32 -7
View File
@@ -455,7 +455,7 @@ describe("DaemonClient", () => {
expect(request.message.requestId.length).toBeGreaterThan(0);
});
test("fetches project-grouped agents via RPC", async () => {
test("fetches agents via RPC with filters, sort, and pagination", async () => {
const logger = createMockLogger();
const mock = createMockTransport();
@@ -471,29 +471,49 @@ describe("DaemonClient", () => {
mock.triggerOpen();
await connectPromise;
const promise = client.fetchAgentsGroupedByProject({
const promise = client.fetchAgents({
filter: { labels: { ui: "true" } },
sort: [
{ key: "status_priority", direction: "asc" },
{ key: "created_at", direction: "desc" },
],
page: { limit: 25, cursor: "cursor-1" },
});
expect(mock.sent).toHaveLength(1);
const request = JSON.parse(mock.sent[0]) as {
type: "session";
message: {
type: "fetch_agents_grouped_by_project_request";
type: "fetch_agents_request";
requestId: string;
filter?: { labels?: Record<string, string> };
sort?: Array<{
key: "status_priority" | "created_at" | "updated_at" | "title";
direction: "asc" | "desc";
}>;
page?: { limit: number; cursor?: string };
};
};
expect(request.message.type).toBe("fetch_agents_grouped_by_project_request");
expect(request.message.type).toBe("fetch_agents_request");
expect(request.message.sort).toEqual([
{ key: "status_priority", direction: "asc" },
{ key: "created_at", direction: "desc" },
]);
expect(request.message.page).toEqual({ limit: 25, cursor: "cursor-1" });
mock.triggerMessage(
JSON.stringify({
type: "session",
message: {
type: "fetch_agents_grouped_by_project_response",
type: "fetch_agents_response",
payload: {
requestId: request.message.requestId,
groups: [],
entries: [],
pageInfo: {
nextCursor: null,
prevCursor: "cursor-1",
hasMore: false,
},
},
},
})
@@ -501,7 +521,12 @@ describe("DaemonClient", () => {
await expect(promise).resolves.toEqual({
requestId: request.message.requestId,
groups: [],
entries: [],
pageInfo: {
nextCursor: null,
prevCursor: "cursor-1",
hasMore: false,
},
});
});
packages/server/src/client/daemon-client.ts
+14 -33
View File
@@ -244,10 +244,19 @@ type AgentRefreshedStatusPayload = z.infer<
type RestartRequestedStatusPayload = z.infer<
typeof RestartRequestedStatusPayloadSchema
>;
type FetchAgentsGroupedByProjectPayload = Extract<
type FetchAgentsPayload = Extract<
SessionOutboundMessage,
{ type: "fetch_agents_grouped_by_project_response" }
{ type: "fetch_agents_response" }
>["payload"];
type FetchAgentsRequest = Extract<
SessionInboundMessage,
{ type: "fetch_agents_request" }
>;
export type FetchAgentsOptions = Omit<FetchAgentsRequest, "type" | "requestId"> & {
requestId?: string;
};
export type FetchAgentsEntry = FetchAgentsPayload["entries"][number];
export type FetchAgentsPageInfo = FetchAgentsPayload["pageInfo"];
export type WaitForFinishResult = {
status: "idle" | "error" | "permission" | "timeout";
@@ -976,15 +985,14 @@ export class DaemonClient {
// Agent RPCs (requestId-correlated)
// ============================================================================
async fetchAgents(options?: {
filter?: { labels?: Record<string, string> };
requestId?: string;
}): Promise<AgentSnapshotPayload[]> {
async fetchAgents(options?: FetchAgentsOptions): Promise<FetchAgentsPayload> {
const resolvedRequestId = this.createRequestId(options?.requestId);
const message = SessionInboundMessageSchema.parse({
type: "fetch_agents_request",
requestId: resolvedRequestId,
...(options?.filter ? { filter: options.filter } : {}),
...(options?.sort ? { sort: options.sort } : {}),
...(options?.page ? { page: options.page } : {}),
});
return this.sendRequest({
requestId: resolvedRequestId,
@@ -998,33 +1006,6 @@ export class DaemonClient {
if (msg.payload.requestId !== resolvedRequestId) {
return null;
}
return msg.payload.agents;
},
});
}
async fetchAgentsGroupedByProject(options?: {
filter?: { labels?: Record<string, string> };
requestId?: string;
}): Promise<FetchAgentsGroupedByProjectPayload> {
const resolvedRequestId = this.createRequestId(options?.requestId);
const message = SessionInboundMessageSchema.parse({
type: "fetch_agents_grouped_by_project_request",
requestId: resolvedRequestId,
...(options?.filter ? { filter: options.filter } : {}),
});
return this.sendRequest({
requestId: resolvedRequestId,
message,
timeout: 15000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "fetch_agents_grouped_by_project_response") {
return null;
}
if (msg.payload.requestId !== resolvedRequestId) {
return null;
}
return msg.payload;
},
});
packages/server/src/server/agent/agent-manager.ts
+3
View File
@@ -953,6 +953,9 @@ export class AgentManager {
agent.pendingRun = streamForwarder;
agent.lifecycle = "running";
// Bump updatedAt when lifecycle changes so downstream consumers can
// deterministically order idle->running transitions.
agent.updatedAt = new Date();
self.emitState(agent);
return streamForwarder;
packages/server/src/server/daemon-e2e/terminal.e2e.test.ts
+38
View File
@@ -317,6 +317,44 @@ const shouldRun = !process.env.CI;
30000
);
test(
"emits terminal_stream_exit and removes terminal when shell exits",
async () => {
const cwd = tmpCwd();
const list = await ctx.client.listTerminals(cwd);
const terminalId = list.terminals[0].id;
const attach = await ctx.client.attachTerminalStream(terminalId, { rows: 24, cols: 80 });
expect(attach.error).toBeNull();
const streamId = attach.streamId!;
let sawExit = false;
const unsubscribeExit = ctx.client.on("terminal_stream_exit", (message) => {
if (message.type !== "terminal_stream_exit") {
return;
}
if (
message.payload.terminalId === terminalId &&
message.payload.streamId === streamId
) {
sawExit = true;
}
});
ctx.client.sendTerminalStreamKey(streamId, { key: "d", ctrl: true });
await waitForCondition(() => sawExit, 10000);
const next = await ctx.client.listTerminals(cwd);
expect(next.terminals).toHaveLength(1);
expect(next.terminals[0].id).not.toBe(terminalId);
unsubscribeExit();
rmSync(cwd, { recursive: true, force: true });
},
30000
);
test(
"replays detached terminal output from resume offset (scrollback continuity)",
async () => {
packages/server/src/server/session.ts
+376 -110
View File
@@ -28,6 +28,7 @@ import {
type ProjectPlacementPayload,
} from "./messages.js";
import type { TerminalManager } from "../terminal/terminal-manager.js";
import type { TerminalSession } from "../terminal/terminal.js";
import {
BinaryMuxChannel,
TerminalBinaryFlags,
@@ -148,8 +149,6 @@ const pendingAgentInitializations = new Map<string, Promise<ManagedAgent>>();
let restartRequested = false;
const DEFAULT_AGENT_PROVIDER = AGENT_PROVIDER_IDS[0];
const RESTART_EXIT_DELAY_MS = 250;
const PROJECT_PLACEMENT_CACHE_TTL_MS = 10_000;
const MAX_AGENTS_PER_PROJECT = 5;
const CHECKOUT_DIFF_WATCH_DEBOUNCE_MS = 150;
const CHECKOUT_DIFF_FALLBACK_REFRESH_MS = 5_000;
const TERMINAL_STREAM_WINDOW_BYTES = 256 * 1024;
@@ -280,6 +279,30 @@ type TerminalStreamPendingChunk = {
replay: boolean;
};
type FetchAgentsRequestMessage = Extract<
SessionInboundMessage,
{ type: "fetch_agents_request" }
>;
type FetchAgentsRequestSort = NonNullable<FetchAgentsRequestMessage["sort"]>[number];
type FetchAgentsResponsePayload = Extract<
SessionOutboundMessage,
{ type: "fetch_agents_response" }
>["payload"];
type FetchAgentsResponseEntry = FetchAgentsResponsePayload["entries"][number];
type FetchAgentsResponsePageInfo = FetchAgentsResponsePayload["pageInfo"];
type FetchAgentsCursor = {
sort: FetchAgentsRequestSort[];
values: Record<string, string | number | null>;
id: string;
};
class SessionRequestError extends Error {
constructor(readonly code: string, message: string) {
super(message);
this.name = "SessionRequestError";
}
}
const PCM_SAMPLE_RATE = 16000;
const PCM_CHANNELS = 1;
const PCM_BITS_PER_SAMPLE = 16;
@@ -524,10 +547,6 @@ export class Session {
filter?: { labels?: Record<string, string>; agentId?: string };
}
| null = null;
private readonly projectPlacementCache = new Map<
string,
{ expiresAt: number; promise: Promise<ProjectPlacementPayload> }
>();
private clientActivity: {
deviceType: "web" | "mobile";
focusedAgentId: string | null;
@@ -538,6 +557,7 @@ export class Session {
private readonly MOBILE_BACKGROUND_STREAM_GRACE_MS = 60_000;
private readonly terminalManager: TerminalManager | null;
private terminalSubscriptions: Map<string, () => void> = new Map();
private terminalExitSubscriptions: Map<string, () => void> = new Map();
private readonly terminalStreams = new Map<
number,
{
@@ -1111,21 +1131,6 @@ export class Session {
};
}
private getProjectPlacement(cwd: string): Promise<ProjectPlacementPayload> {
const now = Date.now();
const cached = this.projectPlacementCache.get(cwd);
if (cached && cached.expiresAt > now) {
return cached.promise;
}
const promise = this.buildProjectPlacement(cwd);
this.projectPlacementCache.set(cwd, {
expiresAt: now + PROJECT_PLACEMENT_CACHE_TTL_MS,
promise,
});
return promise;
}
private async forwardAgentUpdate(agent: ManagedAgent): Promise<void> {
try {
const subscription = this.agentUpdatesSubscription;
@@ -1137,7 +1142,7 @@ export class Session {
const matches = this.matchesAgentFilter(payload, subscription.filter);
if (matches) {
const project = await this.getProjectPlacement(payload.cwd);
const project = await this.buildProjectPlacement(payload.cwd);
this.emit({
type: "agent_update",
payload: { kind: "upsert", agent: payload, project },
@@ -1173,11 +1178,7 @@ export class Session {
break;
case "fetch_agents_request":
await this.handleFetchAgents(msg.requestId, msg.filter);
break;
case "fetch_agents_grouped_by_project_request":
await this.handleFetchAgentsGroupedByProject(msg.requestId, msg.filter);
await this.handleFetchAgents(msg);
break;
case "fetch_agent_request":
@@ -4876,102 +4877,320 @@ export class Session {
return this.buildStoredAgentPayload(record);
}
private async handleFetchAgents(
requestId: string,
filter?: { labels?: Record<string, string> }
): Promise<void> {
try {
const agents = await this.listAgentPayloads(filter);
this.emit({
type: "fetch_agents_response",
payload: { requestId, agents },
});
} catch (error) {
this.sessionLogger.error({ err: error }, "Failed to handle fetch_agents_request");
this.emit({
type: "fetch_agents_response",
payload: { requestId, agents: [] },
});
}
}
private async listAgentsGroupedByProjectPayload(filter?: {
labels?: Record<string, string>;
}): Promise<Array<{
projectKey: string;
projectName: string;
agents: Array<{
agent: AgentSnapshotPayload;
checkout: ProjectCheckoutLitePayload;
}>;
}>> {
const agents = await this.listAgentPayloads(filter);
const visibleAgents = agents
.filter((agent) => !agent.archivedAt)
.sort(
(left, right) =>
Date.parse(right.updatedAt || "") - Date.parse(left.updatedAt || "")
);
const grouped = new Map<
string,
{
projectKey: string;
projectName: string;
agents: Array<{
agent: AgentSnapshotPayload;
checkout: ProjectCheckoutLitePayload;
}>;
}
>();
// Warm project placement status for all visible roots up front to avoid serial N+1 latency.
for (const agent of visibleAgents) {
void this.getProjectPlacement(agent.cwd);
private normalizeFetchAgentsSort(
sort: FetchAgentsRequestSort[] | undefined
): FetchAgentsRequestSort[] {
const fallback: FetchAgentsRequestSort[] = [
{ key: "updated_at", direction: "desc" },
];
if (!sort || sort.length === 0) {
return fallback;
}
for (const agent of visibleAgents) {
const project = await this.getProjectPlacement(agent.cwd);
const projectKey = project.projectKey;
let group = grouped.get(projectKey);
if (!group) {
group = {
projectKey,
projectName: project.projectName,
agents: [],
};
grouped.set(projectKey, group);
}
if (group.agents.length >= MAX_AGENTS_PER_PROJECT) {
const deduped: FetchAgentsRequestSort[] = [];
const seen = new Set<string>();
for (const entry of sort) {
if (seen.has(entry.key)) {
continue;
}
group.agents.push({ agent, checkout: project.checkout });
seen.add(entry.key);
deduped.push(entry);
}
return Array.from(grouped.values());
return deduped.length > 0 ? deduped : fallback;
}
private async handleFetchAgentsGroupedByProject(
requestId: string,
filter?: { labels?: Record<string, string> }
private getStatusPriority(agent: AgentSnapshotPayload): number {
const requiresAttention = agent.requiresAttention ?? false;
const attentionReason = agent.attentionReason ?? null;
if (requiresAttention && attentionReason === "permission") {
return 0;
}
if (agent.status === "error" || attentionReason === "error") {
return 1;
}
if (agent.status === "running") {
return 2;
}
if (agent.status === "initializing") {
return 3;
}
return 4;
}
private getFetchAgentsSortValue(
entry: FetchAgentsResponseEntry,
key: FetchAgentsRequestSort["key"]
): string | number | null {
switch (key) {
case "status_priority":
return this.getStatusPriority(entry.agent);
case "created_at":
return Date.parse(entry.agent.createdAt);
case "updated_at":
return Date.parse(entry.agent.updatedAt);
case "title":
return entry.agent.title?.toLocaleLowerCase() ?? "";
}
}
private compareSortValues(
left: string | number | null,
right: string | number | null
): number {
if (left === right) {
return 0;
}
if (left === null) {
return -1;
}
if (right === null) {
return 1;
}
if (typeof left === "number" && typeof right === "number") {
return left < right ? -1 : 1;
}
return String(left).localeCompare(String(right));
}
private compareFetchAgentsEntries(
left: FetchAgentsResponseEntry,
right: FetchAgentsResponseEntry,
sort: FetchAgentsRequestSort[]
): number {
for (const spec of sort) {
const leftValue = this.getFetchAgentsSortValue(left, spec.key);
const rightValue = this.getFetchAgentsSortValue(right, spec.key);
const base = this.compareSortValues(leftValue, rightValue);
if (base === 0) {
continue;
}
return spec.direction === "asc" ? base : -base;
}
return left.agent.id.localeCompare(right.agent.id);
}
private encodeFetchAgentsCursor(
entry: FetchAgentsResponseEntry,
sort: FetchAgentsRequestSort[]
): string {
const values: Record<string, string | number | null> = {};
for (const spec of sort) {
values[spec.key] = this.getFetchAgentsSortValue(entry, spec.key);
}
return Buffer.from(
JSON.stringify({
sort,
values,
id: entry.agent.id,
}),
"utf8"
).toString("base64url");
}
private decodeFetchAgentsCursor(
cursor: string,
sort: FetchAgentsRequestSort[]
): FetchAgentsCursor {
let parsed: unknown;
try {
parsed = JSON.parse(Buffer.from(cursor, "base64url").toString("utf8"));
} catch {
throw new SessionRequestError("invalid_cursor", "Invalid fetch_agents cursor");
}
if (!parsed || typeof parsed !== "object") {
throw new SessionRequestError("invalid_cursor", "Invalid fetch_agents cursor");
}
const payload = parsed as {
sort?: unknown;
values?: unknown;
id?: unknown;
};
if (!Array.isArray(payload.sort) || typeof payload.id !== "string") {
throw new SessionRequestError("invalid_cursor", "Invalid fetch_agents cursor");
}
if (!payload.values || typeof payload.values !== "object") {
throw new SessionRequestError("invalid_cursor", "Invalid fetch_agents cursor");
}
const cursorSort: FetchAgentsRequestSort[] = [];
for (const item of payload.sort) {
if (
!item ||
typeof item !== "object" ||
typeof (item as { key?: unknown }).key !== "string" ||
typeof (item as { direction?: unknown }).direction !== "string"
) {
throw new SessionRequestError("invalid_cursor", "Invalid fetch_agents cursor");
}
const key = (item as { key: string }).key;
const direction = (item as { direction: string }).direction;
if (
(key !== "status_priority" &&
key !== "created_at" &&
key !== "updated_at" &&
key !== "title") ||
(direction !== "asc" && direction !== "desc")
) {
throw new SessionRequestError("invalid_cursor", "Invalid fetch_agents cursor");
}
cursorSort.push({ key, direction });
}
if (
cursorSort.length !== sort.length ||
cursorSort.some(
(entry, index) =>
entry.key !== sort[index]?.key ||
entry.direction !== sort[index]?.direction
)
) {
throw new SessionRequestError(
"invalid_cursor",
"fetch_agents cursor does not match current sort"
);
}
return {
sort: cursorSort,
values: payload.values as Record<string, string | number | null>,
id: payload.id,
};
}
private compareEntryWithCursor(
entry: FetchAgentsResponseEntry,
cursor: FetchAgentsCursor,
sort: FetchAgentsRequestSort[]
): number {
for (const spec of sort) {
const leftValue = this.getFetchAgentsSortValue(entry, spec.key);
const rightValue =
cursor.values[spec.key] !== undefined ? cursor.values[spec.key] ?? null : null;
const base = this.compareSortValues(leftValue, rightValue);
if (base === 0) {
continue;
}
return spec.direction === "asc" ? base : -base;
}
return entry.agent.id.localeCompare(cursor.id);
}
private async listFetchAgentsEntries(
request: Extract<SessionInboundMessage, { type: "fetch_agents_request" }>
): Promise<{
entries: FetchAgentsResponseEntry[];
pageInfo: FetchAgentsResponsePageInfo;
}> {
const filter = request.filter;
const sort = this.normalizeFetchAgentsSort(request.sort);
const includeArchived = filter?.includeArchived ?? false;
let agents = await this.listAgentPayloads({
labels: filter?.labels,
});
if (!includeArchived) {
agents = agents.filter((agent) => !agent.archivedAt);
}
if (filter?.statuses && filter.statuses.length > 0) {
const statuses = new Set(filter.statuses);
agents = agents.filter((agent) => statuses.has(agent.status));
}
if (typeof filter?.requiresAttention === "boolean") {
agents = agents.filter(
(agent) =>
(agent.requiresAttention ?? false) === filter.requiresAttention
);
}
const placementByCwd = new Map<string, Promise<ProjectPlacementPayload>>();
const getPlacement = (cwd: string): Promise<ProjectPlacementPayload> => {
const existing = placementByCwd.get(cwd);
if (existing) {
return existing;
}
const placementPromise = this.buildProjectPlacement(cwd);
placementByCwd.set(cwd, placementPromise);
return placementPromise;
};
let entries = await Promise.all(
agents.map(async (agent) => ({
agent,
project: await getPlacement(agent.cwd),
}))
);
if (filter?.projectKeys && filter.projectKeys.length > 0) {
const projectKeys = new Set(filter.projectKeys.filter((item) => item.trim().length > 0));
entries = entries.filter((entry) => projectKeys.has(entry.project.projectKey));
}
entries.sort((left, right) =>
this.compareFetchAgentsEntries(left, right, sort)
);
const cursorToken = request.page?.cursor;
if (cursorToken) {
const cursor = this.decodeFetchAgentsCursor(cursorToken, sort);
entries = entries.filter(
(entry) => this.compareEntryWithCursor(entry, cursor, sort) > 0
);
}
const limit = request.page?.limit ?? entries.length;
const pagedEntries = entries.slice(0, limit);
const hasMore = entries.length > limit;
const nextCursor =
hasMore && pagedEntries.length > 0
? this.encodeFetchAgentsCursor(
pagedEntries[pagedEntries.length - 1],
sort
)
: null;
return {
entries: pagedEntries,
pageInfo: {
nextCursor,
prevCursor: request.page?.cursor ?? null,
hasMore,
},
};
}
private async handleFetchAgents(
request: Extract<SessionInboundMessage, { type: "fetch_agents_request" }>
): Promise<void> {
try {
const groups = await this.listAgentsGroupedByProjectPayload(filter);
const payload = await this.listFetchAgentsEntries(request);
this.emit({
type: "fetch_agents_grouped_by_project_response",
payload: { requestId, groups },
type: "fetch_agents_response",
payload: {
requestId: request.requestId,
...payload,
},
});
} catch (error) {
this.sessionLogger.error(
{ err: error },
"Failed to handle fetch_agents_grouped_by_project_request"
);
const code =
error instanceof SessionRequestError ? error.code : "fetch_agents_failed";
const message =
error instanceof Error ? error.message : "Failed to fetch agents";
this.sessionLogger.error({ err: error }, "Failed to handle fetch_agents_request");
this.emit({
type: "fetch_agents_grouped_by_project_response",
payload: { requestId, groups: [] },
type: "rpc_error",
payload: {
requestId: request.requestId,
requestType: request.type,
error: message,
code,
},
});
}
}
@@ -5962,6 +6181,10 @@ export class Session {
unsubscribe();
}
this.terminalSubscriptions.clear();
for (const unsubscribeExit of this.terminalExitSubscriptions.values()) {
unsubscribeExit();
}
this.terminalExitSubscriptions.clear();
this.detachAllTerminalStreams({ emitExit: false });
for (const target of this.checkoutDiffTargets.values()) {
@@ -5975,6 +6198,43 @@ export class Session {
// Terminal Handlers
// ============================================================================
private ensureTerminalExitSubscription(terminal: TerminalSession): void {
if (this.terminalExitSubscriptions.has(terminal.id)) {
return;
}
const unsubscribeExit = terminal.onExit(() => {
this.handleTerminalExited(terminal.id);
});
this.terminalExitSubscriptions.set(terminal.id, unsubscribeExit);
}
private handleTerminalExited(terminalId: string): void {
const unsubscribeExit = this.terminalExitSubscriptions.get(terminalId);
if (unsubscribeExit) {
unsubscribeExit();
this.terminalExitSubscriptions.delete(terminalId);
}
const unsubscribe = this.terminalSubscriptions.get(terminalId);
if (unsubscribe) {
try {
unsubscribe();
} catch (error) {
this.sessionLogger.warn(
{ err: error, terminalId },
"Failed to unsubscribe terminal after process exit"
);
}
this.terminalSubscriptions.delete(terminalId);
}
const streamId = this.terminalStreamByTerminalId.get(terminalId);
if (typeof streamId === "number") {
this.detachTerminalStream(streamId, { emitExit: true });
}
}
private async handleListTerminalsRequest(msg: ListTerminalsRequest): Promise<void> {
if (!this.terminalManager) {
this.emit({
@@ -5990,6 +6250,9 @@ export class Session {
try {
const terminals = await this.terminalManager.getTerminals(msg.cwd);
for (const terminal of terminals) {
this.ensureTerminalExitSubscription(terminal);
}
this.emit({
type: "list_terminals_response",
payload: {
@@ -6029,6 +6292,7 @@ export class Session {
cwd: msg.cwd,
name: msg.name,
});
this.ensureTerminalExitSubscription(session);
this.emit({
type: "create_terminal_response",
payload: {
@@ -6077,6 +6341,7 @@ export class Session {
});
return;
}
this.ensureTerminalExitSubscription(session);
// Unsubscribe from previous subscription if any
const existing = this.terminalSubscriptions.get(msg.terminalId);
@@ -6128,6 +6393,7 @@ export class Session {
this.sessionLogger.warn({ terminalId: msg.terminalId }, "Terminal not found for input");
return;
}
this.ensureTerminalExitSubscription(session);
session.send(msg.message);
}
packages/server/src/shared/messages.ts
+26 -35
View File
@@ -458,16 +458,24 @@ export const FetchAgentsRequestMessageSchema = z.object({
filter: z
.object({
labels: z.record(z.string()).optional(),
projectKeys: z.array(z.string()).optional(),
statuses: z.array(AgentStatusSchema).optional(),
includeArchived: z.boolean().optional(),
requiresAttention: z.boolean().optional(),
})
.optional(),
});
export const FetchAgentsGroupedByProjectRequestMessageSchema = z.object({
type: z.literal("fetch_agents_grouped_by_project_request"),
requestId: z.string(),
filter: z
sort: z
.array(
z.object({
key: z.enum(["status_priority", "created_at", "updated_at", "title"]),
direction: z.enum(["asc", "desc"]),
})
)
.optional(),
page: z
.object({
labels: z.record(z.string()).optional(),
limit: z.number().int().positive().max(1000),
cursor: z.string().min(1).optional(),
})
.optional(),
});
@@ -1002,7 +1010,6 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
AbortRequestMessageSchema,
AudioPlayedMessageSchema,
FetchAgentsRequestMessageSchema,
FetchAgentsGroupedByProjectRequestMessageSchema,
FetchAgentRequestMessageSchema,
SubscribeAgentUpdatesMessageSchema,
UnsubscribeAgentUpdatesMessageSchema,
@@ -1399,26 +1406,17 @@ export const FetchAgentsResponseMessageSchema = z.object({
type: z.literal("fetch_agents_response"),
payload: z.object({
requestId: z.string(),
agents: z.array(AgentSnapshotPayloadSchema),
}),
});
const ProjectGroupedAgentEntryPayloadSchema = z.object({
agent: AgentSnapshotPayloadSchema,
checkout: ProjectCheckoutLitePayloadSchema,
});
const ProjectGroupPayloadSchema = z.object({
projectKey: z.string(),
projectName: z.string(),
agents: z.array(ProjectGroupedAgentEntryPayloadSchema),
});
export const FetchAgentsGroupedByProjectResponseMessageSchema = z.object({
type: z.literal("fetch_agents_grouped_by_project_response"),
payload: z.object({
requestId: z.string(),
groups: z.array(ProjectGroupPayloadSchema),
entries: z.array(
z.object({
agent: AgentSnapshotPayloadSchema,
project: ProjectPlacementPayloadSchema,
})
),
pageInfo: z.object({
nextCursor: z.string().nullable(),
prevCursor: z.string().nullable(),
hasMore: z.boolean(),
}),
}),
});
@@ -1972,7 +1970,6 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
AgentStreamMessageSchema,
AgentStatusMessageSchema,
FetchAgentsResponseMessageSchema,
FetchAgentsGroupedByProjectResponseMessageSchema,
FetchAgentResponseMessageSchema,
FetchAgentTimelineResponseMessageSchema,
SendAgentMessageResponseMessageSchema,
@@ -2042,9 +2039,6 @@ export type ProjectPlacementPayload = z.infer<typeof ProjectPlacementPayloadSche
export type FetchAgentsResponseMessage = z.infer<
typeof FetchAgentsResponseMessageSchema
>;
export type FetchAgentsGroupedByProjectResponseMessage = z.infer<
typeof FetchAgentsGroupedByProjectResponseMessageSchema
>;
export type FetchAgentResponseMessage = z.infer<
typeof FetchAgentResponseMessageSchema
>;
@@ -2079,9 +2073,6 @@ export type ActivityLogPayload = z.infer<typeof ActivityLogPayloadSchema>;
// Type exports for inbound message types
export type VoiceAudioChunkMessage = z.infer<typeof VoiceAudioChunkMessageSchema>;
export type FetchAgentsRequestMessage = z.infer<typeof FetchAgentsRequestMessageSchema>;
export type FetchAgentsGroupedByProjectRequestMessage = z.infer<
typeof FetchAgentsGroupedByProjectRequestMessageSchema
>;
export type FetchAgentRequestMessage = z.infer<typeof FetchAgentRequestMessageSchema>;
export type SendAgentMessageRequest = z.infer<typeof SendAgentMessageRequestSchema>;
export type WaitForFinishRequest = z.infer<typeof WaitForFinishRequestSchema>;
packages/server/src/terminal/terminal-manager.test.ts
+34 -2
View File
@@ -1,6 +1,21 @@
import { describe, it, expect, afterEach } from "vitest";
import { createTerminalManager, type TerminalManager } from "./terminal-manager.js";
async function waitForCondition(
predicate: () => boolean,
timeoutMs: number,
intervalMs = 25
): Promise<void> {
const start = Date.now();
while (Date.now() - start < timeoutMs) {
if (predicate()) {
return;
}
await new Promise((resolve) => setTimeout(resolve, intervalMs));
}
throw new Error(`Timed out after ${timeoutMs}ms waiting for condition`);
}
describe("TerminalManager", () => {
let manager: TerminalManager;
@@ -136,6 +151,21 @@ describe("TerminalManager", () => {
manager = createTerminalManager();
expect(() => manager.killTerminal("unknown-id")).not.toThrow();
});
it("auto-removes terminal when shell exits", async () => {
manager = createTerminalManager();
const terminals = await manager.getTerminals("/tmp");
const exitedId = terminals[0].id;
terminals[0].send({ type: "input", data: "\u0004" });
await waitForCondition(() => manager.getTerminal(exitedId) === undefined, 10000);
expect(manager.getTerminal(exitedId)).toBeUndefined();
const remaining = await manager.getTerminals("/tmp");
expect(remaining).toHaveLength(1);
expect(remaining[0].id).not.toBe(exitedId);
});
});
describe("listDirectories", () => {
@@ -161,12 +191,14 @@ describe("TerminalManager", () => {
manager = createTerminalManager();
const tmpTerminals = await manager.getTerminals("/tmp");
const homeTerminals = await manager.getTerminals("/home");
const tmpId = tmpTerminals[0].id;
const homeId = homeTerminals[0].id;
manager.killAll();
expect(manager.listDirectories()).toEqual([]);
expect(manager.getTerminal(tmpTerminals[0].id)).toBeUndefined();
expect(manager.getTerminal(homeTerminals[0].id)).toBeUndefined();
expect(manager.getTerminal(tmpId)).toBeUndefined();
expect(manager.getTerminal(homeId)).toBeUndefined();
});
});
});
packages/server/src/terminal/terminal-manager.ts
+52 -27
View File
@@ -12,6 +12,7 @@ export interface TerminalManager {
export function createTerminalManager(): TerminalManager {
const terminalsByCwd = new Map<string, TerminalSession[]>();
const terminalsById = new Map<string, TerminalSession>();
const terminalExitUnsubscribeById = new Map<string, () => void>();
function assertAbsolutePath(cwd: string): void {
if (!cwd.startsWith("/")) {
@@ -19,16 +20,56 @@ export function createTerminalManager(): TerminalManager {
}
}
function removeSessionById(id: string, options: { kill: boolean }): void {
const session = terminalsById.get(id);
if (!session) {
return;
}
const unsubscribeExit = terminalExitUnsubscribeById.get(id);
if (unsubscribeExit) {
unsubscribeExit();
terminalExitUnsubscribeById.delete(id);
}
terminalsById.delete(id);
const terminals = terminalsByCwd.get(session.cwd);
if (terminals) {
const index = terminals.findIndex((terminal) => terminal.id === id);
if (index !== -1) {
terminals.splice(index, 1);
}
if (terminals.length === 0) {
terminalsByCwd.delete(session.cwd);
}
}
if (options.kill) {
session.kill();
}
}
function registerSession(session: TerminalSession): TerminalSession {
terminalsById.set(session.id, session);
const unsubscribeExit = session.onExit(() => {
removeSessionById(session.id, { kill: false });
});
terminalExitUnsubscribeById.set(session.id, unsubscribeExit);
return session;
}
return {
async getTerminals(cwd: string): Promise<TerminalSession[]> {
assertAbsolutePath(cwd);
let terminals = terminalsByCwd.get(cwd);
if (!terminals || terminals.length === 0) {
const session = await createTerminal({ cwd, name: "Terminal 1" });
const session = registerSession(
await createTerminal({ cwd, name: "Terminal 1" })
);
terminals = [session];
terminalsByCwd.set(cwd, terminals);
terminalsById.set(session.id, session);
}
return terminals;
},
@@ -38,14 +79,15 @@ export function createTerminalManager(): TerminalManager {
const terminals = terminalsByCwd.get(options.cwd) ?? [];
const defaultName = `Terminal ${terminals.length + 1}`;
const session = await createTerminal({
cwd: options.cwd,
name: options.name ?? defaultName,
});
const session = registerSession(
await createTerminal({
cwd: options.cwd,
name: options.name ?? defaultName,
})
);
terminals.push(session);
terminalsByCwd.set(options.cwd, terminals);
terminalsById.set(session.id, session);
return session;
},
@@ -55,22 +97,7 @@ export function createTerminalManager(): TerminalManager {
},
killTerminal(id: string): void {
const session = terminalsById.get(id);
if (!session) return;
session.kill();
terminalsById.delete(id);
const terminals = terminalsByCwd.get(session.cwd);
if (terminals) {
const index = terminals.indexOf(session);
if (index !== -1) {
terminals.splice(index, 1);
}
if (terminals.length === 0) {
terminalsByCwd.delete(session.cwd);
}
}
removeSessionById(id, { kill: true });
},
listDirectories(): string[] {
@@ -78,11 +105,9 @@ export function createTerminalManager(): TerminalManager {
},
killAll(): void {
for (const session of terminalsById.values()) {
session.kill();
for (const id of Array.from(terminalsById.keys())) {
removeSessionById(id, { kill: true });
}
terminalsByCwd.clear();
terminalsById.clear();
},
};
}
packages/server/src/terminal/terminal.ts
+57 -5
View File
@@ -64,6 +64,7 @@ export interface TerminalSession {
cwd: string;
send(msg: ClientMessage): void;
subscribe(listener: (msg: ServerMessage) => void): () => void;
onExit(listener: () => void): () => void;
subscribeRaw(
listener: (chunk: TerminalRawChunk) => void,
options?: { fromOffset?: number }
@@ -190,6 +191,7 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
const id = randomUUID();
const listeners = new Set<(msg: ServerMessage) => void>();
const rawListeners = new Set<(chunk: TerminalRawChunk) => void>();
const exitListeners = new Set<() => void>();
const rawOutputChunks: Array<{
startOffset: number;
endOffset: number;
@@ -199,6 +201,8 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
let rawBufferBytes = 0;
let outputOffset = 0;
let killed = false;
let disposed = false;
let exitEmitted = false;
// Create xterm.js headless terminal
const terminal = new Terminal({
@@ -221,6 +225,32 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
},
});
function emitExit(): void {
if (exitEmitted) {
return;
}
exitEmitted = true;
for (const listener of Array.from(exitListeners)) {
try {
listener();
} catch {
// no-op
}
}
exitListeners.clear();
}
function disposeResources(): void {
if (disposed) {
return;
}
disposed = true;
terminal.dispose();
listeners.clear();
rawListeners.clear();
exitListeners.clear();
}
function appendRawOutputChunk(data: string): void {
const chunkBytes = Buffer.byteLength(data);
if (chunkBytes <= 0) {
@@ -292,6 +322,8 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
ptyProcess.onExit(() => {
killed = true;
emitExit();
disposeResources();
});
function getState(): TerminalState {
@@ -340,6 +372,24 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
};
}
function onExit(listener: () => void): () => void {
if (killed) {
queueMicrotask(() => {
try {
listener();
} catch {
// no-op
}
});
return () => {};
}
exitListeners.add(listener);
return () => {
exitListeners.delete(listener);
};
}
function subscribeRaw(
listener: (chunk: TerminalRawChunk) => void,
options?: { fromOffset?: number }
@@ -380,11 +430,12 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
}
function kill(): void {
if (killed) return;
killed = true;
ptyProcess.kill();
terminal.dispose();
rawListeners.clear();
if (!killed) {
killed = true;
ptyProcess.kill();
emitExit();
}
disposeResources();
}
// Small delay to let shell initialize
@@ -396,6 +447,7 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
cwd,
send,
subscribe,
onExit,
subscribeRaw,
getOutputOffset,
getState,