From 3ed3fa213f3fad33cc035821816377e9e91bb167 Mon Sep 17 00:00:00 2001 From: c0da Date: Wed, 5 Aug 2026 22:27:43 -0400 Subject: [PATCH 1/2] feat: stream Pi session turns --- .mise/tasks/check | 4 +- README.md | 13 +- bun.lock | 28 +++++ mise.toml | 1 + package.json | 11 +- src/session/pi-session-stream.ts | 136 +++++++++++++++++++++ test/session/pi-session-stream.test.ts | 159 +++++++++++++++++++++++++ tsconfig.json | 15 +++ 8 files changed, 360 insertions(+), 7 deletions(-) create mode 100644 bun.lock create mode 100644 src/session/pi-session-stream.ts create mode 100644 test/session/pi-session-stream.test.ts create mode 100644 tsconfig.json diff --git a/.mise/tasks/check b/.mise/tasks/check index e79b15f..ce2bfdd 100755 --- a/.mise/tasks/check +++ b/.mise/tasks/check @@ -3,4 +3,6 @@ #MISE dir="{{config_root}}" set -euo pipefail -exec codebase lint "$MISE_CONFIG_ROOT" +bun run check +codebase lint "$MISE_CONFIG_ROOT" +git diff --check diff --git a/README.md b/README.md index 4682d69..af899c3 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,11 @@ # Stein -Stein is being shaped as a TypeScript package for Pi. It currently contains no -extension or application behavior. The first repository slices establish one -reviewable development contract at a time. +Stein is a TypeScript harness for building applications around Pi sessions. +The first runtime primitive wraps an injected Pi-like session, runs one prompt +at a time, and exposes ordered assistant text deltas as an async stream. + +The real Pi SDK factory, transport adapters, persistence, prompts, and product +behavior remain separate review boundaries. ## Local checks @@ -14,9 +17,9 @@ mise install mise run check ``` -`mise run check` applies the repository's configured +`mise run check` runs strict TypeScript checking, deterministic Bun tests, [KnickKnackLabs/codebase](https://github.com/KnickKnackLabs/codebase) convention -lints. CI, runtime behavior, and Pi resources remain later review boundaries. +lints, and a whitespace check. CI invokes this same public command. An optional local pre-commit hook can run the same configured lints: diff --git a/bun.lock b/bun.lock new file mode 100644 index 0000000..183473b --- /dev/null +++ b/bun.lock @@ -0,0 +1,28 @@ +{ + "lockfileVersion": 1, + "configVersion": 1, + "workspaces": { + "": { + "name": "@knickknacklabs/stein", + "devDependencies": { + "@types/bun": "1.3.2", + "typescript": "5.9.3", + }, + }, + }, + "packages": { + "@types/bun": ["@types/bun@1.3.2", "", { "dependencies": { "bun-types": "1.3.2" } }, "sha512-t15P7k5UIgHKkxwnMNkJbWlh/617rkDGEdSsDbu+qNHTaz9SKf7aC8fiIlUdD5RPpH6GEkP0cK7WlvmrEBRtWg=="], + + "@types/node": ["@types/node@26.1.2", "", { "dependencies": { "undici-types": "~8.3.0" } }, "sha512-Vu4a5UFA9rIIFJ7rB/Vaafh9lrCQszopTCx6KjFboXTGQbPNasehVR5TEiithSDGyd1DEiUByggTZsg8jukeIg=="], + + "@types/react": ["@types/react@19.2.18", "", { "dependencies": { "csstype": "^3.2.2" } }, "sha512-AnzbBERsrLKtk2XSfTbYRLjQPdy116Sty4q+T+Bp3IC4l6jNBvreVPAHmpq9qhXQM7CXZPjLVmGMw9sy+hxQ3w=="], + + "bun-types": ["bun-types@1.3.2", "", { "dependencies": { "@types/node": "*" }, "peerDependencies": { "@types/react": "^19" } }, "sha512-i/Gln4tbzKNuxP70OWhJRZz1MRfvqExowP7U6JKoI8cntFrtxg7RJK3jvz7wQW54UuvNC8tbKHHri5fy74FVqg=="], + + "csstype": ["csstype@3.2.3", "", {}, "sha512-z1HGKcYy2xA8AGQfwrn0PAy+PB7X/GSj3UVJW9qKyn43xWa+gl5nXmU4qqLMRzWVLFC8KusUX8T/0kCiOYpAIQ=="], + + "typescript": ["typescript@5.9.3", "", { "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" } }, "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw=="], + + "undici-types": ["undici-types@8.3.0", "", {}, "sha512-j375ScV60dom+YkPFIfTLcOiPxkN/buHz5GobjLhixFuANaNs3C9l4GmrWqejgXWJ7BbJcFYpTEUkS1Ge8bpZQ=="], + } +} diff --git a/mise.toml b/mise.toml index ac1de68..212b6bc 100644 --- a/mise.toml +++ b/mise.toml @@ -7,6 +7,7 @@ experimental = true shiv = "https://github.com/KnickKnackLabs/vfox-shiv" [tools] +bun = "1.3.12" "shiv:codebase" = "0.4" [_.codebase] diff --git a/package.json b/package.json index 6f5c397..955645a 100644 --- a/package.json +++ b/package.json @@ -2,12 +2,21 @@ "name": "@knickknacklabs/stein", "private": true, "version": "0.0.0", - "description": "An intentionally behavior-free TypeScript Pi package skeleton.", + "description": "A TypeScript harness for building applications around Pi sessions.", "keywords": [ "pi-package" ], "type": "module", + "scripts": { + "check": "bun run typecheck && bun test", + "test": "bun test", + "typecheck": "tsc --noEmit" + }, "pi": { "extensions": [] + }, + "devDependencies": { + "@types/bun": "1.3.2", + "typescript": "5.9.3" } } diff --git a/src/session/pi-session-stream.ts b/src/session/pi-session-stream.ts new file mode 100644 index 0000000..3c2a955 --- /dev/null +++ b/src/session/pi-session-stream.ts @@ -0,0 +1,136 @@ +export interface PiSessionBackend { + subscribe(listener: (event: unknown) => void): () => void; + prompt(input: string): Promise; + abort?(): Promise; + dispose(): void; +} + +type Waiter = { + resolve(result: IteratorResult): void; + reject(error: unknown): void; +}; + +class AsyncTextQueue implements AsyncIterable { + readonly #values: string[] = []; + readonly #waiters: Waiter[] = []; + #closed = false; + #failed = false; + #failure: unknown; + + push(value: string): void { + if (this.#closed) return; + const waiter = this.#waiters.shift(); + if (waiter) waiter.resolve({ value, done: false }); + else this.#values.push(value); + } + + close(): void { + if (this.#closed) return; + this.#closed = true; + this.#settle(); + } + + fail(error: unknown): void { + if (this.#closed) return; + this.#failed = true; + this.#failure = error; + this.#closed = true; + this.#settle(); + } + + async *[Symbol.asyncIterator](): AsyncIterator { + while (true) { + const result = await this.#next(); + if (result.done) return; + yield result.value; + } + } + + #next(): Promise> { + const value = this.#values.shift(); + if (value !== undefined) return Promise.resolve({ value, done: false }); + if (this.#failed) return Promise.reject(this.#failure); + if (this.#closed) return Promise.resolve({ value: undefined, done: true }); + return new Promise((resolve, reject) => this.#waiters.push({ resolve, reject })); + } + + #settle(): void { + while (this.#waiters.length > 0) { + const waiter = this.#waiters.shift(); + if (!waiter) return; + if (this.#failed) waiter.reject(this.#failure); + else waiter.resolve({ value: undefined, done: true }); + } + } +} + +export class PiSessionStream { + readonly #session: PiSessionBackend; + #active = false; + #disposed = false; + + constructor(session: PiSessionBackend) { + this.#session = session; + } + + async *run(prompt: string): AsyncIterable { + if (this.#disposed) throw new Error("Pi session stream is disposed"); + if (this.#active) throw new Error("Pi session stream already has an active turn"); + if (!prompt.trim()) throw new Error("Prompt must not be empty"); + + this.#active = true; + const output = new AsyncTextQueue(); + let unsubscribe: () => void; + try { + unsubscribe = this.#session.subscribe((event) => { + const delta = textDelta(event); + if (delta !== undefined) output.push(delta); + }); + } catch (error) { + this.#active = false; + throw error; + } + + let completion: Promise; + try { + completion = this.#session.prompt(prompt); + } catch (error) { + unsubscribe(); + this.#active = false; + throw error; + } + completion.then(() => output.close(), (error: unknown) => output.fail(error)); + + let completed = false; + try { + for await (const chunk of output) yield chunk; + await completion; + completed = true; + } finally { + unsubscribe(); + if (!completed) await this.#session.abort?.(); + this.#active = false; + } + } + + async abort(): Promise { + await this.#session.abort?.(); + } + + dispose(): void { + if (this.#disposed) return; + this.#disposed = true; + this.#session.dispose(); + } +} + +function textDelta(event: unknown): string | undefined { + if (!isRecord(event) || event.type !== "message_update") return undefined; + const update = event.assistantMessageEvent; + if (!isRecord(update) || update.type !== "text_delta") return undefined; + return typeof update.delta === "string" ? update.delta : undefined; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null; +} diff --git a/test/session/pi-session-stream.test.ts b/test/session/pi-session-stream.test.ts new file mode 100644 index 0000000..be9b270 --- /dev/null +++ b/test/session/pi-session-stream.test.ts @@ -0,0 +1,159 @@ +import { describe, expect, test } from "bun:test"; +import { + PiSessionStream, + type PiSessionBackend, +} from "../../src/session/pi-session-stream.ts"; + +class Deferred { + readonly promise: Promise; + resolve!: () => void; + + constructor() { + this.promise = new Promise((resolve) => { + this.resolve = resolve; + }); + } +} + +class FakeSession implements PiSessionBackend { + readonly prompts: string[] = []; + readonly #listeners = new Set<(event: unknown) => void>(); + promptHandler: (input: string) => Promise = async () => {}; + abortCount = 0; + disposeCount = 0; + unsubscribeCount = 0; + + subscribe(listener: (event: unknown) => void): () => void { + this.#listeners.add(listener); + return () => { + this.#listeners.delete(listener); + this.unsubscribeCount += 1; + }; + } + + prompt(input: string): Promise { + this.prompts.push(input); + return this.promptHandler(input); + } + + emit(event: unknown): void { + for (const listener of this.#listeners) listener(event); + } + + async abort(): Promise { + this.abortCount += 1; + } + + dispose(): void { + this.disposeCount += 1; + } +} + +async function collect(chunks: AsyncIterable): Promise { + const output: string[] = []; + for await (const chunk of chunks) output.push(chunk); + return output; +} + +function textDelta(delta: string): unknown { + return { + type: "message_update", + assistantMessageEvent: { type: "text_delta", delta }, + }; +} + +describe("PiSessionStream", () => { + test("forwards the prompt and streams ordered text deltas", async () => { + const session = new FakeSession(); + session.promptHandler = async () => { + session.emit({ type: "message_update", assistantMessageEvent: { type: "thinking_delta" } }); + session.emit(textDelta("one ")); + session.emit(textDelta("two")); + }; + const stream = new PiSessionStream(session); + + expect(await collect(stream.run("A fictional prompt"))).toEqual(["one ", "two"]); + expect(session.prompts).toEqual(["A fictional prompt"]); + expect(session.unsubscribeCount).toBe(1); + }); + + test("rejects empty prompts before touching the session", async () => { + const session = new FakeSession(); + const stream = new PiSessionStream(session); + + await expect(collect(stream.run(" "))).rejects.toThrow("Prompt must not be empty"); + expect(session.prompts).toEqual([]); + }); + + test("prevents overlapping turns", async () => { + const session = new FakeSession(); + const deferred = new Deferred(); + session.promptHandler = () => { + session.emit(textDelta("started")); + return deferred.promise; + }; + const stream = new PiSessionStream(session); + const first = stream.run("first")[Symbol.asyncIterator](); + + expect(await first.next()).toEqual({ value: "started", done: false }); + await expect(collect(stream.run("second"))).rejects.toThrow("already has an active turn"); + deferred.resolve(); + expect(await first.next()).toEqual({ value: undefined, done: true }); + }); + + test("propagates failures and releases the active-turn guard", async () => { + const session = new FakeSession(); + session.promptHandler = async () => { + throw new Error("deterministic failure"); + }; + const stream = new PiSessionStream(session); + + await expect(collect(stream.run("first"))).rejects.toThrow("deterministic failure"); + session.promptHandler = async () => {}; + expect(await collect(stream.run("second"))).toEqual([]); + }); + + test("preserves undefined rejection values", async () => { + const session = new FakeSession(); + session.promptHandler = () => Promise.reject(undefined); + const stream = new PiSessionStream(session); + let rejected = false; + + try { + await collect(stream.run("prompt")); + } catch (error) { + rejected = true; + expect(error).toBeUndefined(); + } + expect(rejected).toBeTrue(); + }); + + test("aborts when the consumer stops before prompt completion", async () => { + const session = new FakeSession(); + const deferred = new Deferred(); + session.promptHandler = () => { + session.emit(textDelta("first")); + return deferred.promise; + }; + const stream = new PiSessionStream(session); + const iterator = stream.run("prompt")[Symbol.asyncIterator](); + + expect(await iterator.next()).toEqual({ value: "first", done: false }); + await iterator.return?.(); + expect(session.abortCount).toBe(1); + expect(session.unsubscribeCount).toBe(1); + deferred.resolve(); + }); + + test("delegates explicit abort and disposes once", async () => { + const session = new FakeSession(); + const stream = new PiSessionStream(session); + + await stream.abort(); + stream.dispose(); + stream.dispose(); + expect(session.abortCount).toBe(1); + expect(session.disposeCount).toBe(1); + await expect(collect(stream.run("after disposal"))).rejects.toThrow("disposed"); + }); +}); diff --git a/tsconfig.json b/tsconfig.json new file mode 100644 index 0000000..408b2a1 --- /dev/null +++ b/tsconfig.json @@ -0,0 +1,15 @@ +{ + "compilerOptions": { + "allowImportingTsExtensions": true, + "exactOptionalPropertyTypes": true, + "module": "Preserve", + "moduleResolution": "Bundler", + "noEmit": true, + "noUncheckedIndexedAccess": true, + "skipLibCheck": true, + "strict": true, + "target": "ESNext", + "types": ["bun"] + }, + "include": ["src", "test"] +} From abef9dabb0f032b55cc86831d8ddee4c4e29570f Mon Sep 17 00:00:00 2001 From: c0da Date: Wed, 5 Aug 2026 22:30:09 -0400 Subject: [PATCH 2/2] ci: install locked package dependencies --- .github/workflows/check.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index 3c32304..1b569b3 100644 --- a/.github/workflows/check.yml +++ b/.github/workflows/check.yml @@ -25,5 +25,8 @@ jobs: with: install: true + - name: Install package dependencies + run: bun install --frozen-lockfile + - name: Run local checks run: mise run check