Skip to content
Merged
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
10 changes: 5 additions & 5 deletions apps/dispatcher/src/env.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
// V0 surface only: Workflow + R2 + D1 + Container. Queue / DO / Browser
// bindings are deferred to V1+ and intentionally absent here.

import type { Sandbox } from "@cloudflare/sandbox";
import type { RunSandbox } from "./sandbox";
import type { SubstrateFacade } from "@fractalboxdev/flare-dispatch-substrate-contract";

export interface Env {
Expand Down Expand Up @@ -116,7 +116,7 @@ export interface Env {
* id)` in `@fractalboxdev/flare-dispatch-runtime-cf` resolves the typed sandbox RPC surface.
* The default for every run except those declaring `sandboxImage: "browser"`.
*/
readonly RUNS_SANDBOX: DurableObjectNamespace<Sandbox>;
readonly RUNS_SANDBOX: DurableObjectNamespace<RunSandbox>;

/**
* Container binding — the chromium-baked sandbox image (built from the same
Expand All @@ -131,7 +131,7 @@ export interface Env {
* soft guard against a missing binding, not a guarantee browser runs work
* without the second container.
*/
readonly RUNS_SANDBOX_BROWSER?: DurableObjectNamespace<Sandbox>;
readonly RUNS_SANDBOX_BROWSER?: DurableObjectNamespace<RunSandbox>;

/**
* Container binding — the agent-tier sandbox image (same Dockerfile with
Expand All @@ -144,10 +144,10 @@ export interface Env {
* `flare-agent` and fails (not a guarantee, just a non-crash). Egress is NOT
* configurable away — it is part of the tier. specs/08-self-healing.md § 6.2.
*/
readonly RUNS_SANDBOX_AGENT?: DurableObjectNamespace<Sandbox>;
readonly RUNS_SANDBOX_AGENT?: DurableObjectNamespace<RunSandbox>;

/** Pinned Rust, Zig, and native C++ toolchain for release builds. */
readonly RUNS_SANDBOX_RELEASE?: DurableObjectNamespace<Sandbox>;
readonly RUNS_SANDBOX_RELEASE?: DurableObjectNamespace<RunSandbox>;

/** R2 bucket — `logs/<execution-id>/<step>.ndjson` + `artifacts/...`. */
readonly RUNS_STORAGE: R2Bucket;
Expand Down
6 changes: 4 additions & 2 deletions apps/dispatcher/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
//
// Spec: specs/pm/plan.md § PR5, specs/04-gha-integration.md § Schedule mode.

import { proxyToSandbox } from "@cloudflare/sandbox";
import { proxyToSandbox, type SandboxEnv } from "@cloudflare/sandbox";

import type { Env } from "./env";
import { handleRequest } from "./router";
Expand Down Expand Up @@ -45,7 +45,9 @@ export default {
// never exposes a port, so RUNS_SANDBOX_BROWSER has no preview surface. If a
// future browser-image run needs `exposePort`, the proxy must learn to pick
// the binding from the preview host (tracked in issue #68).
const proxied = await proxyToSandbox(request, { Sandbox: env.RUNS_SANDBOX });
const proxied = await proxyToSandbox<RunSandbox, SandboxEnv<RunSandbox>>(request, {
Sandbox: env.RUNS_SANDBOX,
});
if (proxied !== null) return proxied;
return handleRequest(request, env);
},
Expand Down
49 changes: 49 additions & 0 deletions apps/dispatcher/src/sandbox.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,15 @@
import { Sandbox } from "@cloudflare/sandbox";
import { recordContainerStop, takeRequested, type StopParamsLike } from "./container-stop";
import type { Env } from "./env";
import {
makeCheckCommandOwner,
checkCommandStorage,
} from "@fractalboxdev/flare-dispatch-runtime-cf";
import type {
CheckCommandHandle,
ExecResult,
CheckCommandLogState,
} from "@fractalboxdev/flare-dispatch-core";

/**
* Idle window before a finished container is put to sleep.
Expand Down Expand Up @@ -103,6 +112,46 @@ const recordStopFor = async (
abstract class RecordingSandbox extends Sandbox<Env> {
override sleepAfter = SANDBOX_SLEEP_AFTER;

private checkOwner() {
return makeCheckCommandOwner(checkCommandStorage(this.ctx.storage), this);
}
async startCheckCommand(opts: {
handle: CheckCommandHandle;
command: string;
cwd?: string;
env?: Record<string, string>;
redactValues?: readonly string[];
}) {
return this.checkOwner().start(opts);
}
async observeCheckCommand(handle: CheckCommandHandle) {
return this.checkOwner().observe(handle);
}
async readCheckCommand(
handle: CheckCommandHandle,
stream: "stdout" | "stderr",
offset: number,
length: number,
) {
return this.checkOwner().read(handle, stream, offset, length);
}
async checkCommandLogs(handle: CheckCommandHandle) {
return this.checkOwner().logs(handle);
}
async advanceCheckCommandLogs(
handle: CheckCommandHandle,
expected: CheckCommandLogState,
next: CheckCommandLogState,
) {
return this.checkOwner().advanceLogs(handle, expected, next);
}
async checkCommandReceipt(handle: CheckCommandHandle) {
return this.checkOwner().receipt(handle);
}
async finishCheckCommand(handle: CheckCommandHandle, result: ExecResult) {
return this.checkOwner().finish(handle, result);
}

/**
* Set by our own `destroy()` so the record can say who asked. `workflow.ts`
* tears every run down through an `Effect.ensuring`, so without this the
Expand Down
18 changes: 17 additions & 1 deletion apps/dispatcher/src/workflow.ts
Original file line number Diff line number Diff line change
Expand Up @@ -499,7 +499,23 @@ export class RunWorkflow extends WorkflowEntrypoint<Env> {
approvalFor: (command, scope) => preAssertedApproval(payload.run, command, scope),
},
}
: { sandboxNs }),
: {
sandboxNs,
checkCommandOwner: (container) => {
const box = getSandbox(sandboxNs, container.id);
return {
start: (opts) => box.startCheckCommand(opts),
observe: (handle) => box.observeCheckCommand(handle),
read: (handle, stream, offset, length) =>
box.readCheckCommand(handle, stream, offset, length),
logs: (handle) => box.checkCommandLogs(handle),
advanceLogs: (handle, expected, next) =>
box.advanceCheckCommandLogs(handle, expected, next),
receipt: (handle) => box.checkCommandReceipt(handle),
finish: (handle, result) => box.finishCheckCommand(handle, result),
};
},
}),
workflowStep: step,
// The `Workflow` binding backs the `childRuns` capability — a run can
// `spawnChildRun` / `fanOut` to independent child `RunWorkflow` instances.
Expand Down
56 changes: 56 additions & 0 deletions packages/core/src/check-command.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
import { Schema } from "effect";
import type { ExecOpts, ExecResult } from "./services/sandbox";

export const CheckCommandHandle = Schema.Struct({
id: Schema.String,
container: Schema.Struct({ id: Schema.String }),
fingerprint: Schema.String,
startedAt: Schema.Number,
deadline: Schema.Number,
});
export type CheckCommandHandle = typeof CheckCommandHandle.Type;
export const CheckCommandObservation = Schema.Struct({
state: Schema.Literal("running", "exited", "timeout", "missing"),
exitCode: Schema.optional(Schema.Number),
durationMs: Schema.optional(Schema.Number),
});
export type CheckCommandObservation = typeof CheckCommandObservation.Type;
export const CheckCommandLogState = Schema.Struct({
stdoutOffset: Schema.Number,
stderrOffset: Schema.Number,
stdoutTail: Schema.String,
stderrTail: Schema.String,
chunks: Schema.Array(Schema.String),
});
export type CheckCommandLogState = typeof CheckCommandLogState.Type;
export const CheckCommandChunk = Schema.Struct({ text: Schema.String, bytes: Schema.Number });
export type CheckCommandChunk = typeof CheckCommandChunk.Type;

/** The process owner's RPC boundary; process identity survives Workflow retries. */
export interface CheckCommandOwner {
start(opts: {
handle: CheckCommandHandle;
command: string;
cwd?: string;
env?: Record<string, string>;
redactValues?: readonly string[];
}): Promise<void>;
observe(handle: CheckCommandHandle): Promise<CheckCommandObservation>;
read(
handle: CheckCommandHandle,
stream: "stdout" | "stderr",
offset: number,
length: number,
): Promise<CheckCommandChunk>;
logs(handle: CheckCommandHandle): Promise<CheckCommandLogState>;
advanceLogs(
handle: CheckCommandHandle,
expected: CheckCommandLogState,
next: CheckCommandLogState,
): Promise<void>;
receipt(handle: CheckCommandHandle): Promise<ExecResult | undefined>;
finish(handle: CheckCommandHandle, result: ExecResult): Promise<ExecResult>;
}
/** Durable commands declare their log-redaction policy; [] declares no secret values. */
export type CheckCommandOpts = ExecOpts & { readonly redactValues: readonly string[] };
export type CheckCommandPrepare = CheckCommandOpts & { readonly stepName: string };
13 changes: 12 additions & 1 deletion packages/core/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,17 @@ export {
type ScheduleContext,
type WebhookPayload,
} from "./define-run";
export { step, runEffect, type StepOpts } from "./step";
export { step, checkpointSleep, runEffect, type StepOpts } from "./step";
export { CurrentStep, type CurrentStepInfo } from "./current-step";
export {
CheckCommandHandle,
CheckCommandObservation,
CheckCommandLogState,
CheckCommandChunk,
type CheckCommandOwner,
type CheckCommandPrepare,
type CheckCommandOpts,
} from "./check-command";
export { type RunContext } from "./context";

// --- Writeback (run-declared "propose a diff as a PR") -----------------------
Expand Down Expand Up @@ -99,6 +108,8 @@ export {
type Container,
type DetachedHandle,
type ExecResult,
ExecResultSchema,
type CheckCommandService,
type ExecOpts,
type ExposeResult,
} from "./services/sandbox";
Expand Down
39 changes: 31 additions & 8 deletions packages/core/src/services/sandbox.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,13 @@
//
// Spec: specs/03-dsl.md § sandbox.

import { Context, type Duration, Effect } from "effect";
import { Context, type Duration, Effect, Schema } from "effect";
import type {
CheckCommandHandle,
CheckCommandObservation,
CheckCommandPrepare,
CheckCommandOpts,
} from "../check-command";
import type {
CheckoutFailed,
ContainerBusy,
Expand All @@ -31,13 +37,28 @@ export type DetachedHandle = {
* The result of a command that ran to completion. A non-zero `exitCode` is a
* normal result (a failing test), surfaced to the run — not an Effect failure.
*/
export type ExecResult = {
readonly exitCode: number;
readonly durationMs: number;
readonly logPath: string; // R2 key for the captured stdout/stderr
readonly stdout: string; // last N KB inlined; full log streamed to R2
readonly stderr: string;
};
export const ExecResultSchema = Schema.Struct({
exitCode: Schema.Number,
durationMs: Schema.Number,
logPath: Schema.String,
stdout: Schema.String,
stderr: Schema.String,
});
export type ExecResult = typeof ExecResultSchema.Type;

/** Durable command boundaries require an explicit policy, including deliberate []. */
export interface CheckCommandService {
prepare(opts: CheckCommandPrepare): Effect.Effect<CheckCommandHandle, ExecFailed>;
start(handle: CheckCommandHandle, opts: CheckCommandOpts): Effect.Effect<void, ExecFailed>;
observe(
handle: CheckCommandHandle,
opts: CheckCommandOpts,
): Effect.Effect<CheckCommandObservation, ExecFailed>;
finalize(
handle: CheckCommandHandle,
opts: CheckCommandOpts,
): Effect.Effect<ExecResult, ExecFailed | ExecTimeout>;
}

export type ExecOpts = {
readonly cwd?: string;
Expand Down Expand Up @@ -156,6 +177,8 @@ export interface SandboxService {
container?: Container;
}) => Effect.Effect<string, CheckoutFailed>;
readonly exec: (opts: ExecOpts) => Effect.Effect<ExecResult, ExecFailed | ExecTimeout>;
/** Available on the SDK backend; other execution backends retain their own contract. */
readonly checkCommand?: CheckCommandService;
/**
* Read a container file's full text content into the Worker. The companion
* to `exec` for large outputs: `ExecResult.stdout` inlines only a bounded
Expand Down
1 change: 1 addition & 0 deletions packages/core/src/services/step-runner.ts
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ import type { StepOpts } from "../step-opts";
* `RunContext`).
*/
export interface StepRunnerService {
readonly sleep?: (name: string, milliseconds: number) => Effect.Effect<void, StepFailed>;
readonly run: <A, E>(
name: string,
body: () => Effect.Effect<A, E, RunContext>,
Expand Down
5 changes: 5 additions & 0 deletions packages/core/src/step.ts
Original file line number Diff line number Diff line change
Expand Up @@ -152,3 +152,8 @@ const waitForEvent = <P, I>(
Effect.flatMap(StepRunner, (runner) => runner.waitForEvent(name, opts));

export const step: StepFn = Object.assign(stepImpl, { waitForEvent });
/** Native Workflow sleep persists its wake-up; inline tests use the Effect clock. */
export const checkpointSleep = (name: string, milliseconds: number) =>
Effect.flatMap(StepRunner, (runner) =>
runner.sleep !== undefined ? runner.sleep(name, milliseconds) : Effect.sleep(milliseconds),
);
4 changes: 2 additions & 2 deletions packages/runtime-cf/src/artifact-r2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -96,10 +96,10 @@ const artifactUrl = (executionId: string, name: string, publicOrigin?: string):
* returned artifact URLs so they survive GitHub's
* markdown base-URL resolution. `undefined` → relative.
*/
export const makeR2ArtifactLive = (
export const makeR2ArtifactLive = <T extends Sandbox>(
bucket: R2Bucket,
executionId: string,
ns?: DurableObjectNamespace<Sandbox>,
ns?: DurableObjectNamespace<T>,
publicOrigin?: string,
): Layer.Layer<Artifact> => {
const service: ArtifactService = {
Expand Down
6 changes: 3 additions & 3 deletions packages/runtime-cf/src/cache-r2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -53,12 +53,12 @@ const TARBALL_PATH = "/tmp/fd-cache.tar.gz";
* binding. Cache entries are reused across executions but isolated per repo.
*
* @param bucket the R2 binding (`env.RUNS_STORAGE`).
* @param ns the `RUNS_SANDBOX` DurableObjectNamespace<Sandbox>.
* @param ns the `RUNS_SANDBOX` namespace for the supplied Sandbox subclass.
* @param repo the `owner/name` slug — scopes the archive key per repo.
*/
export const makeCacheR2Live = (
export const makeCacheR2Live = <T extends Sandbox>(
bucket: R2Bucket,
ns: DurableObjectNamespace<Sandbox>,
ns: DurableObjectNamespace<T>,
repo: string,
): Layer.Layer<Cache> => {
const save: SaveFn = (opts) =>
Expand Down
Loading
Loading