harness effect sdk

sanchitrk · Sep 1, 2026 · Aug 24, 2026 · Aug 24, 2026 · Aug 24, 2026 · Aug 24, 2026
commit fffa345bfeee471ba40d69dedb177dbfe59bbdcc
diff --git a/packages/harness/package.json b/packages/harness/package.json
@@ -21,6 +21,9 @@
 		"url": "git+https://github.com/codeworksh/codework.git",
 		"directory": "packages/harness"
 	},
+	"bin": {
+		"codework": "./dist/pack/cli/index.mjs"
+	},
 	"files": [
 		"dist",
 		"src"
@@ -33,6 +36,11 @@
 			"import": "./dist/pack/index.mjs",
 			"default": "./dist/pack/index.mjs"
 		},
+		"./effect": {
+			"types": "./dist/pack/effect.d.mts",
+			"import": "./dist/pack/effect.mjs",
+			"default": "./dist/pack/effect.mjs"
+		},
 		"./package.json": "./package.json"
 	},
 	"scripts": {

diff --git a/packages/harness/src/cli/index.ts b/packages/harness/src/cli/index.ts
@@ -0,0 +1,109 @@
+#!/usr/bin/env node
+
+import * as NodeRuntime from "@effect/platform-node/NodeRuntime";
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import { Effect, Fiber, Option, Queue, Schema, Stream } from "effect";
+import { Argument, CliError, Command, Flag } from "effect/unstable/cli";
+import { resolve } from "node:path";
+import { Harness } from "../effect/harness.ts";
+import { Session } from "../effect/session.ts";
+import type { EventSchema } from "../event/schema.ts";
+
+const writeOut = (value: string) => Effect.sync(() => void process.stdout.write(value));
+const writeError = (value: string) => Effect.sync(() => void process.stderr.write(value));
+
+class InvalidInputError extends Schema.TaggedError<InvalidInputError>()("CLI.InvalidInputError", {
+	message: Schema.String,
+}) {}
+
+const render = (ended: Queue.Queue<string>) =>
+	Effect.fn("CLI.render")(function* (event: EventSchema.Payload) {
+		if (event.type === "session.llm.text.delta") {
+			yield* writeOut((event.data as { readonly delta: string }).delta);
+			return;
+		}
+		if (event.type === "session.turn.ended") {
+			yield* Queue.offer(ended, (event.data as { readonly messageId: string }).messageId);
+		}
+	});
+
+const awaitMessage = Effect.fn("CLI.awaitMessage")(function* (ended: Queue.Queue<string>, messageId: string) {
+	while ((yield* Queue.take(ended)) !== messageId) {
+		// Earlier turns in the same drain are expected when a tool call continues.
+	}
+});
+
+const root = Command.make("codework").pipe(
+	Command.withSharedFlags({
+		home: Flag.string("home").pipe(Flag.withDescription("Harness data directory"), Flag.optional),
+		database: Flag.string("database").pipe(Flag.withDescription("SQLite path or :memory:"), Flag.optional),
+	}),
+	Command.withDescription("Run Codework agent sessions"),
+);
+
+const run = Command.make(
+	"run",
+	{
+		prompt: Argument.string("prompt").pipe(Argument.withDescription("Prompt for the agent")),
+		session: Flag.string("session").pipe(
+			Flag.withAlias("s"),
+			Flag.withDescription("Continue an existing session ID"),
+			Flag.optional,
+		),
+		cwd: Flag.string("cwd").pipe(
+			Flag.withAlias("C"),
+			Flag.withDescription("Working directory for a new local session"),
+			Flag.optional,
+		),
+	},
+	Effect.fn("CLI.run")(function* ({ prompt, session, cwd }) {
+		const shared = yield* root;
+		if (Option.isSome(session) && Option.isSome(cwd)) {
+			return yield* new InvalidInputError({ message: "--cwd can only be used when creating a new session" });
+		}
+		const program = Effect.gen(function* () {
+			const handle = Option.isSome(session)
+				? yield* Session.attach({ sessionId: Session.SessionSchema.ID.make(session.value) })
+				: yield* Session.create({
+						title: "CLI",
+						...(Option.isNone(cwd) ? {} : { directory: resolve(cwd.value) }),
+					});
+			const ended = yield* Queue.unbounded<string>();
+			const printer = yield* handle
+				.events()
+				.pipe(Stream.runForEach(render(ended)), Effect.forkScoped({ startImmediately: true }));
+
+			yield* writeError(`session ${handle.id}\n`);
+			yield* handle.run(prompt);
+			const path = yield* handle.path();
+			const leaf = path.at(-1);
+			if (leaf !== undefined) yield* awaitMessage(ended, leaf.entry.id);
+			yield* Fiber.interrupt(printer);
+			yield* writeOut("\n");
+		});
+
+		return yield* program.pipe(
+			Effect.provide(
+				Harness.layer({
+					...(Option.isNone(shared.home) ? {} : { home: shared.home.value }),
+					...(Option.isNone(shared.database) ? {} : { database: shared.database.value }),
+				}),
+			),
+			Effect.scoped,
+		);
+	}),
+).pipe(
+	Command.withDescription("Start or continue a local agent session"),
+	Command.withExamples([
+		{ command: 'codework run "Inspect the failing tests"', description: "Create a session" },
+		{ command: 'codework run --session <id> "Now fix them"', description: "Continue a session" },
+	]),
+);
+
+export const command = root.pipe(Command.withSubcommands([run]));
+
+export const main: Effect.Effect<void, Command.Error<typeof command> | CliError.CliError> = Command.run(command, {
+	version: "0.0.1",
+}).pipe(Effect.provide(NodeServices.layer));
+
+NodeRuntime.runMain(main);
diff --git a/packages/harness/src/control.ts b/packages/harness/src/control.ts
@@ -15,7 +15,7 @@
  */
 
 import { Context, Effect, Layer, Option, Schema } from "effect";
-import { RunnerExecution } from "./runner/execution.ts";
+import * as RunnerExecution from "./runner/execution.ts";
 import type { Runner } from "./runner/run.ts";
 import { SessionInput } from "./session/input/input.ts";
 import type { Admitted } from "./session/input/schema.ts";
@@ -54,6 +54,10 @@ export interface Interface {
 	 * that window recoverable rather than a user message nothing ever answered.
 	 */
 	readonly prompt: (input: PromptInput) => Effect.Effect<Admitted, Session.SessionNotFoundError | PromptConflictError>;
+	/** Records a prompt and awaits the drain generation responsible for it. */
+	readonly run: (
+		input: PromptInput,
+	) => Effect.Effect<Admitted, Session.SessionNotFoundError | PromptConflictError | Runner.RunError>;
 	/** Starts a drain while the session is idle, or joins the one already running. */
 	readonly resume: (
 		sessionId: SessionSchema.ID,
@@ -80,39 +84,44 @@ export const layer = Layer.effect(
 		 * which is idempotent on the id, so a retry converges rather than
 		 * duplicating.
 		 */
+		const admit = Effect.fn("Control.admit")(function* (input: PromptInput) {
+			const session = yield* sessions.get(input.sessionId);
+			if (Option.isNone(session)) {
+				return yield* new Session.SessionNotFoundError({ sessionId: input.sessionId });
+			}
+
+			const messageId = input.id ?? SessionMessageSchema.ID.create();
+			const delivery = input.delivery ?? "steer";
+			const admitted = yield* inputs
+				.admit({ id: messageId, sessionId: input.sessionId, prompt: input.prompt, delivery })
+				.pipe(
+					// The id already left the inbox for the conversation. That is a
+					// conflict the caller can see, not a broken invariant.
+					Effect.catchDefect((defect) =>
+						isLifecycleConflict(defect)
+							? new PromptConflictError({ sessionId: input.sessionId, messageId })
+							: Effect.die(defect),
+					),
+				);
+
+			// A retry that reuses an id but changes the prompt is not the same
+			// request. The stored row wins, and the caller is told.
+			if (!SessionInput.equivalent(admitted, { sessionId: input.sessionId, prompt: input.prompt, delivery })) {
+				return yield* new PromptConflictError({ sessionId: input.sessionId, messageId });
+			}
+
+			return admitted;
+		});
+
 		const prompt = Effect.fn("Control.prompt")((input: PromptInput) =>
-			Effect.uninterruptible(
+			Effect.uninterruptible(admit(input).pipe(Effect.tap((admitted) => execution.wake(admitted.sessionId)))),
+		);
+
+		const run = Effect.fn("Control.run")((input: PromptInput) =>
+			Effect.uninterruptibleMask((restore) =>
 				Effect.gen(function* () {
-					const session = yield* sessions.get(input.sessionId);
-					if (Option.isNone(session)) {
-						return yield* new Session.SessionNotFoundError({ sessionId: input.sessionId });
-					}
-
-					const messageId = input.id ?? SessionMessageSchema.ID.create();
-					const delivery = input.delivery ?? "steer";
-					const admitted = yield* inputs
-						.admit({ id: messageId, sessionId: input.sessionId, prompt: input.prompt, delivery })
-						.pipe(
-							// The id already left the inbox for the conversation. That is a
-							// conflict the caller can see, not a broken invariant.
-							Effect.catchDefect((defect) =>
-								isLifecycleConflict(defect)
-									? new PromptConflictError({ sessionId: input.sessionId, messageId })
-									: Effect.die(defect),
-							),
-						);
-
-					// A retry that reuses an id but changes the prompt is not the same
-					// request. The stored row wins, and the caller is told.
-					if (!SessionInput.equivalent(admitted, { sessionId: input.sessionId, prompt: input.prompt, delivery })) {
-						return yield* new PromptConflictError({ sessionId: input.sessionId, messageId });
-					}
-
-					// A lossy signal over durable truth: the wake carries no payload, and
-					// a drain that misses it loses nothing, because the inbox still holds
-					// the prompt for the next pass. Retries wake too — a crash between
-					// admit and drain is exactly when the second call needs to start one.
-					yield* execution.wake(admitted.sessionId);
+					const admitted = yield* admit(input);
+					yield* restore(execution.drain(admitted.sessionId));
 					return admitted;
 				}),
 			),
@@ -130,7 +139,7 @@ export const layer = Layer.effect(
 			Effect.uninterruptible(execution.interrupt(sessionId)),
 		);
 
-		return Service.of({ prompt, resume, interrupt, active: execution.active });
+		return Service.of({ prompt, run, resume, interrupt, active: execution.active });
 	}),
 );
 

diff --git a/packages/harness/src/db/db.ts b/packages/harness/src/db/db.ts
@@ -1,6 +1,5 @@
-/* oxlint-disable effecttsgo/process-env -- This module is the process configuration boundary for the database. */
 import { SqliteClient, SqliteMigrator } from "@effect/sql-sqlite-node";
-import { Effect, Layer, String as Str } from "effect";
+import { Config, Effect, Layer, String as Str } from "effect";
 import { Migrator, SqlClient } from "effect/unstable/sql";
 import * as fs from "node:fs/promises";
 import { dirname, isAbsolute, join } from "node:path";
@@ -44,16 +43,18 @@ export function layer(location: string) {
 	);
 }
 
-export function path() {
-	const envPath = process.env.CODEWORK_DB;
-	if (envPath) {
-		if (envPath === ":memory:" || isAbsolute(envPath)) return envPath;
-		return join(Global.Path.data, envPath);
-	}
-	return join(Global.Path.data, "codework.db");
+export const locationConfig = Config.string("CODEWORK_DB").pipe(Config.withDefault("codework.db"));
+
+export function resolveLocation(configured: string, data: string) {
+	if (configured === ":memory:" || isAbsolute(configured)) return configured;
+	return join(data, configured);
 }
 
-// Deferred so `path()` reads CODEWORK_DB at construction time, not at import.
-export const defaultLayer = Layer.unwrap(Effect.sync(() => layer(path())));
+export const path = Effect.fn("Database.path")(function* (data?: string) {
+	const directory = data ?? (yield* Global.resolve()).data;
+	return resolveLocation(yield* locationConfig, directory);
+});
+
+export const defaultLayer = Layer.unwrap(path().pipe(Effect.map(layer)));
 
 export * as Database from "./db.ts";
diff --git a/packages/harness/src/effect.ts b/packages/harness/src/effect.ts
@@ -0,0 +1,5 @@
+export { Harness } from "./effect/harness.ts";
+export { Sandbox } from "./effect/sandbox.ts";
+export { Session } from "./effect/session.ts";
+export * as Tool from "./tools/tool.ts";
+export { Tools } from "./effect/tools.ts";
diff --git a/packages/harness/src/effect/harness.ts b/packages/harness/src/effect/harness.ts
@@ -0,0 +1,49 @@
+import { Effect, Layer } from "effect";
+import { Context } from "../context/context.ts";
+import { Control } from "../control.ts";
+import { Database } from "../db/db.ts";
+import { Event } from "../event/event.ts";
+import { Global } from "../global.ts";
+import { RunnerExecute } from "../runner/execute.ts";
+import { LLM } from "../runner/llm.ts";
+import { Loop } from "../runner/loop.ts";
+import { SandboxController } from "../sandbox/control.ts";
+import { SandboxDriver } from "../sandbox/driver.ts";
+import { SessionLive } from "../session/live.ts";
+import { SessionRuntime } from "../session/runtime.ts";
+import { State } from "../state/state.ts";
+import type { Driver } from "./sandbox.ts";
+
+export interface Options {
+	readonly database?: string;
+	readonly home?: string;
+	readonly sandboxes?: ReadonlyArray<Driver>;
+	readonly llm?: LLM.Open;
+}
+
+export const layer = (options: Options = {}) =>
+	Layer.unwrap(
+		Effect.gen(function* () {
+			const paths = yield* Global.resolve(options.home === undefined ? {} : { home: options.home });
+			const configuredDatabase = options.database ?? (yield* Database.locationConfig);
+			const global = Global.layerWith(paths);
+			const database = Database.layer(Database.resolveLocation(configuredDatabase, paths.data));
+			const drivers = SandboxDriver.layer(...(options.sandboxes ?? []));
+			const sandboxes = SandboxController.layer().pipe(Layer.provideMerge(drivers), Layer.provideMerge(database));
+			const loop = options.llm === undefined ? Loop.layer() : Loop.layer({ request: LLM.make(options.llm) });
+
+			return Control.layer.pipe(
+				Layer.provideMerge(RunnerExecute.layer.pipe(Layer.provide(loop))),
+				Layer.provideMerge(State.layer()),
+				Layer.provideMerge(SessionRuntime.layer),
+				Layer.provideMerge(sandboxes),
+				Layer.provideMerge(Context.layer),
+				Layer.provideMerge(SessionLive.layer),
+				Layer.provideMerge(Event.layer),
+				Layer.provideMerge(database),
+				Layer.provideMerge(global),
+			);
+		}),
+	);
+
+export * as Harness from "./harness.ts";