Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion CONTEXT.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ Glossary of project-specific terms. pie integrates the Pi coding agent into the
## Session Domain

**Project**:
A working directory the user has registered with the server, identified by a server-generated UUID. The single source of the projectId → directory mapping; the directory field is `path`. Registration is the trust boundary for declarative Project-local Pi resources: session children approve prompts, skills, and context from that Project. Automatic Pi extension discovery is disabled for every Pie-owned child because Pie does not yet model inputs that an extension handles without starting a model turn; this intentionally makes discovered global and Project extension commands, tools, hooks, and UI unavailable. The extension request plumbing remains at the Pi protocol seam for compatibility and future explicit extension support. Sessions always resolve their working directory through a Project, never from a caller-supplied path. A session may start without picking an existing Project: `project.allocateChatProjectDir` creates a folder under the **chat-project root**, registers it with `type: "chat"`, then `session.create` uses that id as usual. Chat projects stay off the **Projects** sidebar and picker; their sessions appear under **Recent**.
A working directory the user has registered with the server, identified by a server-generated UUID. The single source of the projectId → directory mapping; the directory field is `path`. Registration is the trust boundary for Project-local Pi resources: session children approve extensions, prompts, skills, and context from that Project. Pi loads the user's extensions and configured packages for model/command discovery and live sessions, including provider registrations. Discovery without a Project does not approve Project-local resources. Extension commands and input hooks can handle a prompt without starting a model turn; Pie acknowledges that input without creating or waiting for a turn. Supported extension UI requests use the Pi protocol seam. Blocking dialogs without an active turn (including startup and noninteractive command discovery) receive an explicit decline/cancellation; they are never auto-approved or left waiting for a consumer. Terminal-only custom UI remains unavailable. Sessions always resolve their working directory through a Project, never from a caller-supplied path. A session may start without picking an existing Project: `project.allocateChatProjectDir` creates a folder under the **chat-project root**, registers it with `type: "chat"`, then `session.create` uses that id as usual. Chat projects stay off the **Projects** sidebar and picker; their sessions appear under **Recent**.
_Avoid_: workspace, repo, cwd (for the Project field)

**Chat-project root**:
Expand Down
3 changes: 1 addition & 2 deletions packages/server/src/harness/agent-service.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
import { getAgentDir } from "@earendil-works/pi-coding-agent";
import type { ListAgentCommandsOutput, ListAgentModelsOutput } from "@getpie/contract";
import { Context, Effect, Layer } from "effect";

Expand Down Expand Up @@ -28,7 +27,7 @@ export const makePiAgentService = (): PiAgentServiceShape => ({
});
}),
listModels: Effect.fn("PiAgentService.listModels")(function* (cwd?: string) {
return yield* listAvailablePiModels(cwd ?? getAgentDir());
return yield* listAvailablePiModels(cwd);
}),
});

Expand Down
10 changes: 8 additions & 2 deletions packages/server/src/harness/pi/list-available-commands.ts
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
import { getAgentDir } from "@earendil-works/pi-coding-agent";
import * as NodeServices from "@effect/platform-node/NodeServices";
import type { AgentCommand } from "@getpie/contract";
import { Effect } from "effect";
import { Effect, Stream } from "effect";

import { PI_PROJECT_PROCESS_ARGS } from "./project-resource-policy";
import { declineUiResponse } from "./request";
import { resolvePiExecutable } from "./resolve-executable";
import { makePiTransport } from "./transport";

