From 1e3da9ca67ada088753767d9d5259b5f5867a28d Mon Sep 17 00:00:00 2001 From: "operator-stack-publisher[bot]" Date: Thu, 23 Jul 2026 12:57:22 +0000 Subject: [PATCH] Sync Pitot from Intelligence Flow Labs @ 67082ad54276 --- README.md | 75 +++++- UPSTREAM.json | 26 +- cmd/pitot/main.go | 6 +- cmd/pitot/workbench.go | 503 +++++++++++++++++++++++++++++------ cmd/pitot/workbench_test.go | 151 +++++++++++ runtime/runtime.go | 35 ++- sdk/python/pyproject.toml | 2 +- sdk/rust/Cargo.toml | 14 + sdk/rust/src/runner.rs | 79 ++++-- sdk/tests/go_test.go | 13 - sdk/tests/go_test_runner.go | 13 - sdk/tests/python_test.py | 10 - sdk/tests/test_runners.py | 382 +++++++++++++++++++++----- sdk/tests/ts_test.ts | 7 - sdk/typescript/package.json | 12 +- sdk/typescript/src/runner.ts | 35 ++- sdk/typescript/tsconfig.json | 3 +- 17 files changed, 1099 insertions(+), 267 deletions(-) create mode 100644 cmd/pitot/workbench_test.go create mode 100644 sdk/rust/Cargo.toml delete mode 100644 sdk/tests/go_test.go delete mode 100644 sdk/tests/go_test_runner.go delete mode 100644 sdk/tests/python_test.py delete mode 100644 sdk/tests/ts_test.ts diff --git a/README.md b/README.md index 01be6c6..d175b0e 100644 --- a/README.md +++ b/README.md @@ -158,13 +158,79 @@ Download a release binary for macOS, Linux, or Windows, or install from source: go install github.com/operatorstack/pitot/cmd/pitot@latest ``` -Inspect the effective local boundary: +Inspect the effective local boundary at any time: ```bash pitot doctor ``` -Start Pitot with repository-owned configuration and an owner-only runtime +## Quickstart + +From a clean repository to one real allow/deny decision in two commands. + +**1. Scaffold a project.** `pitot init` detects the language from the files +already in the directory, or prompts you to choose when it cannot. It writes a +runnable project — source, package manifest, and `.pitot.yaml` — and never +overwrites existing files unless you pass `--force`: + +```bash +pitot init +``` + +``` +Detected python project in . +Initialized python controller in . +Files written: .pitot.yaml, main.py, pyproject.toml, requirements.txt +Next: cd . && pitot dev --host claude --exec "python3 main.py" +``` + +You can skip detection and prompts with flags — handy for CI: + +```bash +pitot init --language python --role controller --dir ./approval +``` + +The four first-class languages (`python`, `typescript`, `go`, `rust`) each +generate a complete project: `python3 main.py`, `npx tsx main.ts`, +`go run main.go`, and `cargo run` all work after installing dependencies. + +**2. Run it against an agent.** `pitot dev` starts the runtime on a private +loopback endpoint, waits until it is ready, launches your Controller, and prints +each decision as the agent makes it. `--exec` takes the full command line (or use +`-- CMD ARGS`): + +```bash +pitot dev --host claude --exec "python3 main.py" +``` + +``` +Starting Pitot dev environment for host claude... +Runtime ready. Starting agent: python3 main.py +Decisions: + [ALLOW] release.approval (act_7f2) — v1.4.0 is approved for publication. + [DENY] shell.exec (act_1a9) — destructive command blocked +Agent finished. Runtime stopped. +``` + +`--host` must name a supported agent (`claude`, `codex`, `copilot`, `cursor`, +`gemini`, `kimi`, `opencode`, `pi`, `qwen`). The runtime descriptor lives in a +per-invocation temporary path and is removed on exit, so concurrent `pitot dev` +sessions never collide. + +**3. Swap the agent.** The same project — the same Controller and `.pitot.yaml` — +works with any other supported host. Change only `--host`: + +```bash +pitot dev --host cursor --exec "python3 main.py" +``` + +The boundary is language- and agent-neutral: one Controller, every agent. + +## Advanced: manual runtime + +`pitot dev` is the recommended path. If you need to manage the runtime yourself +(for example, sharing one runtime across several long-lived agent sessions), +start it with repository-owned configuration and an owner-only runtime descriptor: ```bash @@ -185,6 +251,11 @@ $env:PITOT_RUNTIME = Join-Path $env:LOCALAPPDATA "Pitot\project.json" pitot run --config .pitot.yaml --runtime $env:PITOT_RUNTIME ``` +## Connect your agent + +The per-host hooks below wire each agent's native blocking boundary to Pitot for +the manual runtime flow. `pitot dev` configures the selected `--host` for you. + ### Kimi Code Install Kimi Code on macOS or Linux using its official installer: diff --git a/UPSTREAM.json b/UPSTREAM.json index 6690ecf..4736af1 100644 --- a/UPSTREAM.json +++ b/UPSTREAM.json @@ -1,7 +1,7 @@ { "files": { "CONTRIBUTING.md": "23728d8a132d62b8adfb2e5c3eb9d9bfcf8a4d04543765b1e22ad8d55424af8f", - "README.md": "091e47c27d0ee66a533b9aa9f3046a1bc9b84c0db5021926c6ab573d465ca57a", + "README.md": "77995d36de1a6ac5f5c687fad4a152b8824ea3ad25abe65325a7c4ee9a2ef52d", "adapter-verification.json": "f8ad4e206571650f698826a8b66d8c00822be425e8d2de8ae98d98239e575eb4", "adapters/adapters.go": "1b46ba131fa3b2c93eed23526330275a3506451ba4bbd4f497e5378dfab2b6a8", "assets/pitot-boundary.png": "8a0ddb7d81831d94e14813f50ea4ca8670d77417f339ed2f91f0c653bf52f41d", @@ -14,9 +14,10 @@ "bridge/bridge.go": "5adfcd3f743cae46e4446a6e030d53464ada97de0261a8588fa2a9fcd62136b8", "bridge/bridge_test.go": "6dcc6d05f2b39c25955fc0b2d21d3d148dd9d77600fb12799941f86bdb1acb61", "cmd/generate-schema/main.go": "6e9d0030290d99e36967433f96e38385a122974f899ad9421aac1ef7e50d8fcb", - "cmd/pitot/main.go": "d78ce2331aed6db847b29bd683bb6df57eb9e3bd82fdb438b29371c02a893a88", + "cmd/pitot/main.go": "14de1d6bf7ef7a75172ffd015c7cab1b66c38706b879de8d354137b34596c7f3", "cmd/pitot/main_test.go": "544997295e0c4b75ef8f3d698b3de0883153f671b6b8f62057cc6e3452d6dc93", - "cmd/pitot/workbench.go": "3b4fe94e4663842194e9439af3a4049a86cf71fc2b38aac8af8ca1f667a59999", + "cmd/pitot/workbench.go": "2e2522491437c624b241fd594d678be2dab6824fc5aa3235db1ec82e00a669c7", + "cmd/pitot/workbench_test.go": "457caa11cd4b73c1fb4e0dad806b3050b196ddac690a8125f1615a4c695cc073", "config/config.go": "e6666567d0c0cca41de69361e8f1243adda1ec0a54a9300b39a84d2290bff319", "config/config_test.go": "87d3e5ddc4a3b43c736070de671d03e03ffe29cdd759771526ad27fd9bc0034c", "conformance/conformance.go": "43b692114f45c8b52958e34b35aee1cee339d8321c90f92ab4f5b963e79935bb", @@ -49,7 +50,7 @@ "runtime/capabilities.go": "76c27bdd7ddfa11a1d639915ac2eb2d573b7ccdd687a5f548b3731b7c1e1828f", "runtime/descriptor_unix.go": "df41b6867e9840933f186c93b6bc61861e7ea5252a365455686ea0414bfa0044", "runtime/descriptor_windows.go": "2d9ffefe3af0154fa8042de6b67460d4e86dd3f4cdd9e986f180f7d0c535c9a5", - "runtime/runtime.go": "693189eadd040629b616614e25dcdebb954e9c3546c25d064d0b8c3a7d01394a", + "runtime/runtime.go": "e0624a16ac9f79e8080246042912c73b1588f1b9aeb4f9f95b3711c7acef7e81", "runtime/runtime_test.go": "afd78d122af20148bf30d0db873ff002544189df0dfec0f6167b8cf5cd0d42b1", "runtime/transport.go": "83e2218fb28474e875dafa6943bc5b665acef0565aaf5955fa88b1b4fd21614e", "runtime/transport_test.go": "9b69f590f1e258470adea249b3ac6d4a00f1001f10bb08dfa7b56c6e2d6709ae", @@ -69,21 +70,18 @@ "sdk/python/pitot/__init__.py": "9cab11b333536f167d4e7bb6089ef00488690ec7b5cd5701f08e8bbb13cef0f7", "sdk/python/pitot/runner.py": "8d4537ffc3aee2ea22b1d691559ac1eb1e95521df9ad5bf241ee6ac54c9925b1", "sdk/python/pitot/types.py": "c9a7221f1ad6627f152f26148155d3c74f249c52262c73a515251f469e8eb4ad", - "sdk/python/pyproject.toml": "3f42de8223640978ad88e3f7b4ad1464053c43a393f0edf48445cacd31dd537c", + "sdk/python/pyproject.toml": "4b43380a1ebc12350ee3c30633695160ba0fc0bb64501b1e16f990a6fbe56d02", "sdk/runner.go": "e9a8db96d3cf6df7ea7e661e6755650f97e580970995dfe881b4268db0c3f832", + "sdk/rust/Cargo.toml": "b8435f6c600ad0791bd29ada4bc396b14e54b96ee3804896c8e610583cc70c1a", "sdk/rust/src/lib.rs": "fd2dcb9bf9df47fb58e52e4e94136f95867616d941ba66b4b324413d1c5b1777", - "sdk/rust/src/runner.rs": "96b9a04332b03999449ab24cc4e38df68dfc575a82d8fd599ad5211c7bfeabe6", + "sdk/rust/src/runner.rs": "168eb3f926aed4b86aa27ad66ab605326974b6a345b07a35126d4b66e0968b33", "sdk/rust/src/types.rs": "82cd5a5ec12effa6688e8b9b6028bbef67ec43f24d9fd9d19b4235934061ffb8", - "sdk/tests/go_test.go": "4700cc0bcebb0be8af79e84438ef47a75d2892fcfef63352620ca22a2ac5fb33", - "sdk/tests/go_test_runner.go": "4700cc0bcebb0be8af79e84438ef47a75d2892fcfef63352620ca22a2ac5fb33", - "sdk/tests/python_test.py": "c52fae5cfa0e433ec95fa8c79a67804857facd456d75b6523d1ed16ddbb2672c", - "sdk/tests/test_runners.py": "19a6f950ee5b53b34d275ec39d3f9536bd2d693c2e11efa0e20ffe423f101f4c", - "sdk/tests/ts_test.ts": "63fe8bd68e4e8aa22d7ae24929f34c502978c19ec41c0f1645c1d513cb17329b", - "sdk/typescript/package.json": "c26c726559ff0bc7c5d92b68303efc3c35314613248fe346fa125aea06c8dba6", + "sdk/tests/test_runners.py": "9e86a03b53b108b70204e797b4c65b3d157c6765543a9bb9d7bb892daaf4d8c4", + "sdk/typescript/package.json": "8bd1fa7716b65c3c6599307d8d2e9398c75ae842c6047e9e0ad7716122db9a03", "sdk/typescript/src/index.ts": "bddb336bc120ecd979de0288e432ed2b41749d39f11c697f019af5e54c2e9fba", "sdk/typescript/src/pitot.ts": "9c243824cbb7edc54b1e125abfd828bf2ed77e7151a0bd5c5d4f63e81ca9b00c", - "sdk/typescript/src/runner.ts": "2022d4ee7efedec2c4c3baca2229440aa9773f5c32b58ffed8b1a2915add0278", - "sdk/typescript/tsconfig.json": "e4d7ecb203fcb7d93cd9b9fb235d7fd75d1b6fb2b28eff773f4aecfdc1924d5d", + "sdk/typescript/src/runner.ts": "c58babd3ec3996a05988f3cb7dec061cf7c48f41cd5125625be3408cb9b207f0", + "sdk/typescript/tsconfig.json": "3acd617cc06089125bcd963a3624131d765bb59e1a93db9736e78de6cf4fb40d", "sensor/decode_fuzz_test.go": "d27f2fbbc069eded26a73c9cd9bace98dd8a9e34949576790b81a08d130fbaf2", "sensor/sensor.go": "5c503d07ac33e7894d635f2d98bcd6d165d442d60c127ed6aeac80ec319086c5", "sensor/sensor_test.go": "4ddbdde3e486c9e189a2ed4174ad413df107d43dc98f5242da3667a24c7a5da1", diff --git a/cmd/pitot/main.go b/cmd/pitot/main.go index d2f34b9..ffa67f0 100644 --- a/cmd/pitot/main.go +++ b/cmd/pitot/main.go @@ -42,7 +42,7 @@ func runWithIO(ctx context.Context, args []string, stdin io.Reader, stdout, stde } switch args[0] { case "init": - return runInit(args[1:], stdout, stderr) + return runInit(args[1:], stdin, stdout, stderr) case "dev": return runDev(ctx, args[1:], stdout, stderr) case "doctor": @@ -283,8 +283,8 @@ func usage() string { return `pitot — the open sensor and control transport for coding-agent tooling usage: - pitot init --language [python|typescript|go|rust] --role [consumer|controller] --dir PATH - pitot dev --host HOST --exec CMD + pitot init [--language python|typescript|go|rust] [--role consumer|controller] [--dir PATH] [--force] + pitot dev --host HOST --exec "CMD ARGS" pitot doctor pitot run --config PATH --runtime PATH pitot hook HOST [--runtime PATH] diff --git a/cmd/pitot/workbench.go b/cmd/pitot/workbench.go index 7dea582..124ca23 100644 --- a/cmd/pitot/workbench.go +++ b/cmd/pitot/workbench.go @@ -1,6 +1,7 @@ package main import ( + "bufio" "context" "errors" "fmt" @@ -8,15 +9,30 @@ import ( "os" "os/exec" "path/filepath" + "strings" + "time" + "github.com/operatorstack/pitot/adapters" "github.com/operatorstack/pitot/config" "github.com/operatorstack/pitot/runtime" ) -func runInit(args []string, stdout, stderr io.Writer) error { - lang := "python" - role := "controller" +// pitotPackageVersion is the published SDK version wired into generated +// manifests so a freshly initialized project can resolve its dependency. +const pitotPackageVersion = "0.1.0" + +var supportedLanguages = []string{"python", "typescript", "go", "rust"} +var supportedRoles = []string{"consumer", "controller"} + +// runInit scaffolds a complete, runnable Pitot project. It validates its inputs, +// detects or interactively selects the language and role, refuses to overwrite +// existing files unless --force is set, and writes a package manifest alongside +// the source so the generated project builds and runs without further setup. +func runInit(args []string, stdin io.Reader, stdout, stderr io.Writer) error { + lang := "" + role := "" dir := "pitot-project" + force := false for i := 0; i < len(args); i++ { switch args[i] { @@ -38,111 +54,265 @@ func runInit(args []string, stdout, stderr io.Writer) error { } dir = args[i+1] i++ + case "--force": + force = true default: return fmt.Errorf("pitot init: unexpected argument %q", args[i]) } } - if err := os.MkdirAll(dir, 0755); err != nil { - return fmt.Errorf("failed to create project directory: %v", err) - } + interactive := isInteractive(stdin) + reader := bufio.NewReader(stdin) - switch lang { - case "python": - if role == "controller" { - err := writeTemplate(filepath.Join(dir, "main.py"), pythonControllerTemplate) + // Resolve language: explicit flag, then detection, then prompt. + if lang == "" { + if detected := detectLanguage(dir); detected != "" { + fmt.Fprintf(stdout, "Detected %s project in %s\n", detected, dir) + lang = detected + } else if interactive { + choice, err := promptChoice(reader, stdout, "Select a language", supportedLanguages, "python") if err != nil { return err } + lang = choice } else { - err := writeTemplate(filepath.Join(dir, "main.py"), pythonConsumerTemplate) - if err != nil { - return err - } + return errors.New("pitot init: --language is required when it cannot be detected and no terminal is attached") } - case "typescript": - if role == "controller" { - err := writeTemplate(filepath.Join(dir, "main.ts"), tsControllerTemplate) + } + if !contains(supportedLanguages, lang) { + return fmt.Errorf("pitot init: unsupported language %q (want python, typescript, go, rust)", lang) + } + + // Resolve role: explicit flag, then prompt, then default. + if role == "" { + if interactive { + choice, err := promptChoice(reader, stdout, "Select a role", supportedRoles, "controller") if err != nil { return err } + role = choice } else { - err := writeTemplate(filepath.Join(dir, "main.ts"), tsConsumerTemplate) - if err != nil { - return err + role = "controller" + } + } + if !contains(supportedRoles, role) { + return fmt.Errorf("pitot init: unsupported role %q (want consumer, controller)", role) + } + + files, err := projectFiles(lang, role) + if err != nil { + return err + } + + // Non-destructive: refuse to clobber existing files unless --force. + if !force { + var conflicts []string + for name := range files { + if _, statErr := os.Stat(filepath.Join(dir, name)); statErr == nil { + conflicts = append(conflicts, name) } } - case "go": - if role == "controller" { - err := writeTemplate(filepath.Join(dir, "main.go"), goControllerTemplate) - if err != nil { - return err + if len(conflicts) > 0 { + return fmt.Errorf("pitot init: refusing to overwrite existing files in %s: %s (use --force)", dir, strings.Join(sorted(conflicts), ", ")) + } + } + + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("pitot init: create project directory: %w", err) + } + for _, name := range sorted(keys(files)) { + path := filepath.Join(dir, name) + if parent := filepath.Dir(path); parent != dir { + if err := os.MkdirAll(parent, 0o755); err != nil { + return fmt.Errorf("pitot init: create %s: %w", parent, err) } + } + if err := os.WriteFile(path, []byte(files[name]), 0o644); err != nil { + return fmt.Errorf("pitot init: write %s: %w", name, err) + } + } + + fmt.Fprintf(stdout, "Initialized %s %s in %s\n", lang, role, dir) + fmt.Fprintf(stdout, "Files written: %s\n", strings.Join(sorted(keys(files)), ", ")) + fmt.Fprintf(stdout, "Next: cd %s && pitot dev --host claude --exec %q\n", dir, runCommandString(lang)) + return nil +} + +// projectFiles returns the complete file set for a language/role: source, +// package manifest(s), and the .pitot.yaml runtime configuration. +func projectFiles(lang, role string) (map[string]string, error) { + files := map[string]string{} + + controller := role == "controller" + switch lang { + case "python": + if controller { + files["main.py"] = pythonControllerTemplate } else { - err := writeTemplate(filepath.Join(dir, "main.go"), goConsumerTemplate) - if err != nil { - return err - } + files["main.py"] = pythonConsumerTemplate + } + files["requirements.txt"] = fmt.Sprintf("pitot>=%s\n", pitotPackageVersion) + files["pyproject.toml"] = pythonProjectManifest + case "typescript": + if controller { + files["main.ts"] = tsControllerTemplate + } else { + files["main.ts"] = tsConsumerTemplate } + files["package.json"] = tsProjectManifest + files["tsconfig.json"] = tsProjectTSConfig + case "go": + if controller { + files["main.go"] = goControllerTemplate + } else { + files["main.go"] = goConsumerTemplate + } + files["go.mod"] = goProjectManifest case "rust": - if role == "controller" { - err := writeTemplate(filepath.Join(dir, "main.rs"), rustControllerTemplate) - if err != nil { - return err - } + if controller { + files["main.rs"] = rustControllerTemplate } else { - err := writeTemplate(filepath.Join(dir, "main.rs"), rustConsumerTemplate) - if err != nil { - return err - } + files["main.rs"] = rustConsumerTemplate } + files["Cargo.toml"] = rustProjectManifest default: - return fmt.Errorf("unsupported language: %s", lang) + return nil, fmt.Errorf("pitot init: unsupported language %q", lang) } - // Write a basic pitot configuration - cfg := `controllers: + files[".pitot.yaml"] = pitotConfig(lang, role) + return files, nil +} + +// pitotConfig renders the .pitot.yaml wiring the generated role to its run command. +func pitotConfig(lang, role string) string { + cmdList := runCommandList(lang) + if role == "consumer" { + return `consumers: + - id: local-consumer + command: ` + cmdList + ` + events: ["action.requested"] + projection: + content: full +` + } + return fmt.Sprintf(`controllers: test.approval: id: local-controller command: %s deadline_ms: 2000 on_timeout: deny on_unavailable: deny -` - var cmdList string +`, cmdList) +} + +// runCommandList is the JSON array form embedded in .pitot.yaml. +func runCommandList(lang string) string { switch lang { case "python": - cmdList = `["python3", "main.py"]` + return `["python3", "main.py"]` case "typescript": - cmdList = `["npx", "tsx", "main.ts"]` + return `["npx", "tsx", "main.ts"]` case "go": - cmdList = `["go", "run", "main.go"]` + return `["go", "run", "main.go"]` case "rust": - cmdList = `["cargo", "run"]` + return `["cargo", "run", "--quiet"]` + default: + return `[]` } +} - if role == "consumer" { - cfg = `consumers: - - id: local-consumer - command: ` + cmdList + ` - events: ["action.requested"] - projection: - content: full -` - } else { - cfg = fmt.Sprintf(cfg, cmdList) +// runCommandString is the human-readable command shown in the init next-step hint. +func runCommandString(lang string) string { + switch lang { + case "python": + return "python3 main.py" + case "typescript": + return "npx tsx main.ts" + case "go": + return "go run main.go" + case "rust": + return "cargo run --quiet" + default: + return "" } +} - if err := os.WriteFile(filepath.Join(dir, ".pitot.yaml"), []byte(cfg), 0644); err != nil { - return err +// detectLanguage inspects an existing directory for a language's marker manifest. +func detectLanguage(dir string) string { + markers := []struct { + file string + lang string + }{ + {"go.mod", "go"}, + {"Cargo.toml", "rust"}, + {"package.json", "typescript"}, + {"tsconfig.json", "typescript"}, + {"pyproject.toml", "python"}, + {"requirements.txt", "python"}, + } + for _, m := range markers { + if _, err := os.Stat(filepath.Join(dir, m.file)); err == nil { + return m.lang + } } + return "" +} - fmt.Fprintf(stdout, "Initialized %s %s in %s\n", lang, role, dir) - return nil +// isInteractive reports whether r is a terminal we can prompt on. +func isInteractive(r io.Reader) bool { + f, ok := r.(*os.File) + if !ok { + return false + } + info, err := f.Stat() + if err != nil { + return false + } + return info.Mode()&os.ModeCharDevice != 0 +} + +// promptChoice asks the user to pick from options, returning def on empty input. +func promptChoice(reader *bufio.Reader, stdout io.Writer, question string, options []string, def string) (string, error) { + fmt.Fprintf(stdout, "%s [%s] (default %s): ", question, strings.Join(options, "/"), def) + line, err := reader.ReadString('\n') + if err != nil && err != io.EOF { + return "", fmt.Errorf("pitot init: read choice: %w", err) + } + choice := strings.TrimSpace(line) + if choice == "" { + return def, nil + } + if !contains(options, choice) { + return "", fmt.Errorf("pitot init: %q is not one of %s", choice, strings.Join(options, ", ")) + } + return choice, nil +} + +func contains(items []string, want string) bool { + for _, item := range items { + if item == want { + return true + } + } + return false } -func writeTemplate(path string, tmpl string) error { - return os.WriteFile(path, []byte(tmpl), 0644) +func keys(m map[string]string) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + return out +} + +func sorted(items []string) []string { + out := append([]string(nil), items...) + for i := 1; i < len(out); i++ { + for j := i; j > 0 && out[j-1] > out[j]; j-- { + out[j-1], out[j] = out[j], out[j-1] + } + } + return out } const pythonControllerTemplate = `import sys @@ -167,22 +337,61 @@ if __name__ == "__main__": run_consumer(handler) ` -const tsControllerTemplate = `import { runController, allow } from 'pitot/runner'; -import { ControlRequested } from 'pitot/pitot'; +const pythonProjectManifest = `[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "pitot-project" +version = "0.1.0" +requires-python = ">=3.10" +dependencies = ["pitot>=0.1.0"] +` + +const tsControllerTemplate = `import { runController, allow, ControlRequested } from '@operatorstack/pitot'; runController("local-controller", async (req: ControlRequested) => { return allow("Action approved by TS Controller"); }); ` -const tsConsumerTemplate = `import { runConsumer } from 'pitot/runner'; -import { Event } from 'pitot/pitot'; +const tsConsumerTemplate = `import { runConsumer, Event } from '@operatorstack/pitot'; runConsumer(async (event: Event) => { console.error("Consumed event:", event.type); }); ` +const tsProjectManifest = `{ + "name": "pitot-project", + "version": "0.1.0", + "private": true, + "scripts": { + "dev": "tsx main.ts" + }, + "dependencies": { + "@operatorstack/pitot": "^0.1.0" + }, + "devDependencies": { + "tsx": "^4.7.0", + "typescript": "^5.0.0" + } +} +` + +const tsProjectTSConfig = `{ + "compilerOptions": { + "target": "es2022", + "module": "commonjs", + "moduleResolution": "node", + "strict": true, + "esModuleInterop": true, + "skipLibCheck": true + }, + "include": ["main.ts"] +} +` + const goControllerTemplate = `package main import ( @@ -214,6 +423,13 @@ func main() { } ` +const goProjectManifest = `module pitot-project + +go 1.22 + +require github.com/operatorstack/pitot v0.1.0 +` + const rustControllerTemplate = `use pitot::{run_controller, allow, ControlRequested, Outcome}; fn handler(_req: ControlRequested) -> Outcome { @@ -236,9 +452,26 @@ fn main() { } ` +const rustProjectManifest = `[package] +name = "pitot-project" +version = "0.1.0" +edition = "2021" + +[[bin]] +name = "pitot-project" +path = "main.rs" + +[dependencies] +pitot = "0.1.0" +` + +// runDev launches the runtime and a single agent against a chosen host, waits +// for the runtime to be ready before starting the agent, renders a decision +// timeline, and cleans up its private runtime descriptor on exit. func runDev(ctx context.Context, args []string, stdout, stderr io.Writer) error { host := "" execCmd := "" + var execArgs []string for i := 0; i < len(args); i++ { switch args[i] { @@ -254,55 +487,149 @@ func runDev(ctx context.Context, args []string, stdout, stderr io.Writer) error } execCmd = args[i+1] i++ + case "--": + // Everything after -- is the agent command and its arguments. + rest := args[i+1:] + if len(rest) > 0 { + execCmd = rest[0] + execArgs = rest[1:] + } + i = len(args) default: return fmt.Errorf("pitot dev: unexpected argument %q", args[i]) } } if host == "" || execCmd == "" { - return errors.New("pitot dev: requires both --host and --exec") + return errors.New("pitot dev: requires both --host and --exec (or --exec/-- CMD ARGS)") + } + if !adapters.IsSupported(adapters.Host(host)) { + return fmt.Errorf("pitot dev: unsupported host %q (want one of: %s)", host, hostList()) } - fmt.Fprintf(stdout, "Starting Pitot dev environment for host %s...\n", host) + // If --exec carried a full command line, split it into program + args. + // An explicit `-- CMD ARGS` form (execArgs already set) takes precedence. + program := execCmd + programArgs := execArgs + if len(programArgs) == 0 { + fields := strings.Fields(execCmd) + if len(fields) == 0 { + return errors.New("pitot dev: --exec is empty") + } + program = fields[0] + programArgs = fields[1:] + } configPath := ".pitot.yaml" if _, err := os.Stat(configPath); os.IsNotExist(err) { return errors.New("pitot dev: .pitot.yaml not found in current directory. Run 'pitot init' first") } - loaded, err := config.Load(configPath) if err != nil { - return fmt.Errorf("failed to load config: %v", err) + return fmt.Errorf("pitot dev: load config: %w", err) + } + + // Private, per-invocation runtime directory so concurrent runs never collide. + runtimeDir, err := os.MkdirTemp("", "pitot-dev-") + if err != nil { + return fmt.Errorf("pitot dev: create runtime dir: %w", err) } + defer os.RemoveAll(runtimeDir) + runtimePath := filepath.Join(runtimeDir, "runtime.json") + + fmt.Fprintf(stdout, "Starting Pitot dev environment for host %s...\n", host) + + // Cancelable context so we can reap the server goroutine on exit. + devCtx, cancel := context.WithCancel(ctx) + defer cancel() - runtimePath := filepath.Join(os.TempDir(), "pitot-dev-runtime.json") - - manager, err := runtime.Start(ctx, loaded.Config, stderr) + manager, err := runtime.Start(devCtx, loaded.Config, stderr) if err != nil { - return fmt.Errorf("failed to start runtime: %v", err) + return fmt.Errorf("pitot dev: start runtime: %w", err) } defer manager.Close() - srv := runtime.NewServer(manager, loaded.SHA256, runtimePath, stdout, stderr) - - // Start the runtime server in a goroutine - go func() { - if err := srv.Serve(ctx); err != nil { - fmt.Fprintf(stderr, "runtime server error: %v\n", err) + // Render a decision timeline as controllers resolve actions. + manager.SetDecisionObserver(func(d runtime.Decision) { + marker := "ALLOW" + if d.Outcome != "allow" { + marker = strings.ToUpper(d.Outcome) + } + if d.Message != "" { + fmt.Fprintf(stdout, " [%s] %s (%s) — %s\n", marker, d.Kind, d.ActionID, d.Message) + } else { + fmt.Fprintf(stdout, " [%s] %s (%s)\n", marker, d.Kind, d.ActionID) } - }() + }) + + srv := runtime.NewServer(manager, loaded.SHA256, runtimePath, stdout, stderr) + serveErr := make(chan error, 1) + go func() { serveErr <- srv.Serve(devCtx) }() + + // Wait for the runtime to publish a usable descriptor before starting the agent. + if err := waitForRuntime(devCtx, runtimePath, serveErr, 5*time.Second); err != nil { + return err + } + + fmt.Fprintf(stdout, "Runtime ready. Starting agent: %s %s\n", program, strings.Join(programArgs, " ")) + fmt.Fprintln(stdout, "Decisions:") - fmt.Fprintf(stdout, "Runtime listening. Starting agent %s...\n", execCmd) - - cmd := exec.CommandContext(ctx, execCmd) + cmd := exec.CommandContext(devCtx, program, programArgs...) cmd.Env = append(os.Environ(), "PITOT_RUNTIME="+runtimePath) cmd.Stdout = stdout cmd.Stderr = stderr cmd.Stdin = os.Stdin - - if err := cmd.Run(); err != nil { - return fmt.Errorf("agent execution failed: %v", err) + + runErr := cmd.Run() + + // Stop the runtime and reap its goroutine before returning. + cancel() + select { + case <-serveErr: + case <-time.After(5 * time.Second): } + if runErr != nil { + return fmt.Errorf("pitot dev: agent execution failed: %w", runErr) + } + fmt.Fprintln(stdout, "Agent finished. Runtime stopped.") return nil } + +// waitForRuntime blocks until the runtime descriptor is live, the server errors, +// or the deadline elapses. +func waitForRuntime(ctx context.Context, runtimePath string, serveErr <-chan error, timeout time.Duration) error { + deadline := time.After(timeout) + ticker := time.NewTicker(25 * time.Millisecond) + defer ticker.Stop() + for { + select { + case err := <-serveErr: + if err != nil { + return fmt.Errorf("pitot dev: runtime server error: %w", err) + } + return errors.New("pitot dev: runtime server stopped before becoming ready") + case <-ctx.Done(): + return ctx.Err() + case <-deadline: + return errors.New("pitot dev: runtime did not become ready within timeout") + case <-ticker.C: + client, err := runtime.OpenClient(runtimePath) + if err != nil { + continue + } + if err := client.Health(ctx); err == nil { + return nil + } + } + } +} + +func hostList() string { + hosts := adapters.Supported() + names := make([]string, len(hosts)) + for i, h := range hosts { + names[i] = string(h) + } + return strings.Join(names, ", ") +} diff --git a/cmd/pitot/workbench_test.go b/cmd/pitot/workbench_test.go new file mode 100644 index 0000000..6d00a90 --- /dev/null +++ b/cmd/pitot/workbench_test.go @@ -0,0 +1,151 @@ +package main + +import ( + "bytes" + "context" + "os" + "path/filepath" + "strings" + "testing" +) + +// initExpectations maps each language to the files a fresh controller project +// must contain to be runnable without further setup. +var initExpectations = map[string][]string{ + "python": {"main.py", "requirements.txt", "pyproject.toml", ".pitot.yaml"}, + "typescript": {"main.ts", "package.json", "tsconfig.json", ".pitot.yaml"}, + "go": {"main.go", "go.mod", ".pitot.yaml"}, + "rust": {"main.rs", "Cargo.toml", ".pitot.yaml"}, +} + +func TestInitGeneratesRunnableProjectPerLanguage(t *testing.T) { + for lang, wantFiles := range initExpectations { + lang, wantFiles := lang, wantFiles + t.Run(lang, func(t *testing.T) { + dir := filepath.Join(t.TempDir(), "proj") + var stdout, stderr bytes.Buffer + err := runInit([]string{"--language", lang, "--role", "controller", "--dir", dir}, strings.NewReader(""), &stdout, &stderr) + if err != nil { + t.Fatalf("init %s: %v", lang, err) + } + for _, name := range wantFiles { + if _, statErr := os.Stat(filepath.Join(dir, name)); statErr != nil { + t.Errorf("%s: expected generated file %q: %v", lang, name, statErr) + } + } + cfg, readErr := os.ReadFile(filepath.Join(dir, ".pitot.yaml")) + if readErr != nil { + t.Fatalf("read .pitot.yaml: %v", readErr) + } + if !strings.Contains(string(cfg), "controllers:") { + t.Errorf("%s: controller config missing controllers block:\n%s", lang, cfg) + } + }) + } +} + +func TestInitConsumerRoleWritesConsumerConfig(t *testing.T) { + dir := filepath.Join(t.TempDir(), "proj") + var stdout, stderr bytes.Buffer + if err := runInit([]string{"--language", "go", "--role", "consumer", "--dir", dir}, strings.NewReader(""), &stdout, &stderr); err != nil { + t.Fatalf("init: %v", err) + } + cfg, err := os.ReadFile(filepath.Join(dir, ".pitot.yaml")) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(cfg), "consumers:") { + t.Errorf("consumer config missing consumers block:\n%s", cfg) + } + src, err := os.ReadFile(filepath.Join(dir, "main.go")) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(src), "RunConsumer") { + t.Errorf("consumer source is not a consumer:\n%s", src) + } +} + +func TestInitRejectsInvalidRole(t *testing.T) { + dir := filepath.Join(t.TempDir(), "proj") + var stdout, stderr bytes.Buffer + err := runInit([]string{"--language", "python", "--role", "admin", "--dir", dir}, strings.NewReader(""), &stdout, &stderr) + if err == nil { + t.Fatal("expected invalid role to be rejected") + } + if !strings.Contains(err.Error(), "unsupported role") { + t.Errorf("unexpected error: %v", err) + } +} + +func TestInitRejectsInvalidLanguage(t *testing.T) { + dir := filepath.Join(t.TempDir(), "proj") + var stdout, stderr bytes.Buffer + err := runInit([]string{"--language", "cobol", "--dir", dir}, strings.NewReader(""), &stdout, &stderr) + if err == nil || !strings.Contains(err.Error(), "unsupported language") { + t.Fatalf("expected unsupported language error, got %v", err) + } +} + +func TestInitIsNonDestructive(t *testing.T) { + dir := filepath.Join(t.TempDir(), "proj") + var out, errb bytes.Buffer + if err := runInit([]string{"--language", "go", "--role", "controller", "--dir", dir}, strings.NewReader(""), &out, &errb); err != nil { + t.Fatalf("first init: %v", err) + } + // Second init without --force must refuse. + err := runInit([]string{"--language", "go", "--role", "controller", "--dir", dir}, strings.NewReader(""), &out, &errb) + if err == nil || !strings.Contains(err.Error(), "refusing to overwrite") { + t.Fatalf("expected non-destructive refusal, got %v", err) + } + // With --force it must succeed. + if err := runInit([]string{"--language", "go", "--role", "controller", "--dir", dir, "--force"}, strings.NewReader(""), &out, &errb); err != nil { + t.Fatalf("forced init: %v", err) + } +} + +func TestInitRequiresLanguageWhenNonInteractiveAndUndetectable(t *testing.T) { + dir := filepath.Join(t.TempDir(), "empty") + var stdout, stderr bytes.Buffer + err := runInit([]string{"--dir", dir}, strings.NewReader(""), &stdout, &stderr) + if err == nil || !strings.Contains(err.Error(), "--language is required") { + t.Fatalf("expected language-required error, got %v", err) + } +} + +func TestInitDetectsLanguageFromExistingManifest(t *testing.T) { + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "Cargo.toml"), []byte("[package]\nname=\"x\"\n"), 0o644); err != nil { + t.Fatal(err) + } + var stdout, stderr bytes.Buffer + // Non-interactive with no --language: detection must select rust. --force + // because Cargo.toml already exists. + if err := runInit([]string{"--dir", dir, "--role", "controller", "--force"}, strings.NewReader(""), &stdout, &stderr); err != nil { + t.Fatalf("init with detection: %v", err) + } + if !strings.Contains(stdout.String(), "Detected rust") { + t.Errorf("expected rust detection, got:\n%s", stdout.String()) + } + if _, err := os.Stat(filepath.Join(dir, "main.rs")); err != nil { + t.Errorf("expected rust source generated: %v", err) + } +} + +func TestDevRejectsUnsupportedHost(t *testing.T) { + var stdout, stderr bytes.Buffer + err := runDev(context.Background(), []string{"--host", "notahost", "--exec", "true"}, &stdout, &stderr) + if err == nil || !strings.Contains(err.Error(), "unsupported host") { + t.Fatalf("expected unsupported host error, got %v", err) + } +} + +func TestDevRequiresHostAndExec(t *testing.T) { + var stdout, stderr bytes.Buffer + if err := runDev(context.Background(), []string{"--host", "claude"}, &stdout, &stderr); err == nil { + t.Fatal("expected error when --exec missing") + } + if err := runDev(context.Background(), []string{"--exec", "true"}, &stdout, &stderr); err == nil { + t.Fatal("expected error when --host missing") + } +} diff --git a/runtime/runtime.go b/runtime/runtime.go index 48290a2..97aacc2 100644 --- a/runtime/runtime.go +++ b/runtime/runtime.go @@ -22,6 +22,16 @@ import ( const consumerQueueSize = 128 +// Decision is a resolved controller outcome surfaced to an optional observer. +// It carries no policy meaning; it is a transport receipt for tooling such as +// `pitot dev` that renders a decision timeline. +type Decision struct { + Kind string + ActionID string + Outcome string + Message string +} + // Manager starts configured role processes and exposes their shared delivery path. type Manager struct { ctx context.Context @@ -29,6 +39,26 @@ type Manager struct { stderr io.Writer controllers map[string]*controllerWorker consumers []*consumerWorker + observer func(Decision) +} + +// SetDecisionObserver registers a callback invoked for every resolved controller +// decision. It is optional; a nil observer disables receipts. Not safe to change +// concurrently with active delivery. +func (m *Manager) SetDecisionObserver(observer func(Decision)) { + m.observer = observer +} + +func (m *Manager) reportDecision(kind string, response *schema.ControlResponse) { + if m.observer == nil || response == nil { + return + } + m.observer(Decision{ + Kind: kind, + ActionID: response.ActionID, + Outcome: response.Outcome, + Message: response.Message, + }) } // Start creates the complete configured process boundary. A child that cannot @@ -109,6 +139,7 @@ func (m *Manager) DeliverEvent(ctx context.Context, event schema.Event) (*schema ActionID: event.Action.ID, Data: data, }) + m.reportDecision(event.Action.Kind, &response) return &response, resolveErr } @@ -118,7 +149,9 @@ func (m *Manager) Request(ctx context.Context, request schema.ControlRequested) if !exists { return schema.ControlResponse{}, bridge.ErrNoController } - return worker.resolve(ctx, request) + response, err := worker.resolve(ctx, request) + m.reportDecision(request.Kind, &response) + return response, err } type controllerResult struct { diff --git a/sdk/python/pyproject.toml b/sdk/python/pyproject.toml index 291e90a..0cdaab2 100644 --- a/sdk/python/pyproject.toml +++ b/sdk/python/pyproject.toml @@ -10,7 +10,7 @@ authors = [ { name = "Operator Stack" } ] license = { text = "MIT" } -requires-python = ">=3.8" +requires-python = ">=3.10" dependencies = [] [project.urls] diff --git a/sdk/rust/Cargo.toml b/sdk/rust/Cargo.toml new file mode 100644 index 0000000..867d8f7 --- /dev/null +++ b/sdk/rust/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "pitot" +version = "0.1.0" +edition = "2021" +description = "Pitot: The passive, protocol-first measurement boundary for coding agents." +license = "MIT" +repository = "https://github.com/operatorstack/pitot" + +[lib] +path = "src/lib.rs" + +[dependencies] +serde = { version = "1", features = ["derive"] } +serde_json = "1" diff --git a/sdk/rust/src/runner.rs b/sdk/rust/src/runner.rs index 6b64a7d..de284d8 100644 --- a/sdk/rust/src/runner.rs +++ b/sdk/rust/src/runner.rs @@ -1,4 +1,4 @@ -use std::io::{self, BufRead}; +use std::io::{self, BufRead, Write}; use serde_json; use crate::types::{Event, ControlRequested, ControlResponse}; @@ -11,17 +11,23 @@ where { let stdin = io::stdin(); for line in stdin.lock().lines() { - if let Ok(line_str) = line { - if line_str.trim().is_empty() { + let line_str = match line { + Ok(value) => value, + Err(err) => { + // Surface stdin read faults instead of silently dropping them. + eprintln!("Pitot Consumer error: {}", err); continue; } - match serde_json::from_str::(&line_str) { - Ok(event) => { - handler(event); - } - Err(err) => { - eprintln!("Pitot Consumer error: {}", err); - } + }; + if line_str.trim().is_empty() { + continue; + } + match serde_json::from_str::(&line_str) { + Ok(event) => { + handler(event); + } + Err(err) => { + eprintln!("Pitot Consumer error: {}", err); } } } @@ -53,29 +59,46 @@ where F: Fn(ControlRequested) -> Outcome + Send + Sync + 'static, { let stdin = io::stdin(); + let stdout = io::stdout(); for line in stdin.lock().lines() { - if let Ok(line_str) = line { - if line_str.trim().is_empty() { + let line_str = match line { + Ok(value) => value, + Err(err) => { + eprintln!("Pitot Controller error: {}", err); continue; } - match serde_json::from_str::(&line_str) { - Ok(req) => { - let result = handler(req.clone()); - let response = ControlResponse { - pitot_version: "1".to_string(), - control_response_type: "control.response".to_string(), - controller_id: controller_id.to_string(), - action_id: req.action_id, - outcome: result.outcome, - message: result.message, - }; - if let Ok(json) = serde_json::to_string(&response) { - println!("{}", json); + }; + if line_str.trim().is_empty() { + continue; + } + match serde_json::from_str::(&line_str) { + Ok(req) => { + let result = handler(req.clone()); + let response = ControlResponse { + pitot_version: "1".to_string(), + control_response_type: "control.response".to_string(), + controller_id: controller_id.to_string(), + action_id: req.action_id, + outcome: result.outcome, + message: result.message, + }; + match serde_json::to_string(&response) { + Ok(json) => { + let mut handle = stdout.lock(); + // stdout is block-buffered when piped (the normal case + // behind a runtime), so flush every response to keep the + // controller responsive. + if writeln!(handle, "{}", json).is_err() || handle.flush().is_err() { + eprintln!("Pitot Controller error: failed to write response"); + } + } + Err(err) => { + eprintln!("Pitot Controller error: {}", err); } } - Err(err) => { - eprintln!("Pitot Controller error: {}", err); - } + } + Err(err) => { + eprintln!("Pitot Controller error: {}", err); } } } diff --git a/sdk/tests/go_test.go b/sdk/tests/go_test.go deleted file mode 100644 index 2c378a6..0000000 --- a/sdk/tests/go_test.go +++ /dev/null @@ -1,13 +0,0 @@ - -package main - -import ( - "github.com/operatorstack/pitot/sdk" - "github.com/operatorstack/pitot/schema" -) - -func main() { - sdk.RunController("test-controller", func(req schema.ControlRequested) sdk.Outcome { - return sdk.Allow("test approved") - }) -} diff --git a/sdk/tests/go_test_runner.go b/sdk/tests/go_test_runner.go deleted file mode 100644 index 2c378a6..0000000 --- a/sdk/tests/go_test_runner.go +++ /dev/null @@ -1,13 +0,0 @@ - -package main - -import ( - "github.com/operatorstack/pitot/sdk" - "github.com/operatorstack/pitot/schema" -) - -func main() { - sdk.RunController("test-controller", func(req schema.ControlRequested) sdk.Outcome { - return sdk.Allow("test approved") - }) -} diff --git a/sdk/tests/python_test.py b/sdk/tests/python_test.py deleted file mode 100644 index 9933096..0000000 --- a/sdk/tests/python_test.py +++ /dev/null @@ -1,10 +0,0 @@ - -import sys -from pitot.runner import run_controller, allow -from pitot.types import ControlRequested - -def handler(req: ControlRequested): - return allow("test approved") - -if __name__ == "__main__": - run_controller("test-controller", handler) diff --git a/sdk/tests/test_runners.py b/sdk/tests/test_runners.py index 0ee6398..eb82faf 100644 --- a/sdk/tests/test_runners.py +++ b/sdk/tests/test_runners.py @@ -1,95 +1,341 @@ +"""Shared Consumer/Controller conformance suite for the Pitot SDK runners. + +Every first-class language (Python, TypeScript, Go, Rust) is driven through the +same fixture cases so their observable protocol behavior stays identical: + + * controller stream - N requests produce N responses, in request order + * malformed input - a bad line is reported and skipped, stream continues + * deny outcome - the handler's deny is faithfully serialized + * consumer stream - events are delivered and stdout stays protocol-clean + +Paths are resolved relative to this file (no hardcoded monorepo depth), and each +language is skipped cleanly when its toolchain is unavailable. +""" + import json -import subprocess import os -import sys +import shutil +import subprocess +import tempfile import unittest from pathlib import Path -ROOT = Path(__file__).resolve().parents[5] -SDK_DIR = ROOT / "labs" / "15-pitot" / "pitot-distribution" / "sdk" - -class TestLanguageRunners(unittest.TestCase): - def run_controller_test(self, command: list[str], cwd: Path, extra_env: dict = None): - request = { - "pitot_version": "1", - "type": "control.requested", - "kind": "test.approval", - "action_id": "act_test123", - "data": {"test_field": "value"} - } - - env = os.environ.copy() - if extra_env: - env.update(extra_env) - - proc = subprocess.Popen( - command, - cwd=cwd, - env=env, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True - ) - - stdout, stderr = proc.communicate(json.dumps(request) + "\n") - self.assertEqual(proc.returncode, 0, f"Process failed: {stderr}") - - output = [line for line in stdout.strip().split("\n") if line] - self.assertEqual(len(output), 1) - - response = json.loads(output[0]) - self.assertEqual(response["pitot_version"], "1") - self.assertEqual(response["type"], "control.response") - self.assertEqual(response["controller_id"], "test-controller") - self.assertEqual(response["action_id"], "act_test123") - self.assertEqual(response["outcome"], "allow") - self.assertEqual(response["message"], "test approved") - - def test_python_runner(self): - script_path = SDK_DIR / "tests" / "python_test.py" - script_path.write_text(""" -import sys -from pitot.runner import run_controller, allow +SDK_DIR = Path(__file__).resolve().parent.parent +FIXTURES_DIR = SDK_DIR / "tests" / "_fixtures" + + +def go_module_dir() -> Path | None: + """Locate the Go module root in either the monorepo or flattened layout.""" + for candidate in (SDK_DIR.parent, SDK_DIR.parent.parent / "pitot"): + if (candidate / "go.mod").exists(): + return candidate + return None + + +def control_request(action_id: str) -> str: + return json.dumps({ + "pitot_version": "1", + "type": "control.requested", + "kind": "test.approval", + "action_id": action_id, + "data": {"test_field": "value"}, + }) + + +def action_event(action_id: str) -> str: + return json.dumps({ + "pitot_version": "1", + "type": "action.requested", + "host": {"name": "test-host"}, + "observation": {"fidelity": "full", "source": "test"}, + "action": {"id": action_id, "kind": "shell"}, + }) + + +def run_lines(command, cwd, env, lines): + proc = subprocess.Popen( + command, + cwd=str(cwd), + env=env, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + stdout, stderr = proc.communicate("\n".join(lines) + "\n", timeout=120) + return proc.returncode, stdout, stderr + + +# Fixture sources. Handlers key their outcome off PITOT_TEST_OUTCOME so a single +# fixture serves both allow and deny cases. + +PY_CONTROLLER = """import os +from pitot.runner import run_controller, allow, deny from pitot.types import ControlRequested def handler(req: ControlRequested): + if os.environ.get("PITOT_TEST_OUTCOME") == "deny": + return deny("test denied") return allow("test approved") if __name__ == "__main__": run_controller("test-controller", handler) -""") - self.run_controller_test(["python3", str(script_path)], SDK_DIR / "python", {"PYTHONPATH": str(SDK_DIR / "python")}) +""" + +PY_CONSUMER = """import sys +from pitot.runner import run_consumer +from pitot.types import Event - def test_ts_runner(self): - script_path = SDK_DIR / "tests" / "ts_test.ts" - script_path.write_text(""" -import { runController, allow } from '../typescript/src/runner'; -import { ControlRequested } from '../typescript/src/pitot'; +def handler(event: Event): + print("consumed", event.type, file=sys.stderr) + +if __name__ == "__main__": + run_consumer(handler) +""" + +TS_CONTROLLER = """import { runController, allow, deny } from '../../typescript/src/runner'; +import { ControlRequested } from '../../typescript/src/pitot'; runController("test-controller", async (req: ControlRequested) => { + if (process.env.PITOT_TEST_OUTCOME === "deny") return deny("test denied"); return allow("test approved"); }); -""") - self.run_controller_test(["npx", "tsx", str(script_path)], SDK_DIR / "typescript") +""" + +TS_CONSUMER = """import { runConsumer } from '../../typescript/src/runner'; +import { Event } from '../../typescript/src/pitot'; + +runConsumer(async (event: Event) => { + console.error("consumed", event.type); +}); +""" - def test_go_runner(self): - script_path = SDK_DIR / "tests" / "go_test_runner.go" - script_path.write_text(""" -package main +GO_CONTROLLER = """package main import ( - "github.com/operatorstack/pitot/sdk" - "github.com/operatorstack/pitot/schema" +\t"os" + +\t"github.com/operatorstack/pitot/schema" +\t"github.com/operatorstack/pitot/sdk" ) func main() { - sdk.RunController("test-controller", func(req schema.ControlRequested) sdk.Outcome { - return sdk.Allow("test approved") - }) +\tsdk.RunController("test-controller", func(req schema.ControlRequested) sdk.Outcome { +\t\tif os.Getenv("PITOT_TEST_OUTCOME") == "deny" { +\t\t\treturn sdk.Deny("test denied") +\t\t} +\t\treturn sdk.Allow("test approved") +\t}) +} +""" + +GO_CONSUMER = """package main + +import ( +\t"fmt" +\t"os" + +\t"github.com/operatorstack/pitot/schema" +\t"github.com/operatorstack/pitot/sdk" +) + +func main() { +\tsdk.RunConsumer(func(event schema.Event) { +\t\tfmt.Fprintf(os.Stderr, "consumed %s\\n", event.Type) +\t}) +} +""" + +RUST_CONTROLLER = """use pitot::{run_controller, allow, deny, ControlRequested, Outcome}; + +fn handler(_req: ControlRequested) -> Outcome { + if std::env::var("PITOT_TEST_OUTCOME").as_deref() == Ok("deny") { + return deny(Some("test denied".to_string())); + } + allow(Some("test approved".to_string())) +} + +fn main() { + run_controller("test-controller", Box::new(handler)); } -""") - self.run_controller_test(["go", "run", str(script_path)], ROOT / "labs" / "15-pitot" / "pitot") +""" + +RUST_CONSUMER = """use pitot::{run_consumer, Event}; + +fn handler(event: Event) { + eprintln!("consumed {}", event.type_field); +} + +fn main() { + run_consumer(Box::new(handler)); +} +""" + + +class RunnerContract: + """Mixin defining the shared conformance cases. + + Subclasses provide controller_command()/consumer_command() returning + (command, cwd, env) and available() returning (bool, reason). + """ + + def setUp(self): + ok, reason = self.available() + if not ok: + self.skipTest(reason) + FIXTURES_DIR.mkdir(parents=True, exist_ok=True) + + @classmethod + def tearDownClass(cls): + # Generated fixtures must never linger under the published sdk/ tree, or + # they would show up as projection drift in UPSTREAM.json. + shutil.rmtree(FIXTURES_DIR, ignore_errors=True) + + # --- Controller cases ------------------------------------------------- + + def test_controller_stream_preserves_order(self): + cmd, cwd, env = self.controller_command() + ids = ["act_1", "act_2", "act_3"] + code, stdout, stderr = run_lines(cmd, cwd, env, [control_request(i) for i in ids]) + self.assertEqual(code, 0, f"{self.language}: process failed: {stderr}") + lines = [l for l in stdout.strip().split("\n") if l] + self.assertEqual(len(lines), len(ids), f"{self.language}: expected one response per request:\n{stdout}\n{stderr}") + for expected, raw in zip(ids, lines): + resp = json.loads(raw) + self.assertEqual(resp["pitot_version"], "1") + self.assertEqual(resp["type"], "control.response") + self.assertEqual(resp["controller_id"], "test-controller") + self.assertEqual(resp["action_id"], expected, f"{self.language}: responses out of order") + self.assertEqual(resp["outcome"], "allow") + self.assertEqual(resp["message"], "test approved") + + def test_controller_skips_malformed_line(self): + cmd, cwd, env = self.controller_command() + lines = [control_request("act_1"), "{ this is not valid json", control_request("act_2")] + code, stdout, stderr = run_lines(cmd, cwd, env, lines) + self.assertEqual(code, 0, f"{self.language}: malformed input must not crash the runner: {stderr}") + responses = [json.loads(l) for l in stdout.strip().split("\n") if l] + self.assertEqual([r["action_id"] for r in responses], ["act_1", "act_2"], + f"{self.language}: malformed line should be skipped, not emit a response:\n{stdout}") + + def test_controller_deny(self): + cmd, cwd, env = self.controller_command() + env = dict(env) + env["PITOT_TEST_OUTCOME"] = "deny" + code, stdout, stderr = run_lines(cmd, cwd, env, [control_request("act_1")]) + self.assertEqual(code, 0, f"{self.language}: {stderr}") + responses = [json.loads(l) for l in stdout.strip().split("\n") if l] + self.assertEqual(len(responses), 1) + self.assertEqual(responses[0]["outcome"], "deny") + self.assertEqual(responses[0]["message"], "test denied") + + # --- Consumer cases --------------------------------------------------- + + def test_consumer_stream_keeps_stdout_clean(self): + cmd, cwd, env = self.consumer_command() + code, stdout, stderr = run_lines(cmd, cwd, env, [action_event("act_1"), action_event("act_2")]) + self.assertEqual(code, 0, f"{self.language}: {stderr}") + self.assertEqual(stdout.strip(), "", f"{self.language}: consumer must not write to stdout:\n{stdout}") + self.assertEqual(stderr.count("consumed"), 2, f"{self.language}: expected two delivered events:\n{stderr}") + + +def write_fixture(name: str, contents: str) -> Path: + FIXTURES_DIR.mkdir(parents=True, exist_ok=True) + path = FIXTURES_DIR / name + path.write_text(contents) + return path + + +class TestPythonRunner(RunnerContract, unittest.TestCase): + language = "python" + + def available(self): + return (shutil.which("python3") is not None, "python3 not available") + + def _env(self): + env = os.environ.copy() + env["PYTHONPATH"] = str(SDK_DIR / "python") + return env + + def controller_command(self): + path = write_fixture("py_controller.py", PY_CONTROLLER) + return ["python3", str(path)], FIXTURES_DIR, self._env() + + def consumer_command(self): + path = write_fixture("py_consumer.py", PY_CONSUMER) + return ["python3", str(path)], FIXTURES_DIR, self._env() + + +class TestTypeScriptRunner(RunnerContract, unittest.TestCase): + language = "typescript" + + def available(self): + return (shutil.which("npx") is not None, "npx not available") + + def controller_command(self): + path = write_fixture("ts_controller.ts", TS_CONTROLLER) + return ["npx", "tsx", str(path)], SDK_DIR / "typescript", os.environ.copy() + + def consumer_command(self): + path = write_fixture("ts_consumer.ts", TS_CONSUMER) + return ["npx", "tsx", str(path)], SDK_DIR / "typescript", os.environ.copy() + + +class TestGoRunner(RunnerContract, unittest.TestCase): + language = "go" + + def available(self): + if shutil.which("go") is None: + return (False, "go not available") + if go_module_dir() is None: + return (False, "go module not found in expected layout") + return (True, "") + + def controller_command(self): + path = write_fixture("go_controller.go", GO_CONTROLLER) + return ["go", "run", str(path)], go_module_dir(), os.environ.copy() + + def consumer_command(self): + path = write_fixture("go_consumer.go", GO_CONSUMER) + return ["go", "run", str(path)], go_module_dir(), os.environ.copy() + + +class TestRustRunner(RunnerContract, unittest.TestCase): + language = "rust" + + def available(self): + if shutil.which("cargo") is None: + return (False, "cargo not available") + if not (SDK_DIR / "rust" / "Cargo.toml").exists(): + return (False, "rust SDK crate missing Cargo.toml") + return (True, "") + + def _build(self, fixture_name: str, contents: str) -> Path: + # Build a throwaway crate that path-depends on the Rust SDK, then run + # its compiled binary. Cached across cases within the process. + project = Path(tempfile.mkdtemp(prefix="pitot-rust-conf-")) + self.addCleanup(shutil.rmtree, project, ignore_errors=True) + crate = (SDK_DIR / "rust").resolve() + (project / "src").mkdir(parents=True) + (project / "src" / "main.rs").write_text(contents) + (project / "Cargo.toml").write_text( + "[package]\nname = \"conf\"\nversion = \"0.0.0\"\nedition = \"2021\"\n\n" + f"[dependencies]\npitot = {{ path = \"{crate}\" }}\n" + ) + build = subprocess.run(["cargo", "build", "--quiet"], cwd=str(project), + capture_output=True, text=True, timeout=300) + if build.returncode != 0: + self.fail(f"rust build failed: {build.stderr}") + return project / "target" / "debug" / "conf" + + def controller_command(self): + binary = self._build("controller", RUST_CONTROLLER) + return [str(binary)], SDK_DIR / "rust", os.environ.copy() + + def consumer_command(self): + binary = self._build("consumer", RUST_CONSUMER) + return [str(binary)], SDK_DIR / "rust", os.environ.copy() + if __name__ == "__main__": unittest.main() diff --git a/sdk/tests/ts_test.ts b/sdk/tests/ts_test.ts deleted file mode 100644 index 05e59f6..0000000 --- a/sdk/tests/ts_test.ts +++ /dev/null @@ -1,7 +0,0 @@ - -import { runController, allow } from '../typescript/src/runner'; -import { ControlRequested } from '../typescript/src/pitot'; - -runController("test-controller", async (req: ControlRequested) => { - return allow("test approved"); -}); diff --git a/sdk/typescript/package.json b/sdk/typescript/package.json index cab900e..c8455e1 100644 --- a/sdk/typescript/package.json +++ b/sdk/typescript/package.json @@ -2,15 +2,15 @@ "name": "@operatorstack/pitot", "version": "0.1.0", "description": "Pitot: The passive, protocol-first measurement boundary for coding agents.", - "main": "src/index.js", - "types": "src/index.d.ts", + "main": "dist/index.js", + "types": "dist/index.d.ts", "files": [ - "src/**/*.ts", - "src/**/*.js", - "src/**/*.d.ts" + "dist/**/*.js", + "dist/**/*.d.ts" ], "scripts": { - "build": "tsc" + "build": "tsc", + "prepublishOnly": "npm run build" }, "repository": { "type": "git", diff --git a/sdk/typescript/src/runner.ts b/sdk/typescript/src/runner.ts index 626ac46..cfeb9dc 100644 --- a/sdk/typescript/src/runner.ts +++ b/sdk/typescript/src/runner.ts @@ -3,19 +3,33 @@ import { Event, ControlRequested, ControlResponse } from './pitot'; export type ConsumerHandler = (event: Event) => void | Promise; -export function runConsumer(handler: ConsumerHandler): void { +// serializeLines drives an async per-line handler strictly in arrival order. +// readline does not await listener promises, so without this a slow handler +// could let a later line's response overtake an earlier one. We chain each +// line onto a single promise so handlers run — and responses are emitted — +// in the exact order the lines arrived. +function serializeLines(onLine: (line: string) => Promise): void { const rl = readline.createInterface({ input: process.stdin, output: process.stdout, - terminal: false + terminal: false, + }); + + let chain: Promise = Promise.resolve(); + rl.on('line', (line) => { + chain = chain.then(() => onLine(line)); }); +} - rl.on('line', async (line) => { +export function runConsumer(handler: ConsumerHandler): void { + serializeLines(async (line) => { if (!line.trim()) return; try { const event = JSON.parse(line) as Event; await handler(event); } catch (err) { + // Malformed input and handler faults are reported and skipped; the + // stream continues so one bad line never tears down the runner. console.error("Pitot Consumer error:", err); } }); @@ -34,18 +48,12 @@ export function deny(message?: string): Outcome { export type ControllerHandler = (req: ControlRequested) => Outcome | Promise; export function runController(controllerId: string, handler: ControllerHandler): void { - const rl = readline.createInterface({ - input: process.stdin, - output: process.stdout, - terminal: false - }); - - rl.on('line', async (line) => { + serializeLines(async (line) => { if (!line.trim()) return; try { const req = JSON.parse(line) as ControlRequested; const result = await handler(req); - + const response: ControlResponse = { pitot_version: "1", type: "control.response", @@ -56,9 +64,12 @@ export function runController(controllerId: string, handler: ControllerHandler): if (result.message !== undefined) { response.message = result.message; } - + + // Exactly one response per successfully parsed request, in order. console.log(JSON.stringify(response)); } catch (err) { + // Malformed input and handler faults are reported and skipped; no + // response line is emitted for an unparseable request. console.error("Pitot Controller error:", err); } }); diff --git a/sdk/typescript/tsconfig.json b/sdk/typescript/tsconfig.json index 2e54945..d6c779d 100644 --- a/sdk/typescript/tsconfig.json +++ b/sdk/typescript/tsconfig.json @@ -3,7 +3,8 @@ "target": "es2022", "module": "commonjs", "declaration": true, - "outDir": ".", + "rootDir": "src", + "outDir": "dist", "strict": true, "esModuleInterop": true, "skipLibCheck": true,