Expand All @@ -21,9 +22,14 @@ export function listAvailablePiCommands(
const transport = yield* makePiTransport({
executable: resolvePiExecutable(),
cwd: cwd ?? agentDir,
args: PI_PROJECT_PROCESS_ARGS,
args: cwd === undefined ? [] : PI_PROJECT_PROCESS_ARGS,
env: { PI_CODING_AGENT_DIR: agentDir },
});
// Discovery has no interactive consumer, including during session_start.
// Always send an explicit decline rather than leave a dialog pending.
yield* Stream.runForEach(transport.uiRequests, (request) =>
transport.respondUi(declineUiResponse(request)),
).pipe(Effect.forkScoped);
const data = yield* transport.command<{ commands?: AgentCommand[] }>({
type: "get_commands",
});
Expand Down
10 changes: 7 additions & 3 deletions packages/server/src/harness/pi/list-available-models.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ const listModelsError = (cause: unknown) =>
* RPC `get_available_models`, without spawning pie-pi-process.
*/
export function listAvailablePiModels(
cwd: string,
cwd?: string,
): Effect.Effect<ListAgentModelsOutput, AgentOperationError> {
return Effect.gen(function* () {
const { createAgentSessionServices, getAgentDir, SettingsManager } = yield* Effect.tryPromise({
Expand All @@ -29,9 +29,13 @@ export function listAvailablePiModels(
const services = yield* Effect.tryPromise({
try: () =>
createAgentSessionServices({
cwd,
cwd: cwd ?? getAgentDir(),
agentDir: getAgentDir(),
settingsManager: SettingsManager.create(cwd, getAgentDir(), PI_PROJECT_SETTINGS_OPTIONS),
settingsManager: SettingsManager.create(
cwd ?? getAgentDir(),
getAgentDir(),
cwd === undefined ? { projectTrusted: false } : PI_PROJECT_SETTINGS_OPTIONS,
),
resourceLoaderOptions: PI_PROJECT_LOADER_OPTIONS,
}),
catch: listModelsError,
Expand Down
50 changes: 29 additions & 21 deletions packages/server/src/harness/pi/process.ts
Original file line number Diff line number Diff line change
Expand Up @@ -417,21 +417,6 @@ export const makePiProcessWithDependencies = <R>(
Effect.sync(() => dependencies.onExit?.(sessionId, transport.pid)),
);

// Readiness handshake: pi's CLI front-end resolves the session (and
// may exit with a human-readable error) before the RPC loop starts.
yield* transport.command<RpcSessionState>({ type: "get_state" }).pipe(
Effect.timeoutOrElse({
duration: HANDSHAKE_TIMEOUT,
orElse: () =>
Effect.fail(
new PiTransportError({
operation: "handshake-timeout",
cause: new Error("Pi RPC get_state handshake timed out"),
}),
),
}),
);

const session: SessionState = {
sessionId,
scope,
Expand All @@ -449,18 +434,34 @@ export const makePiProcessWithDependencies = <R>(
turnState: yield* Ref.make<PiTurnState>({ _tag: "Idle" }),
transform: createPiTransform(sessionId),
};
yield* Ref.update(sessions, (current) => new Map(current).set(sessionId, session));

yield* Stream.runForEach(transport.events, (event) => routeEvent(session, event)).pipe(
Effect.catch((error) => reportCrash(session, error)),
Effect.forkIn(scope),
);
// Startup has no active turn or question consumer. Route UI now so
// session_start receives the existing safe decline, not a deadlock.
yield* Stream.runForEach(transport.uiRequests, (request) =>
Effect.forkIn(handleUiRequest(session, request), scope).pipe(Effect.asVoid),
).pipe(
Effect.catch((error) => reportCrash(session, error)),
Effect.forkIn(scope),
);
// Readiness still waits for Pi to resolve the session and finish
// extension startup before publishing it to callers.
yield* transport.command<RpcSessionState>({ type: "get_state" }).pipe(
Effect.timeoutOrElse({
duration: HANDSHAKE_TIMEOUT,
orElse: () =>
Effect.fail(
new PiTransportError({
operation: "handshake-timeout",
cause: new Error("Pi RPC get_state handshake timed out"),
}),
),
}),
);
yield* Ref.update(sessions, (current) => new Map(current).set(sessionId, session));

yield* Stream.runForEach(transport.events, (event) => routeEvent(session, event)).pipe(
Effect.catch((error) => reportCrash(session, error)),
Effect.forkIn(scope),
);
yield* transport.awaitTermination.pipe(
Effect.catch((error) => reportCrash(session, error)),
Effect.forkIn(scope),
Expand Down Expand Up @@ -552,6 +553,13 @@ export const makePiProcessWithDependencies = <R>(

if (!started) {
const active = yield* Ref.get(session.turnState);
if (admission?.disposition === "handled") {
return {
turnId: active._tag === "Active" ? active.turnId : uuid(),
started: false,
output: Stream.empty,
};
}
if (active._tag === "Active") {
return {
turnId: active.turnId,
Expand Down
9 changes: 4 additions & 5 deletions packages/server/src/harness/pi/project-resource-policy.ts
Original file line number Diff line number Diff line change
@@ -1,12 +1,11 @@
// A registered Project is trusted for declarative Pi resources, but Pie does
// not model extension-handled inputs that complete without a model turn.
// Keep cold discovery and live children on the same resource policy.
// Registered Projects are trusted for Pi resources, including extensions.
// User extensions and configured packages also supply providers and commands.
// Cold discovery skips terminal themes/context, not executable extensions.
export const PI_PROJECT_SETTINGS_OPTIONS = { projectTrusted: true } as const;

export const PI_PROJECT_LOADER_OPTIONS = {
noExtensions: true,
noThemes: true,
noContextFiles: true,
} as const;

export const PI_PROJECT_PROCESS_ARGS = ["--approve", "--no-extensions"] as const;
export const PI_PROJECT_PROCESS_ARGS = ["--approve"] as const;
6 changes: 5 additions & 1 deletion packages/server/src/harness/pi/rpc/main.ts
Original file line number Diff line number Diff line change
Expand Up @@ -64,14 +64,18 @@ export const main = async (): Promise<void> => {
cwd: options.cwd,
agentDir: options.agentDir,
modelRuntimeSignal: AbortSignal.timeout(15_000),
settingsManager: SettingsManager.create(options.cwd, options.agentDir, {
projectTrusted: parsed.projectTrustOverride ?? false,
}),
extensionFlagValues: FFF_OVERRIDE_FLAGS,
resourceLoaderOptions: {
noExtensions: parsed.noExtensions ?? false,
extensionFactories: [
...builtInExtensions,
piBashExtension(options.cwd),
sessionToolsExtensionFactory,
],
additionalExtensionPaths: [bundledFff],
additionalExtensionPaths: [bundledFff, ...(parsed.extensions ?? [])],
},
});
const resolved = resolveCliModel({
Expand Down
70 changes: 50 additions & 20 deletions packages/server/src/harness/pi/rpc/rpc-mode.ts
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,14 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
let session = runtimeHost.session;
let unsubscribe: (() => void) | undefined;
let unsubscribeBackpressure: (() => void) | undefined;
const exit = {
reject: (_cause: unknown) => {
/* assigned synchronously by the exit promise below */
},
};
const exited = new Promise<never>((_resolve, reject) => {
exit.reject = reject;
});

const output = (frame: unknown) => {
writeRawStdout(serializeJsonLine(frame));
Expand Down Expand Up @@ -91,6 +99,15 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
{ resolve: (value: RpcExtensionUIResponse) => void; reject: (error: Error) => void }
>();

let inputEnded = false;
const cancelExtensionDialogs = () => {
inputEnded = true;
for (const [id, pending] of pendingExtensionRequests) {
pending.resolve({ type: "extension_ui_response", id, cancelled: true });
}
pendingExtensionRequests.clear();
};

// Shutdown request flag
let shutdownRequested = false;
let shuttingDown = false;
Expand All @@ -103,7 +120,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
request: Record<string, unknown>,
parseResponse: (response: RpcExtensionUIResponse) => T,
): Promise<T> {
if (opts?.signal?.aborted) return Promise.resolve(defaultValue);
if (inputEnded || opts?.signal?.aborted) return Promise.resolve(defaultValue);

const id = crypto.randomUUID();
return new Promise((resolve, reject) => {
Expand Down Expand Up @@ -273,6 +290,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
},

async editor(title: string, prefill?: string): Promise<string | undefined> {
if (inputEnded) return undefined;
const id = crypto.randomUUID();
return new Promise((resolve, reject) => {
pendingExtensionRequests.set(id, {
Expand Down Expand Up @@ -391,6 +409,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
},
});

if (shuttingDown) return;
unsubscribe?.();
unsubscribeBackpressure?.();
unsubscribe = session.subscribe((event) => {
Expand All @@ -412,16 +431,13 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve

for (const signal of signals) {
const handler = () => {
void shutdown(signal === "SIGHUP" ? 129 : 143, signal);
void shutdown(signal === "SIGHUP" ? 129 : 143, signal).catch(exit.reject);
};
process.on(signal, handler);
signalCleanupHandlers.push(() => process.off(signal, handler));
}
};

await rebindSession();
registerSignalHandlers();

// Handle a single command
const handleCommand = async (command: RpcCommand): Promise<RpcResponse | undefined> => {
const id = command.id;
Expand All @@ -442,10 +458,9 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
source: "rpc",
preflightResult: (disposition) => {
preflightSucceeded = true;
// 0.99 reports disposition instead of a boolean. isStreaming is
// already true for a prompt that just started, so it cannot mean
// "queued".
output(success(id, "prompt", { started: disposition === "started" }));
// Keep the boolean for older callers, but distinguish handled input
// from a queued prompt: a handled command need not start a model turn.
output(success(id, "prompt", { started: disposition === "started", disposition }));
},
})
.catch((cause: unknown) => {
Expand Down Expand Up @@ -778,11 +793,10 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
/* assigned once stdin listeners are attached */
};

async function shutdown(exitCode = 0, signal?: NodeJS.Signals): Promise<never> {
if (shuttingDown) {
throw new RpcChildExitError(exitCode);
}
async function shutdown(exitCode = 0, signal?: NodeJS.Signals): Promise<void> {
if (shuttingDown) return;
shuttingDown = true;
cancelExtensionDialogs();
for (const cleanup of signalCleanupHandlers) {
cleanup();
}
Expand All @@ -794,7 +808,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
if (signal !== "SIGTERM") {
await flushRawStdout();
}
throw new RpcChildExitError(exitCode);
exit.reject(new RpcChildExitError(exitCode));
}

async function checkShutdownRequested(): Promise<void> {
Expand Down Expand Up @@ -838,6 +852,10 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
// oxlint-disable-next-line typescript/no-unsafe-type-assertion -- JSONL command; handleCommand switches on type
const command = parsed as RpcCommand;
try {
// UI replies above must bypass initialization: session_start can await
// them. Ordinary commands must still wait for extensions to finish binding.
await binding;
if (shuttingDown) return;
const response = await handleCommand(command);
if (response) {
output(response);
Expand All @@ -856,23 +874,35 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
}
};

const binding = rebindSession();
const pendingInput = new Set<Promise<void>>();

const onInputEnd = () => {
void shutdown();
cancelExtensionDialogs();
// Finite JSONL input can arrive while extensions are still binding. Let
// admitted commands finish before removing their event subscriptions.
void binding
.then(() => Promise.all(pendingInput))
.then(() => shutdown())
.catch(exit.reject);
};
process.stdin.on("end", onInputEnd);

detachInput = (() => {
const detachJsonl = attachJsonlLineReader(process.stdin, (line) => {
void handleInputLine(line);
const handling = handleInputLine(line);
pendingInput.add(handling);
void handling.finally(() => pendingInput.delete(handling)).catch(exit.reject);
});
return () => {
detachJsonl();
process.stdin.off("end", onInputEnd);
};
})();

// Keep process alive forever
return new Promise(() => {
/* the child stays alive until shutdown() exits the process */
});
registerSignalHandlers();
await Promise.race([binding, exited]);

// Propagate shutdown to entry.ts rather than throwing in a detached callback.
return exited;
}
2 changes: 1 addition & 1 deletion packages/server/src/harness/pi/rpc/rpc-types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -144,7 +144,7 @@ export type RpcResponse =
type: "response";
command: "prompt";
success: true;
data: { started: boolean };
data: { started: boolean; disposition?: "started" | "queued" | "handled" };
}
| { id?: string; type: "response"; command: "steer"; success: true }
| { id?: string; type: "response"; command: "follow_up"; success: true }
Expand Down
Loading
Loading