diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 4d633b0..bbd9787 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -2,9 +2,8 @@ name: Test on: push: - branches: [master, main] + branches: [master, main, "agent/**"] pull_request: - branches: [master, main] jobs: test: @@ -26,7 +25,7 @@ jobs: run: go test ./... - name: Build - run: go build ./cmd/scud/ + run: go build ./cmd/scud/ ./cmd/scudv2/ - name: Vet run: go vet ./... diff --git a/README.md b/README.md index b4752c3..caefdb5 100644 --- a/README.md +++ b/README.md @@ -96,6 +96,15 @@ callers continue to use the legacy Rho CLI adapter by default; embedders can opt into the versioned `rho.run/v1` JSONL adapter. See [`pkg/executor`](pkg/executor/README.md) for the protocol integration boundary. +The opt-in v2 core is a smaller embeddable harness: deterministic reducer and +reconciler, DAG bridge, bounded runtime, and replaceable policy, execution, and +event-store adapters. `v2/adapters/tdhttp` persists its canonical event stream +in td under the ticket's live attempt/execution fence and obtains +Shen-authorized BIP340 execution witnesses. `v2/adapters/rho` executes the same +provider-neutral request through Rho, which supports Anthropic, OpenAI, and +xAI. SCUD owns goal/DAG orchestration, td owns identity, leases, policy and +durability, and Rho owns bounded provider/tool execution. + To try the protocol adapter without changing project configuration: ```sh diff --git a/cmd/scudv2/main.go b/cmd/scudv2/main.go new file mode 100644 index 0000000..e75b76a --- /dev/null +++ b/cmd/scudv2/main.go @@ -0,0 +1,140 @@ +// scudv2 is an opt-in, provider-blind view of the v2 core. It intentionally +// only reads a JSON goal/event document; execution and persistence remain +// adapter responsibilities. +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "os" + "strings" + + "github.com/reuben/scud/v2/core" +) + +type document struct { + Goal *core.Goal `json:"goal,omitempty"` + Budget core.Budget `json:"budget,omitempty"` + Events []core.Event `json:"events"` +} + +func main() { + if err := run(context.Background(), os.Args[1:], os.Stdout, os.Stderr); err != nil { + fmt.Fprintln(os.Stderr, "scudv2:", err) + os.Exit(1) + } +} + +func run(ctx context.Context, args []string, stdout, stderr io.Writer) error { + return runWithStdin(ctx, args, os.Stdin, stdout, stderr) +} + +func runWithStdin(ctx context.Context, args []string, stdin io.Reader, stdout, stderr io.Writer) error { + if len(args) == 0 { + return errors.New("usage: scudv2 ") + } + command := args[0] + if command != "validate" && command != "plan" && command != "replay" { + return fmt.Errorf("unknown command %q (want validate, plan, or replay)", command) + } + fs := flag.NewFlagSet(command, flag.ContinueOnError) + fs.SetOutput(stderr) + if err := fs.Parse(args[1:]); err != nil { + return err + } + if fs.NArg() != 1 { + return fmt.Errorf("usage: scudv2 %s ", command) + } + if err := ctx.Err(); err != nil { + return err + } + doc, err := readDocument(fs.Arg(0), stdin) + if err != nil { + return err + } + state, err := replay(doc) + if err != nil { + return err + } + if err := state.Validate(); err != nil { + return fmt.Errorf("validate document: %w", err) + } + + var output any + switch command { + case "validate": + output = map[string]any{"valid": true, "revision": state.Revision} + case "replay": + output = state + case "plan": + decision, err := core.Reconcile(state) + if err != nil { + return fmt.Errorf("plan: %w", err) + } + output = decision + } + encoded, err := json.MarshalIndent(output, "", " ") + if err != nil { + return err + } + _, err = fmt.Fprintf(stdout, "%s\n", encoded) + return err +} + +func readDocument(path string, stdin io.Reader) (document, error) { + var data []byte + var err error + if path == "-" { + data, err = io.ReadAll(stdin) + } else { + data, err = os.ReadFile(path) + } + if err != nil { + return document{}, fmt.Errorf("read document: %w", err) + } + trimmed := strings.TrimSpace(string(data)) + if trimmed == "" { + return document{}, errors.New("document is empty") + } + if strings.HasPrefix(trimmed, "[") { + var events []core.Event + if err := json.Unmarshal(data, &events); err != nil { + return document{}, fmt.Errorf("decode events: %w", err) + } + return document{Events: events}, nil + } + var doc document + if err := json.Unmarshal(data, &doc); err != nil { + return document{}, fmt.Errorf("decode document: %w", err) + } + return doc, nil +} + +func replay(doc document) (core.State, error) { + events := append([]core.Event(nil), doc.Events...) + if doc.Goal != nil { + if len(events) == 0 || events[0].Kind != core.EventGoalCreated { + events = append([]core.Event{{ID: core.ID("goal-" + string(doc.Goal.ID)), Kind: core.EventGoalCreated, Goal: *doc.Goal}}, events...) + } + } + if doc.Budget.MaxSteps == 0 && doc.Budget.MaxCost == 0 { + return core.Replay(events) + } + state := core.NewStateWithBudget(doc.Budget) + for _, event := range events { + var err error + if event.Sequence == 0 { + state, err = state.Append(event) + } else { + state, err = core.Reduce(state, event) + } + if err != nil { + return core.State{}, err + } + } + return state, nil +} diff --git a/cmd/scudv2/main_test.go b/cmd/scudv2/main_test.go new file mode 100644 index 0000000..74a1f10 --- /dev/null +++ b/cmd/scudv2/main_test.go @@ -0,0 +1,77 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" +) + +func fixture(t *testing.T) string { + t.Helper() + doc := `{"events":[{"id":"goal-event","kind":"goal_created","goal":{"id":"g","title":"demo"}},{"id":"obligation-event","kind":"obligation_added","obligation":{"id":"a","goal_id":"g","description":"first"}}]}` + path := filepath.Join(t.TempDir(), "goal.json") + if err := os.WriteFile(path, []byte(doc), 0o600); err != nil { + t.Fatal(err) + } + return path +} + +func TestValidateAndPlan(t *testing.T) { + path := fixture(t) + var out, errOut bytes.Buffer + if err := run(context.Background(), []string{"validate", path}, &out, &errOut); err != nil { + t.Fatalf("validate: %v (%s)", err, errOut.String()) + } + var summary map[string]any + if err := json.Unmarshal(out.Bytes(), &summary); err != nil || summary["valid"] != true { + t.Fatalf("unexpected validate output: %s", out.String()) + } + out.Reset() + if err := run(context.Background(), []string{"plan", path}, &out, &errOut); err != nil { + t.Fatalf("plan: %v", err) + } + if !strings.Contains(out.String(), `"kind": "execute"`) || !strings.Contains(out.String(), `"obligation_id": "a"`) { + t.Fatalf("unexpected plan output: %s", out.String()) + } +} + +func TestReplayFromStdinAndInvalidDocument(t *testing.T) { + path := fixture(t) + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + var out, errOut bytes.Buffer + if err := runWithStdin(context.Background(), []string{"replay", "-"}, bytes.NewReader(data), &out, &errOut); err != nil { + t.Fatal(err) + } + if !strings.Contains(out.String(), `"revision": 2`) { + t.Fatalf("unexpected replay output: %s", out.String()) + } + bad := filepath.Join(t.TempDir(), "bad.json") + if err := os.WriteFile(bad, []byte(`{"events":[{"kind":"obligation_added","obligation":{"id":"x","goal_id":"missing"}}]}`), 0o600); err != nil { + t.Fatal(err) + } + if err := run(context.Background(), []string{"validate", bad}, &out, &errOut); err == nil { + t.Fatal("expected invalid document error") + } +} + +func TestGoalShorthandPrependsGoalEvent(t *testing.T) { + doc := `{"goal":{"id":"g","title":"shorthand"},"events":[{"kind":"obligation_added","obligation":{"id":"a","goal_id":"g"}}]}` + path := filepath.Join(t.TempDir(), "goal-shorthand.json") + if err := os.WriteFile(path, []byte(doc), 0o600); err != nil { + t.Fatal(err) + } + var out, errOut bytes.Buffer + if err := run(context.Background(), []string{"validate", path}, &out, &errOut); err != nil { + t.Fatalf("validate shorthand: %v", err) + } + if !strings.Contains(out.String(), `"revision": 2`) { + t.Fatalf("expected synthesized goal event: %s", out.String()) + } +} diff --git a/docs/scud-v2-architecture.md b/docs/scud-v2-architecture.md new file mode 100644 index 0000000..2db903b --- /dev/null +++ b/docs/scud-v2-architecture.md @@ -0,0 +1,127 @@ +# SCUD v2 migration boundary + +This document records the first migration seam for v2. It is deliberately +additive: the current CLI and graph implementation remain the default until a +consumer opts into the v2 packages. + +## Shape + +The v2 core owns task orchestration, graph decisions, and lifecycle state. It +does not import a model provider, `rho-cli`, Shen, SQLite, SCG, or the `td` +command. Those concerns are represented by interfaces in +`v2/adapters`: + +| concern | v2 seam | concrete integration | +| --- | --- | --- | +| bounded agent run | `adapters.Runner` | `v2/adapters/rho` (`rho.run/v1`) | +| authorization | `adapters.Policy` | `v2/adapters/shen` | +| append-only progress | `adapters.EventStore` | application-owned store; `adapters/memory` for tests | +| task directory (td) | `adapters.TaskDirectory` | SCG/SQLite/remote adapter; `adapters/memory` for tests | + +`RunRequest.Model` is an opaque string. A provider-aware adapter may interpret +`provider/model`; the scheduler must only pass it through. Event `Data` is +opaque bytes for the same reason. This keeps provider credentials and policy +syntax out of graph packages and makes contract tests deterministic. + +## DAG projection and ownership + +The v2 event stream is the source of truth for goal and obligation lifecycle. +`v2/core.State` is a replayable projection of that stream; it is not a second +database and must never be updated by mutating fields in place. `v2/graph` +derives a deterministic DAG view and next decision from a projected state. It +does not own persistence, dispatch, credentials, or policy decisions. + +Ownership is intentionally one-way: + +1. An adapter (or CLI import) appends a domain event. +2. `core.Reduce` validates and applies it, producing a new state. +3. `graph`/`core.Reconcile` reads that state and proposes the next decision. +4. An executor/policy adapter performs or rejects the proposal and emits the + resulting observation event. + +Legacy SCG files and `td` databases are input/output projections during the +migration, never competing sources of truth. Their adapters must preserve +event order and IDs when importing, and must not silently write back status +changes outside the event append path. This lets v1 and v2 run side by side +while making ownership explicit. + +## rho.run/v1 adapter + +`v2/adapters/rho` is a thin translation layer over the existing +`pkg/executor` protocol implementation. It validates the adapter's model +syntax, translates requests and limits, and copies events before delivering +them to a v2 sink. The adapter is optional; callers can provide any +`adapters.Runner` implementation. Existing `pkg/executor.LegacyRho` remains +available for compatibility. + +The JSONL fixtures under `v2/adapters/rho/testdata` are conformance examples: +`completed.jsonl` covers deltas and a terminal result, while +`invalid_after_terminal.jsonl` ensures consumers reject post-terminal events. +They can be replayed through `rho.Consume` without spawning a process. + +## Shen policy adapter + +`v2/adapters/shen` depends only on a local `Evaluator` interface, not a Shen SDK +version. A future SDK integration implements that interface and can be swapped +without changing v2 core. Policy failures are returned as errors; a successful +decision is explicit (`Allowed` plus a reason and optional constraints). + +## Event and task stores + +`EventStore.Append` is append-only and requires strictly increasing sequence +numbers per run. `List` accepts a cursor (`after`) so consumers can resume +without replaying the whole stream. `TaskDirectory` exposes only the task +projection needed by scheduling; richer legacy fields stay inside its adapter. +The in-memory implementations are test fixtures, not durable storage. + +## Legacy package inventory + +The following packages should be retained during the migration: + +- `pkg/model`, `pkg/scg`, `pkg/wave`: stable graph/model APIs and SCG + compatibility. New v2 code should consume a `TaskDirectory` projection + rather than import these packages directly. +- `pkg/executor`: keep `Runner`, `LegacyRho`, and the rho.run/v1 protocol + validator while downstream callers move to `v2/adapters`. +- `internal/db`, `internal/storage`, `internal/config`: retain for the v1 CLI + and implement v2 adapters on top only after persistence semantics are tested. + +The following should move behind adapters, then be deprecated once no v1 +entrypoint imports them: + +- `internal/rho`: legacy process invocation; use `v2/adapters/rho` for new + execution and leave this package as a compatibility shim. +- direct `internal/db` event/session calls: expose them through an + `EventStore` adapter so core code cannot depend on SQLite details. +- direct task-file reads/writes in command packages: expose them through a + `TaskDirectory` adapter. + +No package is deleted by this scaffold. A later release can add deprecation +markers and an integrated `scud v2` alias once the graph implementation and +adapter backends are ready; `scudv2` remains the low-risk standalone entrypoint +during migration. + +## Migration sequence + +1. Add adapter-backed entrypoints and replay the fixtures in CI. +2. Implement durable EventStore and TaskDirectory adapters, preserving v1 + locking and event ordering behavior. +3. Run v2 behind the explicit `scudv2` command/config switch; compare + projections and decisions with the v1 scheduler. +4. Migrate callers package-by-package, then deprecate direct legacy imports. + +## Opt-in CLI + +`cmd/scudv2` is intentionally standalone so the existing `scud` command and +its legacy state files are unaffected. It accepts either a JSON object with an +`events` array (and optional `goal`/`budget`) or a bare JSON event array: + +```sh +scudv2 validate goal.json +scudv2 plan goal.json +scudv2 replay goal.json +``` + +`validate` replays and checks invariants, `plan` prints the deterministic next +decision, and `replay` prints the projected state. The command performs no +agent invocation, provider lookup, policy evaluation, or persistence write. diff --git a/pkg/executor/rho_v1.go b/pkg/executor/rho_v1.go index 9988b9a..c21ab0b 100644 --- a/pkg/executor/rho_v1.go +++ b/pkg/executor/rho_v1.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "io" + "os" "os/exec" "strings" ) @@ -17,6 +18,9 @@ type RhoV1 struct { Command string Args []string Grant Grant + // Authorize receives the exact canonical unsigned request bytes and returns + // a portable witness plus the trusted BIP340 issuer public key. + Authorize func(context.Context, []byte) (witness, issuerPubkey string, err error) } type Grant struct { @@ -82,6 +86,22 @@ func (r RhoV1) Run(ctx context.Context, req Request, handler EventHandler) (*Res if err != nil { return nil, fmt.Errorf("marshal rho.run/v1 request: %w", err) } + issuerPubkey := "" + if r.Authorize != nil { + unsigned, canonicalErr := canonicalJSON(payload) + if canonicalErr != nil { + return nil, fmt.Errorf("canonicalize rho.run/v1 request: %w", canonicalErr) + } + witness, pubkey, authErr := r.Authorize(ctx, unsigned) + if authErr != nil { + return nil, fmt.Errorf("authorize rho.run/v1 request: %w", authErr) + } + wire.Grant.Witness, issuerPubkey = witness, pubkey + payload, err = json.Marshal(wire) + if err != nil { + return nil, fmt.Errorf("marshal authorized rho.run/v1 request: %w", err) + } + } command := r.Command if command == "" { @@ -92,6 +112,9 @@ func (r RhoV1) Run(ctx context.Context, req Request, handler EventHandler) (*Res args = []string{"run", "--request-file", "-", "--events", "jsonl"} } cmd := exec.CommandContext(ctx, command, args...) + if issuerPubkey != "" { + cmd.Env = append(os.Environ(), "RHO_PROTOCOL_GRANT_MODE=require", "RHO_PROTOCOL_GRANT_PUBKEY="+issuerPubkey) + } cmd.Dir = req.WorkingDir cmd.Stdin = strings.NewReader(string(payload)) stdout, err := cmd.StdoutPipe() @@ -123,6 +146,17 @@ func (r RhoV1) Run(ctx context.Context, req Request, handler EventHandler) (*Res return result, nil } +// canonicalJSON matches rho's serde_json Value encoding: object keys are +// lexicographically ordered and no insignificant whitespace is emitted. Grant +// signatures therefore cover identical bytes in Go, Lua, and Rust. +func canonicalJSON(payload []byte) ([]byte, error) { + var value any + if err := json.Unmarshal(payload, &value); err != nil { + return nil, err + } + return json.Marshal(value) +} + // ConsumeRhoV1 validates and reduces one rho.run/v1 JSONL stream. func ConsumeRhoV1(reader io.Reader, expectedRunID string, handler EventHandler) (*Result, error) { result := &Result{RunID: expectedRunID} diff --git a/pkg/executor/rho_v1_test.go b/pkg/executor/rho_v1_test.go index 127cad9..001c95e 100644 --- a/pkg/executor/rho_v1_test.go +++ b/pkg/executor/rho_v1_test.go @@ -1,10 +1,22 @@ package executor import ( + "bytes" "strings" "testing" ) +func TestCanonicalJSONSortsObjectKeysRecursively(t *testing.T) { + got, err := canonicalJSON([]byte(`{"z":1,"nested":{"b":2,"a":1},"a":0}`)) + if err != nil { + t.Fatal(err) + } + want := []byte(`{"a":0,"nested":{"a":1,"b":2},"z":1}`) + if !bytes.Equal(got, want) { + t.Fatalf("canonical JSON = %s, want %s", got, want) + } +} + func TestConsumeRhoV1CompletedStream(t *testing.T) { stream := strings.Join([]string{ `{"protocol":"rho.run/v1","run_id":"run-1","seq":1,"time":"2026-07-31T20:00:00Z","type":"run.started","data":{"provider":"anthropic"}}`, diff --git a/v2/adapters/adapters.go b/v2/adapters/adapters.go new file mode 100644 index 0000000..092e4cd --- /dev/null +++ b/v2/adapters/adapters.go @@ -0,0 +1,114 @@ +// Package adapters contains the narrow seams used by the SCUD v2 runtime. +// +// These interfaces intentionally describe workflow concerns rather than an +// LLM vendor, policy implementation, database, or task CLI. Concrete +// integrations live in subpackages (for example adapters/rho and +// adapters/shen), so the v2 graph and scheduler can be tested without any of +// those dependencies. +package adapters + +import "context" + +// Runner executes one bounded agent run. Model is an opaque routing string; +// interpreting it (for example as provider/model) is an adapter concern. +type Runner interface { + Run(context.Context, RunRequest, EventSink) (RunResult, error) +} + +type RunRequest struct { + RunID string + Prompt string + SystemPrompt string + Model string + WorkingDir string + AllowedTools []string + Limits Limits + Context map[string]any +} + +type Limits struct { + MaxTurns *uint32 + MaxInputTokens *uint64 + MaxOutputTokens *uint64 + MaxCostMicros *uint64 + Deadline string +} + +type RunResult struct { + RunID string + Text string + Outcome string + Failure *Failure + Usage Usage + ExitCode int + Stderr string +} + +type Failure struct { + Code string + Message string + Retryable bool + RetryAfterMS *uint64 +} + +type Usage struct { + InputTokens uint64 + OutputTokens uint64 + CacheReadTokens *uint64 + CostMicros *uint64 +} + +// Event is the provider-neutral progress envelope. Data is opaque to the +// runtime and owned by the producer's event namespace. +type Event struct { + RunID string + Sequence uint64 + Time string + Type string + Data []byte +} + +type EventSink func(Event) + +// Policy makes an authorization decision before an adapter performs an +// operation. Action and Resource are stable SCUD vocabulary; Attributes are +// extension data interpreted by a policy adapter. +type Policy interface { + Authorize(context.Context, PolicyInput) (Decision, error) +} + +type PolicyInput struct { + RunID string + Action string + Resource string + Attributes map[string]string +} + +type Decision struct { + Allowed bool + Reason string + Constraints map[string]string +} + +// EventStore persists append-only run events. Implementations must reject a +// duplicate or out-of-order sequence for the same run. +type EventStore interface { + Append(context.Context, Event) error + List(context.Context, string, uint64) ([]Event, error) +} + +// TaskDirectory is the v2 boundary around the legacy task database ("td"). +// Task is deliberately a small projection; adapters can retain richer legacy +// fields without leaking them into the runtime. +type TaskDirectory interface { + Get(context.Context, string) (Task, error) + SetStatus(context.Context, string, string) error + Ready(context.Context, string) ([]Task, error) +} + +type Task struct { + ID string + Title string + Status string + Dependencies []string +} diff --git a/v2/adapters/memory/events.go b/v2/adapters/memory/events.go new file mode 100644 index 0000000..27c7728 --- /dev/null +++ b/v2/adapters/memory/events.go @@ -0,0 +1,51 @@ +// Package memory provides deterministic test adapters for v2 contracts. It is +// not a production persistence layer. +package memory + +import ( + "context" + "errors" + "sort" + "sync" + + "github.com/reuben/scud/v2/adapters" +) + +type EventStore struct { + mu sync.RWMutex + events map[string][]adapters.Event +} + +func NewEventStore() *EventStore { return &EventStore{events: make(map[string][]adapters.Event)} } + +func (s *EventStore) Append(_ context.Context, event adapters.Event) error { + if event.RunID == "" || event.Sequence == 0 { + return errors.New("event run ID and sequence are required") + } + s.mu.Lock() + defer s.mu.Unlock() + list := s.events[event.RunID] + if len(list) > 0 && event.Sequence <= list[len(list)-1].Sequence { + return errors.New("event sequence must increase") + } + event.Data = append([]byte(nil), event.Data...) + s.events[event.RunID] = append(list, event) + return nil +} + +func (s *EventStore) List(_ context.Context, runID string, after uint64) ([]adapters.Event, error) { + s.mu.RLock() + defer s.mu.RUnlock() + var out []adapters.Event + for _, event := range s.events[runID] { + if event.Sequence > after { + copy := event + copy.Data = append([]byte(nil), event.Data...) + out = append(out, copy) + } + } + sort.Slice(out, func(i, j int) bool { return out[i].Sequence < out[j].Sequence }) + return out, nil +} + +var _ adapters.EventStore = (*EventStore)(nil) diff --git a/v2/adapters/memory/events_test.go b/v2/adapters/memory/events_test.go new file mode 100644 index 0000000..aadf90d --- /dev/null +++ b/v2/adapters/memory/events_test.go @@ -0,0 +1,25 @@ +package memory + +import ( + "context" + "testing" + + "github.com/reuben/scud/v2/adapters" +) + +func TestEventStoreEnforcesMonotonicSequenceAndCopiesData(t *testing.T) { + store := NewEventStore() + ctx := context.Background() + data := []byte("one") + if err := store.Append(ctx, adapters.Event{RunID: "r", Sequence: 1, Data: data}); err != nil { + t.Fatal(err) + } + data[0] = 'X' + if err := store.Append(ctx, adapters.Event{RunID: "r", Sequence: 1}); err == nil { + t.Fatal("expected duplicate sequence error") + } + got, err := store.List(ctx, "r", 0) + if err != nil || len(got) != 1 || string(got[0].Data) != "one" { + t.Fatalf("unexpected list: %+v, %v", got, err) + } +} diff --git a/v2/adapters/memory/tasks.go b/v2/adapters/memory/tasks.go new file mode 100644 index 0000000..445aba9 --- /dev/null +++ b/v2/adapters/memory/tasks.go @@ -0,0 +1,76 @@ +package memory + +import ( + "context" + "errors" + "sort" + "sync" + + "github.com/reuben/scud/v2/adapters" +) + +// TaskDirectory is a tiny test implementation of the td seam. Production +// adapters can map these calls to SCG, SQLite, or a remote task service. +type TaskDirectory struct { + mu sync.RWMutex + tasks map[string]adapters.Task +} + +func NewTaskDirectory(tasks []adapters.Task) *TaskDirectory { + byID := make(map[string]adapters.Task, len(tasks)) + for _, task := range tasks { + task.Dependencies = append([]string(nil), task.Dependencies...) + byID[task.ID] = task + } + return &TaskDirectory{tasks: byID} +} + +func (d *TaskDirectory) Get(_ context.Context, id string) (adapters.Task, error) { + d.mu.RLock() + defer d.mu.RUnlock() + task, ok := d.tasks[id] + if !ok { + return adapters.Task{}, errors.New("task not found") + } + task.Dependencies = append([]string(nil), task.Dependencies...) + return task, nil +} + +func (d *TaskDirectory) SetStatus(_ context.Context, id, status string) error { + d.mu.Lock() + defer d.mu.Unlock() + task, ok := d.tasks[id] + if !ok { + return errors.New("task not found") + } + task.Status = status + d.tasks[id] = task + return nil +} + +func (d *TaskDirectory) Ready(_ context.Context, _ string) ([]adapters.Task, error) { + d.mu.RLock() + defer d.mu.RUnlock() + var ready []adapters.Task + for _, task := range d.tasks { + if task.Status != "pending" { + continue + } + ok := true + for _, dependency := range task.Dependencies { + dep, exists := d.tasks[dependency] + if !exists || dep.Status != "done" { + ok = false + break + } + } + if ok { + task.Dependencies = append([]string(nil), task.Dependencies...) + ready = append(ready, task) + } + } + sort.Slice(ready, func(i, j int) bool { return ready[i].ID < ready[j].ID }) + return ready, nil +} + +var _ adapters.TaskDirectory = (*TaskDirectory)(nil) diff --git a/v2/adapters/memory/tasks_test.go b/v2/adapters/memory/tasks_test.go new file mode 100644 index 0000000..97da095 --- /dev/null +++ b/v2/adapters/memory/tasks_test.go @@ -0,0 +1,16 @@ +package memory + +import ( + "context" + "testing" + + "github.com/reuben/scud/v2/adapters" +) + +func TestTaskDirectoryReadyHonorsDependencies(t *testing.T) { + directory := NewTaskDirectory([]adapters.Task{{ID: "a", Status: "done"}, {ID: "b", Status: "pending", Dependencies: []string{"a"}}, {ID: "c", Status: "pending", Dependencies: []string{"missing"}}}) + ready, err := directory.Ready(context.Background(), "") + if err != nil || len(ready) != 1 || ready[0].ID != "b" { + t.Fatalf("unexpected ready tasks: %+v, %v", ready, err) + } +} diff --git a/v2/adapters/rho/rho.go b/v2/adapters/rho/rho.go new file mode 100644 index 0000000..35aac30 --- /dev/null +++ b/v2/adapters/rho/rho.go @@ -0,0 +1,107 @@ +// Package rho adapts the versioned rho.run/v1 JSONL protocol to the provider- +// neutral v2 adapters.Runner seam. No rho symbols are required by v2 core +// packages; this package is an optional integration. +package rho + +import ( + "context" + "fmt" + "io" + "strings" + + "github.com/reuben/scud/pkg/executor" + "github.com/reuben/scud/v2/adapters" +) + +// Runner invokes a rho.run/v1 producer. Command and Args are passed through +// to executor.RhoV1; an empty command uses rho-cli's normal lookup. +type Runner struct { + Command string + Args []string + Grant executor.Grant + Authorize func(context.Context, []byte) (string, string, error) +} + +func (r Runner) Run(ctx context.Context, req adapters.RunRequest, sink adapters.EventSink) (adapters.RunResult, error) { + provider, model, err := splitModel(req.Model) + if err != nil { + return adapters.RunResult{RunID: req.RunID}, err + } + inner := executor.RhoV1{Command: r.Command, Args: r.Args, Grant: r.Grant, Authorize: r.Authorize} + result, runErr := inner.Run(ctx, executor.Request{ + RunID: req.RunID, + Prompt: req.Prompt, + SystemPrompt: req.SystemPrompt, + Model: executor.ModelRef{Provider: provider, ID: model}, + WorkingDir: req.WorkingDir, + AllowedTools: req.AllowedTools, + Limits: executor.Limits{ + MaxTurns: req.Limits.MaxTurns, + MaxInputTokens: req.Limits.MaxInputTokens, + MaxOutputTokens: req.Limits.MaxOutputTokens, + MaxCostMicros: req.Limits.MaxCostMicros, + Deadline: req.Limits.Deadline, + }, + Context: req.Context, + }, func(event executor.Event) { + if sink == nil { + return + } + sink(adapters.Event{ + RunID: event.RunID, + Sequence: event.Sequence, + Time: event.Time, + Type: event.Type, + Data: append([]byte(nil), event.Data...), + }) + }) + if result == nil { + return adapters.RunResult{RunID: req.RunID}, runErr + } + return adapters.RunResult{ + RunID: result.RunID, + Text: result.Text, + Outcome: result.Outcome, + Failure: convertFailure(result.Failure), + Usage: adapters.Usage{ + InputTokens: result.Usage.InputTokens, + OutputTokens: result.Usage.OutputTokens, + CacheReadTokens: result.Usage.CacheReadTokens, + CostMicros: result.Usage.CostMicros, + }, + ExitCode: result.ExitCode, + Stderr: result.Stderr, + }, runErr +} + +// Consume validates a rho.run/v1 stream and translates events without +// starting a process. It is useful for conformance tests and alternate +// process transports. +func Consume(reader io.Reader, runID string, sink adapters.EventSink) (adapters.RunResult, error) { + result, err := executor.ConsumeRhoV1(reader, runID, func(event executor.Event) { + if sink != nil { + sink(adapters.Event{RunID: event.RunID, Sequence: event.Sequence, Time: event.Time, Type: event.Type, Data: append([]byte(nil), event.Data...)}) + } + }) + if result == nil { + return adapters.RunResult{RunID: runID}, err + } + return adapters.RunResult{RunID: result.RunID, Text: result.Text, Outcome: result.Outcome, Failure: convertFailure(result.Failure), Usage: adapters.Usage{InputTokens: result.Usage.InputTokens, OutputTokens: result.Usage.OutputTokens, CacheReadTokens: result.Usage.CacheReadTokens, CostMicros: result.Usage.CostMicros}, ExitCode: result.ExitCode, Stderr: result.Stderr}, err +} + +func splitModel(model string) (provider, id string, err error) { + provider, id, ok := strings.Cut(model, "/") + if !ok || provider == "" || id == "" || strings.Contains(id, "/") { + return "", "", fmt.Errorf("rho model %q must use provider/model form", model) + } + return provider, id, nil +} + +func convertFailure(failure *executor.Failure) *adapters.Failure { + if failure == nil { + return nil + } + return &adapters.Failure{Code: failure.Code, Message: failure.Message, Retryable: failure.Retryable, RetryAfterMS: failure.RetryAfterMS} +} + +var _ adapters.Runner = Runner{} diff --git a/v2/adapters/rho/rho_test.go b/v2/adapters/rho/rho_test.go new file mode 100644 index 0000000..273d6ce --- /dev/null +++ b/v2/adapters/rho/rho_test.go @@ -0,0 +1,47 @@ +package rho + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/reuben/scud/v2/adapters" +) + +func TestConsumeCompletedFixture(t *testing.T) { + file, err := os.Open(filepath.Join("testdata", "completed.jsonl")) + if err != nil { + t.Fatal(err) + } + defer file.Close() + var events []adapters.Event + result, err := Consume(file, "fixture-1", func(event adapters.Event) { events = append(events, event) }) + if err != nil { + t.Fatal(err) + } + if result.Text != "hello" || result.Outcome != "completed" || result.Usage.InputTokens != 4 { + t.Fatalf("unexpected result: %+v", result) + } + if len(events) != 3 || events[1].Type != "message.delta" { + t.Fatalf("unexpected events: %+v", events) + } +} + +func TestConsumeRejectsAfterTerminal(t *testing.T) { + file, err := os.Open(filepath.Join("testdata", "invalid_after_terminal.jsonl")) + if err != nil { + t.Fatal(err) + } + defer file.Close() + if _, err := Consume(file, "fixture-2", nil); err == nil { + t.Fatal("expected terminal invariant error") + } +} + +func TestRunnerRejectsProviderlessModelBeforeProcess(t *testing.T) { + _, err := (Runner{}).Run(context.Background(), adapters.RunRequest{RunID: "x", Model: "claude"}, nil) + if err == nil { + t.Fatal("expected provider/model validation error") + } +} diff --git a/v2/adapters/rho/testdata/completed.jsonl b/v2/adapters/rho/testdata/completed.jsonl new file mode 100644 index 0000000..18fa637 --- /dev/null +++ b/v2/adapters/rho/testdata/completed.jsonl @@ -0,0 +1,3 @@ +{"protocol":"rho.run/v1","run_id":"fixture-1","seq":1,"time":"2026-07-31T20:00:00Z","type":"run.started","data":{"provider":"anthropic"}} +{"protocol":"rho.run/v1","run_id":"fixture-1","seq":2,"time":"2026-07-31T20:00:01Z","type":"message.delta","data":{"text":"hello"}} +{"protocol":"rho.run/v1","run_id":"fixture-1","seq":3,"time":"2026-07-31T20:00:02Z","type":"run.completed","data":{"status":"succeeded","usage":{"input_tokens":4,"output_tokens":1}}} diff --git a/v2/adapters/rho/testdata/invalid_after_terminal.jsonl b/v2/adapters/rho/testdata/invalid_after_terminal.jsonl new file mode 100644 index 0000000..df88c7a --- /dev/null +++ b/v2/adapters/rho/testdata/invalid_after_terminal.jsonl @@ -0,0 +1,2 @@ +{"protocol":"rho.run/v1","run_id":"fixture-2","seq":1,"time":"2026-07-31T20:00:00Z","type":"run.cancelled","data":{"reason":"user"}} +{"protocol":"rho.run/v1","run_id":"fixture-2","seq":2,"time":"2026-07-31T20:00:01Z","type":"message.delta","data":{"text":"late"}} diff --git a/v2/adapters/shen/shen.go b/v2/adapters/shen/shen.go new file mode 100644 index 0000000..79c0ceb --- /dev/null +++ b/v2/adapters/shen/shen.go @@ -0,0 +1,56 @@ +// Package shen contains the optional Shen policy integration. The v2 runtime +// only depends on adapters.Policy; Shen-specific request/decision plumbing is +// kept here. +package shen + +import ( + "context" + "errors" + + "github.com/reuben/scud/v2/adapters" +) + +// Evaluator is the small portion of a Shen engine needed by SCUD. Keeping it +// local avoids coupling the v2 module to a particular Shen SDK version. +type Evaluator interface { + Evaluate(context.Context, Input) (Output, error) +} + +type Input struct { + RunID string + Action string + Resource string + Attributes map[string]string +} + +type Output struct { + Allowed bool + Reason string + Constraints map[string]string +} + +type Policy struct{ Engine Evaluator } + +func (p Policy) Authorize(ctx context.Context, in adapters.PolicyInput) (adapters.Decision, error) { + if p.Engine == nil { + return adapters.Decision{}, errors.New("shen policy engine is nil") + } + out, err := p.Engine.Evaluate(ctx, Input{RunID: in.RunID, Action: in.Action, Resource: in.Resource, Attributes: clone(in.Attributes)}) + if err != nil { + return adapters.Decision{}, err + } + return adapters.Decision{Allowed: out.Allowed, Reason: out.Reason, Constraints: clone(out.Constraints)}, nil +} + +func clone(values map[string]string) map[string]string { + if len(values) == 0 { + return nil + } + copy := make(map[string]string, len(values)) + for key, value := range values { + copy[key] = value + } + return copy +} + +var _ adapters.Policy = Policy{} diff --git a/v2/adapters/shen/shen_test.go b/v2/adapters/shen/shen_test.go new file mode 100644 index 0000000..8ad5cfd --- /dev/null +++ b/v2/adapters/shen/shen_test.go @@ -0,0 +1,33 @@ +package shen + +import ( + "context" + "testing" + + "github.com/reuben/scud/v2/adapters" +) + +type fakeEngine struct{ seen Input } + +func (f *fakeEngine) Evaluate(_ context.Context, input Input) (Output, error) { + f.seen = input + return Output{Allowed: true, Reason: "test", Constraints: map[string]string{"root": "/tmp"}}, nil +} + +func TestPolicyTranslatesWithoutProviderFields(t *testing.T) { + engine := &fakeEngine{} + policy := Policy{Engine: engine} + decision, err := policy.Authorize(context.Background(), adapters.PolicyInput{RunID: "r", Action: "read", Resource: "file", Attributes: map[string]string{"path": "x"}}) + if err != nil || !decision.Allowed || decision.Constraints["root"] != "/tmp" { + t.Fatalf("unexpected decision: %+v, %v", decision, err) + } + if engine.seen.Attributes["path"] != "x" { + t.Fatalf("input was not translated: %+v", engine.seen) + } +} + +func TestPolicyRequiresEngine(t *testing.T) { + if _, err := (Policy{}).Authorize(context.Background(), adapters.PolicyInput{}); err == nil { + t.Fatal("expected nil engine error") + } +} diff --git a/v2/adapters/tdhttp/tdhttp.go b/v2/adapters/tdhttp/tdhttp.go new file mode 100644 index 0000000..cb19e09 --- /dev/null +++ b/v2/adapters/tdhttp/tdhttp.go @@ -0,0 +1,146 @@ +// Package tdhttp connects SCUD v2 durability to td's fenced agent-run API. +package tdhttp + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "strings" + "sync" + + "github.com/reuben/scud/v2/adapters" +) + +type Client struct { + BaseURL string + Token string + TicketID string + Attempt uint64 + ExecutionID string + HTTP *http.Client + mu sync.Mutex + tails map[string]uint64 +} + +type Run struct { + ID, GoalID, ConfigDigest string +} + +func (c *Client) CreateRun(ctx context.Context, run Run) error { + body := map[string]any{"id": run.ID, "ticket_id": c.TicketID, "goal_id": run.GoalID, + "config_digest": run.ConfigDigest, "attempt": c.Attempt, + "execution_id": c.ExecutionID, "idempotency_key": "scud:create:" + run.ID} + return c.call(ctx, http.MethodPost, "/v1/agent/runs", body, nil) +} + +// Authorize signs the exact unsigned rho.run/v1 request under td's live +// Shen-authorized claim. It matches executor.RhoV1.Authorize. +func (c *Client) Authorize(ctx context.Context, runID string, unsigned []byte) (string, string, error) { + digest := fmt.Sprintf("%x", sha256.Sum256(unsigned)) + body := map[string]any{"request_sha256": digest, "unsigned_request": string(unsigned), "attempt": c.Attempt, + "execution_id": c.ExecutionID, "idempotency_key": "scud:grant:" + runID + ":" + digest} + var response struct { + Allowed bool `json:"allowed"` + Witness string `json:"witness"` + IssuerPubkey string `json:"issuer_pubkey"` + } + if err := c.call(ctx, http.MethodPost, "/v1/agent/runs/"+url.PathEscape(runID)+"/grants", body, &response); err != nil { + return "", "", err + } + if !response.Allowed || response.Witness == "" || response.IssuerPubkey == "" { + return "", "", fmt.Errorf("td denied execution grant") + } + return response.Witness, response.IssuerPubkey, nil +} + +func (c *Client) Append(ctx context.Context, event adapters.Event) error { + c.mu.Lock() + defer c.mu.Unlock() + if c.tails == nil { + c.tails = map[string]uint64{} + } + tail := c.tails[event.RunID] + if tail == 0 && event.Sequence > 1 { + tail = event.Sequence - 1 + } + body := map[string]any{"expected_tail": tail, "attempt": c.Attempt, + "execution_id": c.ExecutionID, "idempotency_key": fmt.Sprintf("scud:%s:%d", event.RunID, event.Sequence), + "event": map[string]any{"sequence": event.Sequence, "event_id": fmt.Sprintf("%s:%d", event.RunID, event.Sequence), + "type": event.Type, "ts": event.Time, "payload": json.RawMessage(event.Data), + "idempotency_key": fmt.Sprintf("scud:%s:%d", event.RunID, event.Sequence)}} + if err := c.call(ctx, http.MethodPost, "/v1/agent/runs/"+url.PathEscape(event.RunID)+"/events", body, nil); err != nil { + return err + } + c.tails[event.RunID] = event.Sequence + return nil +} + +func (c *Client) List(ctx context.Context, runID string, after uint64) ([]adapters.Event, error) { + var response struct { + Events []struct { + Sequence uint64 `json:"sequence"` + Type, TS string + Payload json.RawMessage + } `json:"events"` + Tail uint64 `json:"tail"` + } + path := "/v1/agent/runs/" + url.PathEscape(runID) + "/events?after=" + strconv.FormatUint(after, 10) + if err := c.call(ctx, http.MethodGet, path, nil, &response); err != nil { + return nil, err + } + c.mu.Lock() + if c.tails == nil { + c.tails = map[string]uint64{} + } + c.tails[runID] = response.Tail + c.mu.Unlock() + out := make([]adapters.Event, len(response.Events)) + for i, event := range response.Events { + out[i] = adapters.Event{RunID: runID, Sequence: event.Sequence, Time: event.TS, Type: event.Type, Data: append([]byte(nil), event.Payload...)} + } + return out, nil +} + +func (c *Client) call(ctx context.Context, method, path string, body any, out any) error { + var reader io.Reader + if body != nil { + encoded, err := json.Marshal(body) + if err != nil { + return err + } + reader = bytes.NewReader(encoded) + } + req, err := http.NewRequestWithContext(ctx, method, strings.TrimRight(c.BaseURL, "/")+path, reader) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+c.Token) + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + httpClient := c.HTTP + if httpClient == nil { + httpClient = http.DefaultClient + } + resp, err := httpClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + data, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("td %s %s returned %d: %s", method, path, resp.StatusCode, strings.TrimSpace(string(data))) + } + if out != nil && len(data) > 0 { + return json.Unmarshal(data, out) + } + return nil +} + +var _ adapters.EventStore = (*Client)(nil) diff --git a/v2/adapters/tdhttp/tdhttp_test.go b/v2/adapters/tdhttp/tdhttp_test.go new file mode 100644 index 0000000..0372dd5 --- /dev/null +++ b/v2/adapters/tdhttp/tdhttp_test.go @@ -0,0 +1,47 @@ +package tdhttp + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/reuben/scud/v2/adapters" +) + +func TestAppendAndReplayCarryFenceAndTail(t *testing.T) { + var appended map[string]any + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer token" { + t.Fatal("missing bearer") + } + switch { + case r.Method == http.MethodPost && r.URL.Path == "/v1/agent/runs/run-1/events": + if err := json.NewDecoder(r.Body).Decode(&appended); err != nil { + t.Fatal(err) + } + w.WriteHeader(http.StatusCreated) + _, _ = w.Write([]byte(`{"tail":1}`)) + case r.Method == http.MethodGet: + _, _ = w.Write([]byte(`{"tail":1,"events":[{"sequence":1,"type":"core/goal_created","ts":"now","payload":{"kind":"goal_created"}}]}`)) + default: + t.Fatalf("unexpected %s %s", r.Method, r.URL.Path) + } + })) + defer srv.Close() + c := &Client{BaseURL: srv.URL, Token: "token", TicketID: "ticket", Attempt: 3, ExecutionID: "exec"} + if err := c.Append(context.Background(), adapters.Event{RunID: "run-1", Sequence: 1, Type: "core/goal_created", Data: []byte(`{"kind":"goal_created"}`)}); err != nil { + t.Fatal(err) + } + if appended["expected_tail"].(float64) != 0 || appended["attempt"].(float64) != 3 || appended["execution_id"] != "exec" { + t.Fatalf("bad fenced append: %#v", appended) + } + events, err := c.List(context.Background(), "run-1", 0) + if err != nil { + t.Fatal(err) + } + if len(events) != 1 || events[0].Sequence != 1 { + t.Fatalf("bad replay: %#v", events) + } +} diff --git a/v2/core/core_test.go b/v2/core/core_test.go new file mode 100644 index 0000000..22cc2a0 --- /dev/null +++ b/v2/core/core_test.go @@ -0,0 +1,192 @@ +package core + +import "testing" + +func eventGoal(id ID) Event { + return Event{ID: ID("event-" + string(id)), Kind: EventGoalCreated, Goal: Goal{ID: id}} +} + +func TestReduceDoesNotMutateInputAndReplayMatches(t *testing.T) { + s := NewState() + var err error + s, err = s.Append(eventGoal("g")) + if err != nil { + t.Fatal(err) + } + dep := []ID{"a"} + s2, err := s.Append(Event{Kind: EventObligationAdded, Obligation: Obligation{ID: "a", GoalID: "g"}}) + if err != nil { + t.Fatal(err) + } + s2, err = s2.Append(Event{Kind: EventObligationAdded, Obligation: Obligation{ID: "b", GoalID: "g", DependsOn: dep}}) + if err != nil { + t.Fatal(err) + } + dep[0] = "changed" + if got := s2.Obligations["b"].DependsOn[0]; got != "a" { + t.Fatalf("reducer did not copy dependencies: %q", got) + } + s2, err = s2.Append(Event{Kind: EventDecisionIssued, Decision: Decision{ID: "d", GoalID: "g", ObligationID: "a", Kind: DecisionExecute}}) + if err != nil { + t.Fatal(err) + } + s2, err = s2.Append(Event{Kind: EventObservationAdded, Observation: Observation{ID: "o", GoalID: "g", ObligationID: "a", Outcome: OutcomeSuccess}}) + if err != nil { + t.Fatal(err) + } + s2, err = s2.Append(Event{Kind: EventDecisionIssued, Decision: Decision{ID: "d2", GoalID: "g", ObligationID: "b", Kind: DecisionExecute}}) + if err != nil { + t.Fatal(err) + } + s2, err = s2.Append(Event{Kind: EventObservationAdded, Observation: Observation{ID: "o2", GoalID: "g", ObligationID: "b", Outcome: OutcomeSuccess}}) + if err != nil { + t.Fatal(err) + } + replayed, err := Replay(s2.Events) + if err != nil { + t.Fatal(err) + } + if replayed.Revision != s2.Revision || replayed.Goals["g"].Status != GoalSucceeded { + t.Fatalf("replay differs: %#v %#v", replayed, s2) + } + if s.Goals["g"].Status != GoalPending { + t.Fatalf("append mutated prior state") + } +} + +func TestEventDataIsCopied(t *testing.T) { + s := NewState() + s, err := s.Append(eventGoal("g")) + if err != nil { + t.Fatal(err) + } + data := []byte("opaque") + s, err = s.Append(Event{Kind: EventBudgetConsumed, Data: data, Budget: BudgetDelta{Steps: 1}}) + if err != nil { + t.Fatal(err) + } + data[0] = 'X' + if string(s.Events[1].Data) != "opaque" { + t.Fatalf("event data was aliased: %q", s.Events[1].Data) + } +} + +func TestValidationRejectsCrossGoalAndCycles(t *testing.T) { + s := NewState() + var err error + s, err = s.Append(eventGoal("g1")) + if err != nil { + t.Fatal(err) + } + s, err = s.Append(Event{Kind: EventGoalCreated, Goal: Goal{ID: "g2"}}) + if err != nil { + t.Fatal(err) + } + s, err = s.Append(Event{Kind: EventObligationAdded, Obligation: Obligation{ID: "a", GoalID: "g1", DependsOn: []ID{"b"}}}) + if err == nil { + t.Fatal("expected cycle/reference error") + } + // Direct mutation is rejected by final validation as well. + s = NewState() + s, _ = s.Append(eventGoal("g")) + s, _ = s.Append(Event{Kind: EventObligationAdded, Obligation: Obligation{ID: "a", GoalID: "g"}}) + s.Obligations["a"] = Obligation{ID: "a", GoalID: "g", DependsOn: []ID{"a"}, Status: ObligationPending} + if err := s.Validate(); err == nil { + t.Fatal("expected self-cycle error") + } +} + +func TestReconcileIsDeterministicAndBudgetAware(t *testing.T) { + s := NewState() + var err error + s, err = s.Append(eventGoal("g")) + if err != nil { + t.Fatal(err) + } + s, err = s.Append(Event{Kind: EventObligationAdded, Obligation: Obligation{ID: "z", GoalID: "g"}}) + if err != nil { + t.Fatal(err) + } + s, err = s.Append(Event{Kind: EventObligationAdded, Obligation: Obligation{ID: "a", GoalID: "g"}}) + if err != nil { + t.Fatal(err) + } + d, err := Reconcile(s) + if err != nil { + t.Fatal(err) + } + if d.Kind != DecisionExecute || d.ObligationID != "a" { + t.Fatalf("got %#v", d) + } + s.Budget = Budget{MaxSteps: 1, UsedSteps: 1} + d, err = Reconcile(s) + if err != nil { + t.Fatal(err) + } + if d.Kind != DecisionExhaust { + t.Fatalf("got %#v", d) + } +} + +func TestReconcileBlocksWhenDependencyCannotComplete(t *testing.T) { + s := NewState() + var err error + s, err = s.Append(eventGoal("g")) + if err != nil { + t.Fatal(err) + } + s, err = s.Append(Event{Kind: EventObligationAdded, Obligation: Obligation{ID: "a", GoalID: "g"}}) + if err != nil { + t.Fatal(err) + } + s, err = s.Append(Event{Kind: EventObligationAdded, Obligation: Obligation{ID: "b", GoalID: "g", DependsOn: []ID{"a"}}}) + if err != nil { + t.Fatal(err) + } + s.Obligations["a"] = Obligation{ID: "a", GoalID: "g", Status: ObligationFailed} + d, err := Reconcile(s) + if err != nil { + t.Fatal(err) + } + if d.Kind != DecisionFail || d.ObligationID != "a" { + t.Fatalf("expected failed prerequisite decision, got %#v", d) + } + s.Obligations["a"] = Obligation{ID: "a", GoalID: "g", Status: ObligationSucceeded} + s.Obligations["b"] = Obligation{ID: "b", GoalID: "g", DependsOn: []ID{"a"}, Status: ObligationPending} + d, err = Reconcile(s) + if err != nil { + t.Fatal(err) + } + if d.Kind != DecisionExecute || d.ObligationID != "b" { + t.Fatalf("expected b execution, got %#v", d) + } +} + +func TestBudgetAndSequenceValidation(t *testing.T) { + s := NewState() + var err error + s, err = s.Append(eventGoal("g")) + if err != nil { + t.Fatal(err) + } + if _, err = Reduce(s, Event{Sequence: 4, Kind: EventGoalCreated, Goal: Goal{ID: "x"}}); err == nil { + t.Fatal("expected sequence error") + } + s.Budget = Budget{MaxSteps: 1} + if _, err = s.Append(Event{Kind: EventBudgetConsumed, Budget: BudgetDelta{Steps: 2}}); err == nil { + t.Fatal("expected budget error") + } +} + +func TestTerminalStatuses(t *testing.T) { + for _, status := range []GoalStatus{GoalSucceeded, GoalFailed, GoalBlocked, GoalCancelled, GoalExhausted} { + if !status.terminal() { + t.Errorf("%q not terminal", status) + } + } + for _, status := range []ObligationStatus{ObligationSucceeded, ObligationFailed, ObligationBlocked, ObligationCancelled} { + if !status.terminal() { + t.Errorf("%q not terminal", status) + } + } +} diff --git a/v2/core/doc.go b/v2/core/doc.go new file mode 100644 index 0000000..3c20a0b --- /dev/null +++ b/v2/core/doc.go @@ -0,0 +1,13 @@ +// Package core contains the provider-independent domain kernel for SCUD v2. +// +// The package deliberately has no I/O, clocks, persistence, provider SDKs, or +// application dependencies. A run is represented by State and advanced by +// applying Events with Reduce. Replay applies the same events from an empty +// state and therefore produces the same result. Reconcile computes a +// deterministic next Decision from a state; it does not perform that decision. +// +// Values returned by constructors and reducers own their slices and maps. The +// public structs are intentionally small and can be serialized by an adapter, +// but callers should treat values as immutable and use the returned value from +// each operation rather than changing a State in place. +package core diff --git a/v2/core/reducer.go b/v2/core/reducer.go new file mode 100644 index 0000000..2ca8e40 --- /dev/null +++ b/v2/core/reducer.go @@ -0,0 +1,343 @@ +package core + +import ( + "fmt" + "sort" +) + +// Reduce applies one event without mutating the input state. Sequence zero is +// accepted for manually constructed events; non-zero sequences must be the +// next revision. +func Reduce(state State, event Event) (State, error) { + if !event.Kind.valid() { + return state, fmt.Errorf("invalid event kind %q", event.Kind) + } + if event.Sequence != 0 && event.Sequence != state.Revision+1 { + return state, fmt.Errorf("event sequence %d does not follow revision %d", event.Sequence, state.Revision) + } + n := state.Clone() + if n.Goals == nil { + n.Goals = map[ID]Goal{} + } + if n.Obligations == nil { + n.Obligations = map[ID]Obligation{} + } + if n.Observations == nil { + n.Observations = map[ID]Observation{} + } + if event.ID != "" { + for _, prior := range n.Events { + if prior.ID == event.ID { + return state, fmt.Errorf("duplicate event id %q", event.ID) + } + } + } + + switch event.Kind { + case EventGoalCreated: + g := event.Goal + if err := ValidateID(g.ID); err != nil { + return state, err + } + if g.Status == "" { + g.Status = GoalPending + } + if !g.Status.valid() { + return state, fmt.Errorf("invalid goal status %q", g.Status) + } + if _, exists := n.Goals[g.ID]; exists { + return state, fmt.Errorf("goal %q already exists", g.ID) + } + n.Goals[g.ID] = g + case EventObligationAdded: + o := event.Obligation + if err := ValidateID(o.ID); err != nil { + return state, err + } + if _, exists := n.Goals[o.GoalID]; !exists { + return state, fmt.Errorf("unknown goal %q", o.GoalID) + } + if o.Status == "" { + o.Status = ObligationPending + } + if !o.Status.valid() { + return state, fmt.Errorf("invalid obligation status %q", o.Status) + } + if _, exists := n.Obligations[o.ID]; exists { + return state, fmt.Errorf("obligation %q already exists", o.ID) + } + o.DependsOn = append([]ID(nil), o.DependsOn...) + for _, dep := range o.DependsOn { + d, exists := n.Obligations[dep] + if !exists || d.GoalID != o.GoalID { + return state, fmt.Errorf("obligation %q: unknown or cross-goal dependency %q", o.ID, dep) + } + } + n.Obligations[o.ID] = o + if err := validateAcyclic(n.Obligations); err != nil { + return state, err + } + case EventObservationAdded: + ob := event.Observation + if err := ValidateID(ob.ID); err != nil { + return state, err + } + if _, exists := n.Observations[ob.ID]; exists { + return state, fmt.Errorf("observation %q already exists", ob.ID) + } + target, ok := n.Obligations[ob.ObligationID] + if !ok || target.GoalID != ob.GoalID { + return state, fmt.Errorf("observation targets unknown obligation %q", ob.ObligationID) + } + if !ob.Outcome.valid() { + return state, fmt.Errorf("invalid observation outcome %q", ob.Outcome) + } + ob.Evidence = append([]Evidence(nil), ob.Evidence...) + n.Observations[ob.ID] = ob + switch ob.Outcome { + case OutcomeSuccess: + target.Status = ObligationSucceeded + case OutcomeFailure: + target.Status = ObligationFailed + } + n.Obligations[target.ID] = target + refreshGoal(&n, target.GoalID) + case EventDecisionIssued: + d := event.Decision + if err := validateDecision(n, d); err != nil { + return state, err + } + if d.ID != "" { + for _, prior := range n.Decisions { + if prior.ID == d.ID { + return state, fmt.Errorf("duplicate decision id %q", d.ID) + } + } + } + n.Decisions = append(n.Decisions, d) + if d.ObligationID != "" { + o := n.Obligations[d.ObligationID] + switch d.Kind { + case DecisionExecute: + o.Status = ObligationRunning + case DecisionCancel: + o.Status = ObligationCancelled + case DecisionBlock: + o.Status = ObligationBlocked + case DecisionSucceed: + o.Status = ObligationSucceeded + case DecisionFail: + o.Status = ObligationFailed + } + n.Obligations[o.ID] = o + refreshGoal(&n, o.GoalID) + } else if g, ok := n.Goals[d.GoalID]; ok { + switch d.Kind { + case DecisionSucceed: + g.Status = GoalSucceeded + case DecisionFail: + g.Status = GoalFailed + case DecisionBlock: + g.Status = GoalBlocked + case DecisionCancel: + g.Status = GoalCancelled + case DecisionExhaust: + g.Status = GoalExhausted + } + n.Goals[g.ID] = g + } + case EventBudgetConsumed: + if event.Budget.Steps == 0 && event.Budget.Cost == 0 { + return state, fmt.Errorf("budget delta is empty") + } + if ^uint64(0)-n.Budget.UsedSteps < event.Budget.Steps || ^uint64(0)-n.Budget.UsedCost < event.Budget.Cost { + return state, fmt.Errorf("budget overflow") + } + n.Budget.UsedSteps += event.Budget.Steps + n.Budget.UsedCost += event.Budget.Cost + if n.Budget.MaxSteps != 0 && n.Budget.UsedSteps > n.Budget.MaxSteps || n.Budget.MaxCost != 0 && n.Budget.UsedCost > n.Budget.MaxCost { + return state, fmt.Errorf("budget exceeded") + } + } + + n.Revision++ + e := cloneEvent(event) + e.Sequence = n.Revision + n.Events = append(n.Events, e) + return n, nil +} + +// Append assigns the next sequence number and applies event. +func (s State) Append(event Event) (State, error) { + event.Sequence = s.Revision + 1 + return Reduce(s, event) +} + +// Replay deterministically applies an event stream to an empty state. Zero +// sequence numbers are assigned their stream position. +func Replay(events []Event) (State, error) { + s := NewState() + for i, e := range events { + if e.Sequence == 0 { + e.Sequence = uint64(i + 1) + } + var err error + s, err = Reduce(s, e) + if err != nil { + return State{}, fmt.Errorf("event %d: %w", i, err) + } + } + return s, nil +} + +func validateDecision(s State, d Decision) error { + if !d.Kind.valid() { + return fmt.Errorf("invalid decision kind %q", d.Kind) + } + g, ok := s.Goals[d.GoalID] + if !ok { + return fmt.Errorf("decision references unknown goal %q", d.GoalID) + } + if g.Status.terminal() && d.Kind == DecisionExecute { + return fmt.Errorf("goal %q is terminal", d.GoalID) + } + if d.ObligationID != "" { + o, ok := s.Obligations[d.ObligationID] + if !ok || o.GoalID != d.GoalID { + return fmt.Errorf("decision references invalid obligation %q", d.ObligationID) + } + if o.Status.terminal() && d.Kind == DecisionExecute { + return fmt.Errorf("obligation %q is terminal", d.ObligationID) + } + } + return nil +} + +func refreshGoal(s *State, goalID ID) { + g, ok := s.Goals[goalID] + if !ok || g.Status.terminal() { + return + } + count, succeeded := 0, 0 + for _, o := range s.Obligations { + if o.GoalID != goalID { + continue + } + count++ + if o.Status == ObligationSucceeded { + succeeded++ + } + if o.Status == ObligationFailed { + g.Status = GoalFailed + s.Goals[goalID] = g + return + } + if o.Status == ObligationBlocked { + g.Status = GoalBlocked + s.Goals[goalID] = g + return + } + } + if count > 0 && succeeded == count { + g.Status = GoalSucceeded + } else if count > 0 { + g.Status = GoalRunning + } + s.Goals[goalID] = g +} + +// Reconcile computes the next deterministic proposal. It considers goals and +// obligations by lexicographic ID, so map iteration order cannot affect it. +func Reconcile(s State) (Decision, error) { + if err := s.Validate(); err != nil { + return Decision{}, err + } + goals := make([]ID, 0, len(s.Goals)) + for id := range s.Goals { + goals = append(goals, id) + } + sort.Slice(goals, func(i, j int) bool { return goals[i] < goals[j] }) + for _, gid := range goals { + g := s.Goals[gid] + if g.Status.terminal() { + continue + } + if s.Budget.MaxSteps != 0 && s.Budget.UsedSteps >= s.Budget.MaxSteps || s.Budget.MaxCost != 0 && s.Budget.UsedCost >= s.Budget.MaxCost { + return Decision{ID: ID(fmt.Sprintf("decision:%d:%s", s.Revision+1, gid)), GoalID: gid, Kind: DecisionExhaust, Reason: "budget exhausted"}, nil + } + var nodes []Obligation + for _, o := range s.Obligations { + if o.GoalID == gid { + nodes = append(nodes, o) + } + } + sort.Slice(nodes, func(i, j int) bool { return nodes[i].ID < nodes[j].ID }) + if len(nodes) == 0 { + return Decision{ID: ID(fmt.Sprintf("decision:%d:%s", s.Revision+1, gid)), GoalID: gid, Kind: DecisionWait, Reason: "goal has no obligations"}, nil + } + for _, o := range nodes { + status := o.Status + if status == "" { + status = ObligationPending + } + if status == ObligationFailed { + return Decision{ID: ID(fmt.Sprintf("decision:%d:%s", s.Revision+1, gid)), GoalID: gid, ObligationID: o.ID, Kind: DecisionFail, Reason: "obligation failed"}, nil + } + if status == ObligationBlocked { + return Decision{ID: ID(fmt.Sprintf("decision:%d:%s", s.Revision+1, gid)), GoalID: gid, ObligationID: o.ID, Kind: DecisionBlock, Reason: "obligation blocked"}, nil + } + if status == ObligationCancelled { + return Decision{ID: ID(fmt.Sprintf("decision:%d:%s", s.Revision+1, gid)), GoalID: gid, ObligationID: o.ID, Kind: DecisionBlock, Reason: "obligation cancelled"}, nil + } + } + for _, o := range nodes { + status := o.Status + if status == "" { + status = ObligationPending + } + if status != ObligationPending && status != ObligationReady { + continue + } + for _, dep := range o.DependsOn { + if s.Obligations[dep].Status == ObligationFailed || s.Obligations[dep].Status == ObligationBlocked || s.Obligations[dep].Status == ObligationCancelled { + return Decision{ID: ID(fmt.Sprintf("decision:%d:%s", s.Revision+1, gid)), GoalID: gid, ObligationID: o.ID, Kind: DecisionBlock, Reason: "dependency cannot complete"}, nil + } + } + } + for _, o := range nodes { + status := o.Status + if status == "" { + status = ObligationPending + } + if status != ObligationPending && status != ObligationReady { + continue + } + ready := true + for _, dep := range o.DependsOn { + if s.Obligations[dep].Status != ObligationSucceeded { + ready = false + break + } + } + if ready { + return Decision{ID: ID(fmt.Sprintf("decision:%d:%s:%s", s.Revision+1, gid, o.ID)), GoalID: gid, ObligationID: o.ID, Kind: DecisionExecute, Reason: "dependencies satisfied"}, nil + } + } + allDone := true + for _, o := range nodes { + status := o.Status + if status == "" { + status = ObligationPending + } + if !status.terminal() { + allDone = false + break + } + } + if allDone { + return Decision{ID: ID(fmt.Sprintf("decision:%d:%s", s.Revision+1, gid)), GoalID: gid, Kind: DecisionSucceed, Reason: "all obligations terminal"}, nil + } + return Decision{ID: ID(fmt.Sprintf("decision:%d:%s", s.Revision+1, gid)), GoalID: gid, Kind: DecisionWait, Reason: "awaiting observations"}, nil + } + return Decision{}, fmt.Errorf("no active goals") +} diff --git a/v2/core/types.go b/v2/core/types.go new file mode 100644 index 0000000..f92f7da --- /dev/null +++ b/v2/core/types.go @@ -0,0 +1,268 @@ +package core + +import "fmt" + +// ID identifies a domain object. IDs are compared lexicographically when the +// kernel needs a stable ordering. +type ID string + +// GoalStatus is the lifecycle state of a goal. +type GoalStatus string + +const ( + GoalPending GoalStatus = "pending" + GoalRunning GoalStatus = "running" + GoalSucceeded GoalStatus = "succeeded" + GoalFailed GoalStatus = "failed" + GoalBlocked GoalStatus = "blocked" + GoalCancelled GoalStatus = "cancelled" + GoalExhausted GoalStatus = "exhausted" +) + +func (s GoalStatus) valid() bool { + switch s { + case GoalPending, GoalRunning, GoalSucceeded, GoalFailed, GoalBlocked, GoalCancelled, GoalExhausted: + return true + default: + return false + } +} + +func (s GoalStatus) terminal() bool { + return s == GoalSucceeded || s == GoalFailed || s == GoalBlocked || s == GoalCancelled || s == GoalExhausted +} + +// IsTerminal reports whether a goal can no longer be advanced. +func (s GoalStatus) IsTerminal() bool { return s.terminal() } + +// ObligationStatus is the lifecycle state of one DAG node. +type ObligationStatus string + +const ( + ObligationPending ObligationStatus = "pending" + ObligationReady ObligationStatus = "ready" + ObligationRunning ObligationStatus = "running" + ObligationSucceeded ObligationStatus = "succeeded" + ObligationFailed ObligationStatus = "failed" + ObligationBlocked ObligationStatus = "blocked" + ObligationCancelled ObligationStatus = "cancelled" +) + +func (s ObligationStatus) valid() bool { + switch s { + case ObligationPending, ObligationReady, ObligationRunning, ObligationSucceeded, ObligationFailed, ObligationBlocked, ObligationCancelled: + return true + default: + return false + } +} + +func (s ObligationStatus) terminal() bool { + return s == ObligationSucceeded || s == ObligationFailed || s == ObligationBlocked || s == ObligationCancelled +} + +// IsTerminal reports whether an obligation can no longer be advanced. +func (s ObligationStatus) IsTerminal() bool { return s.terminal() } + +// Goal describes the unit being reconciled. Status is normally maintained by +// the reducer; a zero status is interpreted as pending by NewState. +type Goal struct { + ID ID `json:"id"` + Title string `json:"title,omitempty"` + Description string `json:"description,omitempty"` + Status GoalStatus `json:"status"` +} + +// Obligation is a node in a goal's dependency DAG. Dependencies must belong to +// the same goal and are completed before this node can run. +type Obligation struct { + ID ID `json:"id"` + GoalID ID `json:"goal_id"` + Description string `json:"description,omitempty"` + DependsOn []ID `json:"depends_on,omitempty"` + CapabilityRef string `json:"capability_ref,omitempty"` + PolicyRef string `json:"policy_ref,omitempty"` + GrantRef string `json:"grant_ref,omitempty"` + Status ObligationStatus `json:"status"` +} + +// Observation is an external, provider-neutral report about an obligation. +// The kernel consumes observations but does not interpret their Evidence +// beyond retaining it for replay and auditing. +type Observation struct { + ID ID `json:"id"` + GoalID ID `json:"goal_id"` + ObligationID ID `json:"obligation_id"` + Outcome Outcome `json:"outcome"` + Detail string `json:"detail,omitempty"` + Evidence []Evidence `json:"evidence,omitempty"` +} + +// Outcome is the result reported by an observation. +type Outcome string + +const ( + OutcomeUnknown Outcome = "unknown" + OutcomeSuccess Outcome = "success" + OutcomeFailure Outcome = "failure" +) + +func (o Outcome) valid() bool { + return o == OutcomeUnknown || o == OutcomeSuccess || o == OutcomeFailure +} + +// Evidence is opaque, provider-neutral support for an observation. URI and +// Hash are optional and have no operational meaning to this package. +type Evidence struct { + ID ID `json:"id,omitempty"` + Kind string `json:"kind"` + Value string `json:"value,omitempty"` + URI string `json:"uri,omitempty"` + Hash string `json:"hash,omitempty"` +} + +// Budget limits work and records usage. A zero maximum means unlimited. Used +// values are monotonically increased by BudgetConsumed events. +type Budget struct { + MaxSteps uint64 `json:"max_steps,omitempty"` + MaxCost uint64 `json:"max_cost,omitempty"` + UsedSteps uint64 `json:"used_steps,omitempty"` + UsedCost uint64 `json:"used_cost,omitempty"` +} + +// DecisionKind is an instruction for an adapter/executor. Decisions are +// proposals only; applying a DecisionIssued event records the proposal and, +// for execution/cancellation, updates the corresponding lifecycle state. +type DecisionKind string + +const ( + DecisionExecute DecisionKind = "execute" + DecisionWait DecisionKind = "wait" + DecisionSucceed DecisionKind = "succeed" + DecisionFail DecisionKind = "fail" + DecisionBlock DecisionKind = "block" + DecisionCancel DecisionKind = "cancel" + DecisionExhaust DecisionKind = "exhaust" +) + +func (k DecisionKind) valid() bool { + switch k { + case DecisionExecute, DecisionWait, DecisionSucceed, DecisionFail, DecisionBlock, DecisionCancel, DecisionExhaust: + return true + default: + return false + } +} + +// Decision is a deterministic proposal produced by Reconcile. +type Decision struct { + ID ID `json:"id"` + GoalID ID `json:"goal_id"` + ObligationID ID `json:"obligation_id,omitempty"` + Kind DecisionKind `json:"kind"` + Reason string `json:"reason,omitempty"` + Cost uint64 `json:"cost,omitempty"` +} + +// EventKind identifies the union payload in Event. +type EventKind string + +const ( + EventGoalCreated EventKind = "goal_created" + EventObligationAdded EventKind = "obligation_added" + EventDecisionIssued EventKind = "decision_issued" + EventObservationAdded EventKind = "observation_added" + EventBudgetConsumed EventKind = "budget_consumed" +) + +func (k EventKind) valid() bool { + switch k { + case EventGoalCreated, EventObligationAdded, EventDecisionIssued, EventObservationAdded, EventBudgetConsumed: + return true + default: + return false + } +} + +// Event is an append-only state transition. Exactly one payload is used, +// according to Kind. Sequence is assigned by Append and is checked by Reduce; +// zero is accepted when reducing a standalone event. +type Event struct { + Sequence uint64 `json:"sequence"` + ID ID `json:"id,omitempty"` + Kind EventKind `json:"kind"` + Data []byte `json:"data,omitempty"` + Goal Goal `json:"goal,omitempty"` + Obligation Obligation `json:"obligation,omitempty"` + Decision Decision `json:"decision,omitempty"` + Observation Observation `json:"observation,omitempty"` + Budget BudgetDelta `json:"budget,omitempty"` +} + +// BudgetDelta records usage added by an event. +type BudgetDelta struct { + Steps uint64 `json:"steps,omitempty"` + Cost uint64 `json:"cost,omitempty"` +} + +// State is a snapshot of the reducer. Maps and slices are owned by the value +// returned from NewState, Reduce, and Replay; callers should not mutate them. +type State struct { + Goals map[ID]Goal `json:"goals"` + Obligations map[ID]Obligation `json:"obligations"` + Observations map[ID]Observation `json:"observations"` + Decisions []Decision `json:"decisions"` + Events []Event `json:"events"` + Budget Budget `json:"budget"` + Revision uint64 `json:"revision"` +} + +// NewState returns an empty, ready-to-use state. +func NewState() State { + return State{Goals: map[ID]Goal{}, Obligations: map[ID]Obligation{}, Observations: map[ID]Observation{}} +} + +// NewStateWithBudget returns an empty state with the supplied immutable limits. +func NewStateWithBudget(b Budget) State { + s := NewState() + s.Budget.MaxSteps, s.Budget.MaxCost = b.MaxSteps, b.MaxCost + return s +} + +// Clone returns a deep value copy suitable for callers that need to retain a +// snapshot while advancing another state. +func (s State) Clone() State { + o := NewState() + for k, v := range s.Goals { + o.Goals[k] = v + } + for k, v := range s.Obligations { + v.DependsOn = append([]ID(nil), v.DependsOn...) + o.Obligations[k] = v + } + for k, v := range s.Observations { + v.Evidence = append([]Evidence(nil), v.Evidence...) + o.Observations[k] = v + } + o.Decisions = append([]Decision(nil), s.Decisions...) + for _, e := range s.Events { + o.Events = append(o.Events, cloneEvent(e)) + } + o.Budget, o.Revision = s.Budget, s.Revision + return o +} + +func cloneEvent(e Event) Event { + e.Data = append([]byte(nil), e.Data...) + e.Obligation.DependsOn = append([]ID(nil), e.Obligation.DependsOn...) + e.Observation.Evidence = append([]Evidence(nil), e.Observation.Evidence...) + return e +} + +// ValidateID provides a small shared check for adapters creating contracts. +func ValidateID(id ID) error { + if id == "" { + return fmt.Errorf("id is required") + } + return nil +} diff --git a/v2/core/validate.go b/v2/core/validate.go new file mode 100644 index 0000000..1ea0429 --- /dev/null +++ b/v2/core/validate.go @@ -0,0 +1,102 @@ +package core + +import "fmt" + +// Validate checks all cross-object invariants in a snapshot, including DAG +// references and cycles. It is safe to call on a zero State. +func (s State) Validate() error { + for id, g := range s.Goals { + if id == "" || g.ID != id { + return fmt.Errorf("goal key/id mismatch: %q", id) + } + if err := ValidateID(g.ID); err != nil { + return fmt.Errorf("goal %q: %w", id, err) + } + if g.Status != "" && !g.Status.valid() { + return fmt.Errorf("goal %q: invalid status %q", id, g.Status) + } + } + for id, o := range s.Obligations { + if id == "" || o.ID != id { + return fmt.Errorf("obligation key/id mismatch: %q", id) + } + if err := ValidateID(o.ID); err != nil { + return fmt.Errorf("obligation %q: %w", id, err) + } + if _, ok := s.Goals[o.GoalID]; !ok { + return fmt.Errorf("obligation %q: unknown goal %q", id, o.GoalID) + } + if o.Status != "" && !o.Status.valid() { + return fmt.Errorf("obligation %q: invalid status %q", id, o.Status) + } + seen := map[ID]bool{} + for _, dep := range o.DependsOn { + if dep == o.ID { + return fmt.Errorf("obligation %q depends on itself", id) + } + if seen[dep] { + return fmt.Errorf("obligation %q repeats dependency %q", id, dep) + } + seen[dep] = true + d, ok := s.Obligations[dep] + if !ok { + return fmt.Errorf("obligation %q: unknown dependency %q", id, dep) + } + if d.GoalID != o.GoalID { + return fmt.Errorf("obligation %q: dependency %q belongs to another goal", id, dep) + } + } + } + if err := validateAcyclic(s.Obligations); err != nil { + return err + } + for id, ob := range s.Observations { + if id == "" || ob.ID != id { + return fmt.Errorf("observation key/id mismatch: %q", id) + } + if _, ok := s.Goals[ob.GoalID]; !ok { + return fmt.Errorf("observation %q: unknown goal %q", id, ob.GoalID) + } + target, ok := s.Obligations[ob.ObligationID] + if !ok || target.GoalID != ob.GoalID { + return fmt.Errorf("observation %q: invalid obligation %q", id, ob.ObligationID) + } + if !ob.Outcome.valid() { + return fmt.Errorf("observation %q: invalid outcome %q", id, ob.Outcome) + } + } + if s.Budget.MaxSteps != 0 && s.Budget.UsedSteps > s.Budget.MaxSteps { + return fmt.Errorf("budget steps exceed maximum") + } + if s.Budget.MaxCost != 0 && s.Budget.UsedCost > s.Budget.MaxCost { + return fmt.Errorf("budget cost exceeds maximum") + } + return nil +} + +func validateAcyclic(nodes map[ID]Obligation) error { + marks := make(map[ID]uint8, len(nodes)) + var visit func(ID) error + visit = func(id ID) error { + if marks[id] == 1 { + return fmt.Errorf("obligation dependency cycle includes %q", id) + } + if marks[id] == 2 { + return nil + } + marks[id] = 1 + for _, dep := range nodes[id].DependsOn { + if err := visit(dep); err != nil { + return err + } + } + marks[id] = 2 + return nil + } + for id := range nodes { + if err := visit(id); err != nil { + return err + } + } + return nil +} diff --git a/v2/graph/bridge.go b/v2/graph/bridge.go new file mode 100644 index 0000000..1922fe8 --- /dev/null +++ b/v2/graph/bridge.go @@ -0,0 +1,190 @@ +package graph + +import ( + "fmt" + + "github.com/reuben/scud/v2/core" +) + +// Status conversion is intentionally strict. Core has no deferred or waiting +// obligation states: those are reconciliation-layer states and must not be +// silently serialized as succeeded (or otherwise terminal). +func FromCoreStatus(s core.ObligationStatus) (Status, error) { + switch s { + case "", core.ObligationPending: + return Pending, nil + case core.ObligationReady: + return Ready, nil + case core.ObligationRunning: + return Running, nil + case core.ObligationSucceeded: + return Succeeded, nil + case core.ObligationFailed: + return Failed, nil + case core.ObligationBlocked: + return Blocked, nil + case core.ObligationCancelled: + return Cancelled, nil + default: + return "", fmt.Errorf("unsupported core obligation status %q", s) + } +} + +func ToCoreStatus(s Status) (core.ObligationStatus, error) { + switch s.normalized() { + case Pending: + return core.ObligationPending, nil + case Ready: + return core.ObligationReady, nil + case Running: + return core.ObligationRunning, nil + case Succeeded: + return core.ObligationSucceeded, nil + case Failed: + return core.ObligationFailed, nil + case Blocked: + return core.ObligationBlocked, nil + case Cancelled: + return core.ObligationCancelled, nil + case Deferred, Waiting: + return "", fmt.Errorf("graph status %q is reconciliation-only and has no core representation", s) + default: + return "", fmt.Errorf("unsupported graph status %q", s) + } +} + +// FromCore converts core obligations to a validated graph. GoalID and +// capability/policy references remain available to core adapters; graph owns +// only dependency and lifecycle scheduling fields. +func FromCore(obligations []core.Obligation) (Graph, error) { + nodes := make([]Node, 0, len(obligations)) + for _, o := range obligations { + status, err := FromCoreStatus(o.Status) + if err != nil { + return Graph{}, fmt.Errorf("obligation %q: %w", o.ID, err) + } + nodes = append(nodes, Node{ID: ID(o.ID), GoalID: ID(o.GoalID), Description: o.Description, CapabilityRef: o.CapabilityRef, PolicyRef: o.PolicyRef, GrantRef: o.GrantRef, Dependencies: idsFromCore(o.DependsOn), Status: status}) + } + return Build(nodes) +} + +// FromCoreObligations is an explicit alias for FromCore. +func FromCoreObligations(obligations []core.Obligation) (Graph, error) { + return FromCore(obligations) +} + +// ToCore converts a graph to obligations for one goal. It rejects +// reconciliation-only states rather than silently changing their meaning. +func ToCore(g Graph, goalID core.ID) ([]core.Obligation, error) { + if goalID == "" { + return nil, fmt.Errorf("goal ID is required") + } + if err := g.Validate(); err != nil { + return nil, err + } + out := make([]core.Obligation, 0, len(g.nodes)) + for _, n := range g.Nodes() { + status, err := ToCoreStatus(n.Status) + if err != nil { + return nil, fmt.Errorf("node %q: %w", n.ID, err) + } + var deps []core.ID + if n.Dependencies != nil { + deps = make([]core.ID, len(n.Dependencies)) + for i, dep := range n.Dependencies { + deps[i] = core.ID(dep) + } + } + out = append(out, core.Obligation{ID: core.ID(n.ID), GoalID: goalID, Description: n.Description, CapabilityRef: n.CapabilityRef, PolicyRef: n.PolicyRef, GrantRef: n.GrantRef, DependsOn: deps, Status: status}) + } + return out, nil +} + +// ToCoreObligations is an explicit alias for ToCore. +func ToCoreObligations(g Graph, goalID core.ID) ([]core.Obligation, error) { + return ToCore(g, goalID) +} + +// FromCoreEvidence preserves every field that core exposes. Evidence remains +// opaque to graph; reconciliation only checks presence when required. +func FromCoreEvidence(in []core.Evidence) []Evidence { + if in == nil { + return nil + } + out := make([]Evidence, len(in)) + for i, e := range in { + out[i] = Evidence{ID: ID(e.ID), Kind: e.Kind, Value: e.Value, URI: e.URI, Hash: e.Hash} + } + return out +} + +func ToCoreEvidence(in []Evidence) []core.Evidence { + if in == nil { + return nil + } + out := make([]core.Evidence, len(in)) + for i, e := range in { + out[i] = core.Evidence{ID: core.ID(e.ID), Kind: e.Kind, Value: e.Value, URI: e.URI, Hash: e.Hash} + } + return out +} + +// FromCoreState selects one goal's obligations and carries cumulative budget +// usage into a reconciliation snapshot. Core remains authoritative for event +// reduction; graph is a planning/reconciliation view. +func FromCoreState(s core.State, goalID core.ID) (Snapshot, error) { + if err := s.Validate(); err != nil { + return Snapshot{}, err + } + if goalID == "" { + return Snapshot{}, fmt.Errorf("goal ID is required") + } + if _, ok := s.Goals[goalID]; !ok { + return Snapshot{}, fmt.Errorf("unknown goal %q", goalID) + } + var obligations []core.Obligation + for _, o := range s.Obligations { + if o.GoalID == goalID { + obligations = append(obligations, o) + } + } + g, err := FromCore(obligations) + if err != nil { + return Snapshot{}, err + } + return Snapshot{Graph: g, Budget: Budget{UsedCost: s.Budget.UsedCost, UsedSteps: s.Budget.UsedSteps}, SeenPlans: map[string]bool{Fingerprint(g): true}}, nil +} + +// ToCoreState creates a minimal core state containing one goal and this graph. +// It is intended for adapters/tests; event history and observation history are +// not fabricated by this pure bridge. +func ToCoreState(s Snapshot, goal core.Goal) (core.State, error) { + if goal.ID == "" { + return core.State{}, fmt.Errorf("goal ID is required") + } + obligations, err := ToCore(s.Graph, goal.ID) + if err != nil { + return core.State{}, err + } + out := core.NewState() + out.Goals[goal.ID] = goal + for _, o := range obligations { + out.Obligations[o.ID] = o + } + out.Budget.UsedCost, out.Budget.UsedSteps = s.Budget.UsedCost, s.Budget.UsedSteps + if err := out.Validate(); err != nil { + return core.State{}, err + } + return out, nil +} + +func idsFromCore(in []core.ID) []ID { + if in == nil { + return nil + } + out := make([]ID, len(in)) + for i, id := range in { + out[i] = ID(id) + } + return out +} diff --git a/v2/graph/bridge_test.go b/v2/graph/bridge_test.go new file mode 100644 index 0000000..d51851c --- /dev/null +++ b/v2/graph/bridge_test.go @@ -0,0 +1,76 @@ +package graph + +import ( + "reflect" + "testing" + + "github.com/reuben/scud/v2/core" +) + +func TestCoreGraphRoundTrip(t *testing.T) { + in := []core.Obligation{ + {ID: "b", GoalID: "g", Description: "build", CapabilityRef: "builder", PolicyRef: "safe", GrantRef: "grant-1", DependsOn: []core.ID{"a"}, Status: core.ObligationReady}, + {ID: "a", GoalID: "g", Status: core.ObligationSucceeded}, + } + g, err := FromCore(in) + if err != nil { + t.Fatal(err) + } + out, err := ToCore(g, "g") + if err != nil { + t.Fatal(err) + } + if len(out) != len(in) { + t.Fatalf("round trip count = %d", len(out)) + } + for _, want := range in { + var got core.Obligation + for _, candidate := range out { + if candidate.ID == want.ID { + got = candidate + } + } + if got.ID == "" || got.GoalID != want.GoalID || got.Description != want.Description || got.CapabilityRef != want.CapabilityRef || got.PolicyRef != want.PolicyRef || got.GrantRef != want.GrantRef || got.Status != want.Status || !reflect.DeepEqual(got.DependsOn, want.DependsOn) { + t.Fatalf("round trip %q = %+v, want %+v", want.ID, got, want) + } + } +} + +func TestCoreBridgeRejectsReconciliationOnlyStatuses(t *testing.T) { + g := New(Node{ID: "a", Status: Deferred}) + if _, err := ToCore(g, "g"); err == nil { + t.Fatal("expected deferred status conversion error") + } + if _, err := FromCore([]core.Obligation{{ID: "a", GoalID: "g", Status: core.ObligationStatus("waiting")}}); err == nil { + t.Fatal("expected unknown core status error") + } +} + +func TestEvidenceRoundTrip(t *testing.T) { + in := []core.Evidence{{ID: "e", Kind: "test", Value: "ok", URI: "file://x", Hash: "abc"}} + got := ToCoreEvidence(FromCoreEvidence(in)) + if !reflect.DeepEqual(got, in) { + t.Fatalf("evidence round trip = %+v, want %+v", got, in) + } +} + +func TestCoreStateBridgeCarriesBudget(t *testing.T) { + coreState := core.NewState() + coreState.Goals["g"] = core.Goal{ID: "g", Status: core.GoalRunning} + coreState.Obligations["a"] = core.Obligation{ID: "a", GoalID: "g", Status: core.ObligationRunning} + coreState.Budget.UsedCost, coreState.Budget.UsedSteps = 7, 2 + s, err := FromCoreState(coreState, "g") + if err != nil { + t.Fatal(err) + } + if s.Budget.UsedCost != 7 || s.Budget.UsedSteps != 2 { + t.Fatalf("budget = %+v", s.Budget) + } + round, err := ToCoreState(s, core.Goal{ID: "g", Status: core.GoalRunning}) + if err != nil { + t.Fatal(err) + } + if round.Obligations["a"].Status != core.ObligationRunning || round.Budget.UsedCost != 7 { + t.Fatalf("core state = %+v", round) + } +} diff --git a/v2/graph/doc.go b/v2/graph/doc.go new file mode 100644 index 0000000..81220cf --- /dev/null +++ b/v2/graph/doc.go @@ -0,0 +1,9 @@ +// Package graph provides the SCUD v2 planning view: deterministic DAG +// validation, wave construction, graph rewrites, and pure reconciliation. +// +// The v2/core package owns canonical lifecycle state and event reduction. +// This package owns scheduling decisions and temporary reconciliation states +// (deferred and waiting). bridge.go intentionally rejects those temporary +// states when converting back to core, so adapters must make that transition +// explicit instead of silently changing its meaning. +package graph diff --git a/v2/graph/graph.go b/v2/graph/graph.go new file mode 100644 index 0000000..6793f22 --- /dev/null +++ b/v2/graph/graph.go @@ -0,0 +1,592 @@ +// Package graph contains deterministic, provider-neutral graph operations for +// SCUD v2. It intentionally owns no persistence or execution concerns. +package graph + +import ( + "fmt" + "sort" + "strings" +) + +// ID identifies a node in a graph. +type ID string + +// Status is the lifecycle state used by the reconciliation layer. A zero +// status is treated as Pending. +type Status string + +const ( + Pending Status = "pending" + Ready Status = "ready" + Running Status = "running" + Succeeded Status = "succeeded" + Failed Status = "failed" + Blocked Status = "blocked" + Cancelled Status = "cancelled" + Deferred Status = "deferred" + Waiting Status = "waiting" +) + +func (s Status) normalized() Status { + if s == "" { + return Pending + } + return s +} + +func (s Status) candidate() bool { + s = s.normalized() + return s == Pending || s == Ready || s == Deferred +} + +func (s Status) terminal() bool { + s = s.normalized() + return s == Succeeded || s == Failed || s == Blocked || s == Cancelled +} + +func (s Status) valid() bool { + s = s.normalized() + switch s { + case Pending, Ready, Running, Succeeded, Failed, Blocked, Cancelled, Deferred, Waiting: + return true + default: + return false + } +} + +// Evidence is a small provider-neutral proof attached to a node. Value can be +// a summary, digest, or opaque reference; graph does not interpret it. +type Evidence struct { + ID ID + Kind string + Value string + URI string + Hash string +} + +// Node is a graph vertex. Dependencies are directed edges from this node to +// its prerequisites. Cost is reserved when the node is admitted. +type Node struct { + ID ID + GoalID ID + Description string + CapabilityRef string + PolicyRef string + GrantRef string + Dependencies []ID + Status Status + Cost uint64 + Priority int + Progress float64 + MinProgress float64 + EvidenceRequired bool + Evidence []Evidence +} + +// Graph is an immutable-by-convention graph value. All mutating operations +// return a copy, which makes reconciliation snapshots safe to retain. +type Graph struct { + nodes map[ID]Node +} + +// New constructs a graph from nodes. For validation errors (including +// duplicate IDs), use Build; New is convenient for literals in tests and +// callers that validate immediately afterwards. +func New(nodes ...Node) Graph { + g := Graph{nodes: make(map[ID]Node, len(nodes))} + for _, n := range nodes { + n = cloneNode(n) + if n.Status == "" { + n.Status = Pending + } + g.nodes[n.ID] = n + } + return g +} + +// NewGraph is an explicit alias for New. +func NewGraph(nodes ...Node) Graph { return New(nodes...) } + +// Build validates and constructs a graph atomically. +func Build(nodes []Node) (Graph, error) { + seen := make(map[ID]struct{}, len(nodes)) + for _, n := range nodes { + if n.ID == "" { + return Graph{}, fmt.Errorf("node id is required") + } + if _, ok := seen[n.ID]; ok { + return Graph{}, fmt.Errorf("duplicate node %q", n.ID) + } + seen[n.ID] = struct{}{} + } + g := New(nodes...) + if err := g.Validate(); err != nil { + return Graph{}, err + } + return g, nil +} + +// Nodes returns all nodes in stable ID order. +func (g Graph) Nodes() []Node { + ids := g.ids() + out := make([]Node, 0, len(ids)) + for _, id := range ids { + out = append(out, cloneNode(g.nodes[id])) + } + return out +} + +// Node returns a copy of the requested node. +func (g Graph) Node(id ID) (Node, bool) { + n, ok := g.nodes[id] + if !ok { + return Node{}, false + } + return cloneNode(n), true +} + +// IDs returns all node IDs in stable order. +func (g Graph) IDs() []ID { return g.ids() } + +func (g Graph) ids() []ID { + ids := make([]ID, 0, len(g.nodes)) + for id := range g.nodes { + ids = append(ids, id) + } + sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) + return ids +} + +// Report describes every structural graph error, rather than stopping at the +// first one. Missing dependencies and cycles are sorted deterministically. +type Report struct { + EmptyIDs []ID + InvalidStatuses []StatusIssue + MissingDependencies []MissingDependency + DuplicateDependencies []DuplicateDependency + Cycles [][]ID +} + +type StatusIssue struct { + Node ID + Status Status +} + +type MissingDependency struct { + Node ID + Dependency ID +} + +type DuplicateDependency struct { + Node ID + Dependency ID +} + +func (r Report) Valid() bool { + return len(r.EmptyIDs) == 0 && len(r.InvalidStatuses) == 0 && len(r.MissingDependencies) == 0 && len(r.DuplicateDependencies) == 0 && len(r.Cycles) == 0 +} + +// Err converts a report to a compact, useful validation error. +func (r Report) Err() error { + if r.Valid() { + return nil + } + parts := make([]string, 0, 4) + if len(r.EmptyIDs) > 0 { + parts = append(parts, fmt.Sprintf("empty node IDs: %v", r.EmptyIDs)) + } + if len(r.InvalidStatuses) > 0 { + parts = append(parts, fmt.Sprintf("invalid statuses: %v", r.InvalidStatuses)) + } + if len(r.MissingDependencies) > 0 { + parts = append(parts, fmt.Sprintf("missing dependencies: %v", r.MissingDependencies)) + } + if len(r.DuplicateDependencies) > 0 { + parts = append(parts, fmt.Sprintf("duplicate dependencies: %v", r.DuplicateDependencies)) + } + if len(r.Cycles) > 0 { + parts = append(parts, fmt.Sprintf("cycles: %v", r.Cycles)) + } + return fmt.Errorf("invalid graph: %s", strings.Join(parts, "; ")) +} + +// Check returns a complete structural report. +func (g Graph) Check() Report { + r := Report{} + for _, id := range g.ids() { + if id == "" { + r.EmptyIDs = append(r.EmptyIDs, id) + } + } + for _, id := range g.ids() { + n := g.nodes[id] + if !n.Status.valid() { + r.InvalidStatuses = append(r.InvalidStatuses, StatusIssue{Node: id, Status: n.Status}) + } + seen := map[ID]bool{} + for _, dep := range n.Dependencies { + if seen[dep] { + r.DuplicateDependencies = append(r.DuplicateDependencies, DuplicateDependency{Node: id, Dependency: dep}) + } + if _, ok := g.nodes[dep]; !ok && !seen[dep] { + r.MissingDependencies = append(r.MissingDependencies, MissingDependency{Node: id, Dependency: dep}) + } + seen[dep] = true + } + } + sort.Slice(r.MissingDependencies, func(i, j int) bool { + if r.MissingDependencies[i].Node == r.MissingDependencies[j].Node { + return r.MissingDependencies[i].Dependency < r.MissingDependencies[j].Dependency + } + return r.MissingDependencies[i].Node < r.MissingDependencies[j].Node + }) + sort.Slice(r.DuplicateDependencies, func(i, j int) bool { + if r.DuplicateDependencies[i].Node == r.DuplicateDependencies[j].Node { + return r.DuplicateDependencies[i].Dependency < r.DuplicateDependencies[j].Dependency + } + return r.DuplicateDependencies[i].Node < r.DuplicateDependencies[j].Node + }) + r.Cycles = findCycles(g) + return r +} + +// Cycles returns a deterministic copy of all detected cycle paths. +func (g Graph) Cycles() [][]ID { + r := g.Check() + cycles := make([][]ID, len(r.Cycles)) + for i, cycle := range r.Cycles { + cycles[i] = append([]ID(nil), cycle...) + } + return cycles +} + +// MissingDependencies returns all missing edge references in stable order. +func (g Graph) MissingDependencies() []MissingDependency { + return append([]MissingDependency(nil), g.Check().MissingDependencies...) +} + +// Validate checks IDs, references, duplicate edges, and acyclicity. +func (g Graph) Validate() error { return g.Check().Err() } + +// Validate is a package-level convenience for callers with a graph value. +func Validate(g Graph) error { return g.Validate() } + +// CycleError is returned when wave planning encounters a cycle. +type CycleError struct{ Cycles [][]ID } + +func (e CycleError) Error() string { return fmt.Sprintf("graph contains cycle(s): %v", e.Cycles) } + +func findCycles(g Graph) [][]ID { + marks := map[ID]uint8{} + stack := []ID{} + cycles := [][]ID{} + seen := map[string]bool{} + var visit func(ID) + visit = func(id ID) { + marks[id] = 1 + stack = append(stack, id) + for _, dep := range sortedDeps(g.nodes[id].Dependencies) { + if _, ok := g.nodes[dep]; !ok { + continue + } + switch marks[dep] { + case 0: + visit(dep) + case 1: + start := 0 + for i, x := range stack { + if x == dep { + start = i + break + } + } + cycle := append([]ID(nil), stack[start:]...) + cycle = append(cycle, dep) + key := cycleKey(cycle) + if !seen[key] { + seen[key] = true + cycles = append(cycles, cycle) + } + } + } + stack = stack[:len(stack)-1] + marks[id] = 2 + } + for _, id := range g.ids() { + if marks[id] == 0 { + visit(id) + } + } + sort.Slice(cycles, func(i, j int) bool { return cycleKey(cycles[i]) < cycleKey(cycles[j]) }) + return cycles +} + +func cycleKey(c []ID) string { + parts := make([]string, len(c)) + for i, id := range c { + parts[i] = string(id) + } + return strings.Join(parts, "\x00") +} + +func sortedDeps(in []ID) []ID { + out := append([]ID(nil), in...) + sort.Slice(out, func(i, j int) bool { return out[i] < out[j] }) + return out +} + +// Ready returns candidates whose dependencies are succeeded (or listed in +// completed). The result is always lexicographically sorted. +func (g Graph) Ready(completed map[ID]bool) []ID { + done := map[ID]bool{} + for id, yes := range completed { + if yes { + done[id] = true + } + } + for _, n := range g.Nodes() { + if n.Status.normalized() == Succeeded { + done[n.ID] = true + } + } + ready := []ID{} + for _, n := range g.Nodes() { + if !n.Status.candidate() { + continue + } + ok := true + for _, dep := range n.Dependencies { + if !done[dep] { + ok = false + break + } + } + if ok { + ready = append(ready, n.ID) + } + } + return ready +} + +// ReadySet is an alias for Ready. +func (g Graph) ReadySet(completed map[ID]bool) []ID { return g.Ready(completed) } + +// Waves computes deterministic parallel waves. Nodes blocked by a running or +// terminal-unsuccessful prerequisite are omitted; structural errors are +// returned. A cycle returns a CycleError and no partial plan. +func (g Graph) Waves(completed map[ID]bool) ([][]ID, error) { + if err := g.Validate(); err != nil { + if len(g.Check().Cycles) > 0 { + return nil, CycleError{Cycles: g.Check().Cycles} + } + return nil, err + } + done := map[ID]bool{} + for id, yes := range completed { + if yes { + done[id] = true + } + } + for _, n := range g.Nodes() { + if n.Status.normalized() == Succeeded { + done[n.ID] = true + } + } + remaining := map[ID]bool{} + for _, n := range g.Nodes() { + if n.Status.candidate() && !done[n.ID] { + remaining[n.ID] = true + } + } + var waves [][]ID + for len(remaining) > 0 { + ready := make([]ID, 0) + for _, id := range g.ids() { + if !remaining[id] { + continue + } + n := g.nodes[id] + ok := true + for _, dep := range n.Dependencies { + if remaining[dep] || !done[dep] { + ok = false + break + } + } + if ok { + ready = append(ready, id) + } + } + if len(ready) == 0 { + // A non-candidate prerequisite can legitimately leave nodes out of + // the plan; only report cycles among still-plannable candidates. + cycleNodes := make([]Node, 0, len(remaining)) + for id := range remaining { + cycleNodes = append(cycleNodes, g.nodes[id]) + } + if cg, err := Build(cycleNodes); err == nil { + if c := findCycles(cg); len(c) > 0 { + return nil, CycleError{Cycles: c} + } + } + break + } + waves = append(waves, ready) + for _, id := range ready { + delete(remaining, id) + done[id] = true + } + } + return waves, nil +} + +// Plan is an alias for Waves. +func (g Graph) Plan(completed map[ID]bool) ([][]ID, error) { return g.Waves(completed) } + +// TopologicalOrder flattens Waves while retaining deterministic wave order. +func (g Graph) TopologicalOrder(completed map[ID]bool) ([]ID, error) { + waves, err := g.Waves(completed) + if err != nil { + return nil, err + } + var ids []ID + for _, wave := range waves { + ids = append(ids, wave...) + } + return ids, nil +} + +// Rewrite describes an atomic graph rewrite. AddNodes and RemoveIDs are +// applied first, followed by dependency replacements and edge additions/removals. +type Rewrite struct { + AddNodes []Node + UpdateNodes map[ID]Node + RemoveIDs []ID + ReplaceDependencies map[ID][]ID + AddDependencies map[ID][]ID + RemoveDependencies map[ID][]ID +} + +// Rewrite applies and validates a graph rewrite without mutating the source. +func (g Graph) Rewrite(r Rewrite) (Graph, error) { + next := cloneGraph(g) + for _, id := range r.RemoveIDs { + if _, ok := next.nodes[id]; !ok { + return Graph{}, fmt.Errorf("cannot remove unknown node %q", id) + } + delete(next.nodes, id) + } + for _, n := range r.AddNodes { + if n.ID == "" { + return Graph{}, fmt.Errorf("added node id is required") + } + if _, ok := next.nodes[n.ID]; ok { + return Graph{}, fmt.Errorf("node %q already exists", n.ID) + } + n.Status = n.Status.normalized() + next.nodes[n.ID] = cloneNode(n) + } + for id, n := range r.UpdateNodes { + if id == "" || n.ID != id { + return Graph{}, fmt.Errorf("updated node key/id mismatch: %q", id) + } + if _, ok := next.nodes[id]; !ok { + return Graph{}, fmt.Errorf("cannot update unknown node %q", id) + } + n.Status = n.Status.normalized() + next.nodes[id] = cloneNode(n) + } + for id, deps := range r.ReplaceDependencies { + n, ok := next.nodes[id] + if !ok { + return Graph{}, fmt.Errorf("cannot rewrite unknown node %q", id) + } + n.Dependencies = append([]ID(nil), deps...) + next.nodes[id] = n + } + for id, deps := range r.AddDependencies { + n, ok := next.nodes[id] + if !ok { + return Graph{}, fmt.Errorf("cannot add dependency to unknown node %q", id) + } + seen := map[ID]bool{} + for _, d := range n.Dependencies { + seen[d] = true + } + for _, d := range deps { + if !seen[d] { + n.Dependencies = append(n.Dependencies, d) + seen[d] = true + } + } + next.nodes[id] = n + } + for id, deps := range r.RemoveDependencies { + n, ok := next.nodes[id] + if !ok { + return Graph{}, fmt.Errorf("cannot remove dependency from unknown node %q", id) + } + remove := map[ID]bool{} + for _, d := range deps { + remove[d] = true + } + kept := n.Dependencies[:0] + for _, d := range n.Dependencies { + if !remove[d] { + kept = append(kept, d) + } + } + n.Dependencies = append([]ID(nil), kept...) + next.nodes[id] = n + } + if err := next.Validate(); err != nil { + return Graph{}, err + } + return next, nil +} + +// ApplyRewrite is a descriptive alias for Rewrite. +func (g Graph) ApplyRewrite(r Rewrite) (Graph, error) { return g.Rewrite(r) } + +// AddNode returns a graph with one node added. +func (g Graph) AddNode(n Node) (Graph, error) { return g.Rewrite(Rewrite{AddNodes: []Node{n}}) } + +// UpdateNode replaces one existing node while preserving graph validity. +func (g Graph) UpdateNode(n Node) (Graph, error) { + return g.Rewrite(Rewrite{UpdateNodes: map[ID]Node{n.ID: n}}) +} + +// RemoveNode removes a node. Dependents must be rewritten explicitly. +func (g Graph) RemoveNode(id ID) (Graph, error) { return g.Rewrite(Rewrite{RemoveIDs: []ID{id}}) } + +// SetDependencies replaces one node's dependency list. +func (g Graph) SetDependencies(id ID, deps []ID) (Graph, error) { + return g.Rewrite(Rewrite{ReplaceDependencies: map[ID][]ID{id: deps}}) +} + +// AddDependency adds one edge if it is not already present. +func (g Graph) AddDependency(id, dependency ID) (Graph, error) { + return g.Rewrite(Rewrite{AddDependencies: map[ID][]ID{id: {dependency}}}) +} + +// RemoveDependency removes one edge. +func (g Graph) RemoveDependency(id, dependency ID) (Graph, error) { + return g.Rewrite(Rewrite{RemoveDependencies: map[ID][]ID{id: {dependency}}}) +} + +func cloneGraph(g Graph) Graph { + n := Graph{nodes: make(map[ID]Node, len(g.nodes))} + for id, node := range g.nodes { + n.nodes[id] = cloneNode(node) + } + return n +} + +func cloneNode(n Node) Node { + n.Dependencies = append([]ID(nil), n.Dependencies...) + n.Evidence = append([]Evidence(nil), n.Evidence...) + if n.Status == "" { + n.Status = Pending + } + return n +} diff --git a/v2/graph/graph_test.go b/v2/graph/graph_test.go new file mode 100644 index 0000000..3076296 --- /dev/null +++ b/v2/graph/graph_test.go @@ -0,0 +1,102 @@ +package graph + +import ( + "reflect" + "testing" +) + +func TestWavesAreDeterministic(t *testing.T) { + g, err := Build([]Node{ + {ID: "z", Dependencies: []ID{"b", "a"}}, + {ID: "b"}, {ID: "a"}, {ID: "m", Dependencies: []ID{"z"}}, + }) + if err != nil { + t.Fatal(err) + } + want := [][]ID{{"a", "b"}, {"z"}, {"m"}} + for i := 0; i < 20; i++ { + got, err := g.Waves(nil) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("iteration %d: waves = %#v, want %#v", i, got, want) + } + } + if got := g.Ready(nil); !reflect.DeepEqual(got, []ID{"a", "b"}) { + t.Fatalf("ready = %v", got) + } +} + +func TestValidationReportsMissingAndCycles(t *testing.T) { + g := New( + Node{ID: "a", Dependencies: []ID{"missing", "missing"}}, + Node{ID: "b", Dependencies: []ID{"c"}}, + Node{ID: "c", Dependencies: []ID{"b"}}, + ) + r := g.Check() + if r.Valid() || len(r.MissingDependencies) != 1 || len(r.DuplicateDependencies) != 1 || len(r.Cycles) != 1 { + t.Fatalf("unexpected report: %#v", r) + } + if _, ok := g.Waves(nil); ok == nil { + t.Fatal("expected validation error") + } +} + +func TestRewriteIsAtomicAndGuardsCycles(t *testing.T) { + g, err := Build([]Node{{ID: "a"}, {ID: "b", Dependencies: []ID{"a"}}}) + if err != nil { + t.Fatal(err) + } + if _, err := g.AddDependency("a", "b"); err == nil { + t.Fatal("expected cycle error") + } + if got := g.Ready(nil); !reflect.DeepEqual(got, []ID{"a"}) { + t.Fatalf("source graph mutated after rejected rewrite: %v", got) + } + next, err := g.Rewrite(Rewrite{ + AddNodes: []Node{{ID: "c"}}, + AddDependencies: map[ID][]ID{"c": {"b"}}, + RemoveDependencies: map[ID][]ID{"b": {"a"}}, + }) + if err != nil { + t.Fatal(err) + } + if got := next.Ready(nil); !reflect.DeepEqual(got, []ID{"a", "b"}) { + t.Fatalf("rewritten ready = %v", got) + } +} + +func TestRewriteUpdatesNodeMetadata(t *testing.T) { + g, err := Build([]Node{{ID: "a", Description: "old"}}) + if err != nil { + t.Fatal(err) + } + n, _ := g.Node("a") + n.Description, n.Status = "new", Running + next, err := g.UpdateNode(n) + if err != nil { + t.Fatal(err) + } + got, _ := next.Node("a") + if got.Description != "new" || got.Status != Running { + t.Fatalf("updated node = %+v", got) + } +} + +func TestCompletedSetAndTerminalStatuses(t *testing.T) { + g, err := Build([]Node{{ID: "a", Status: Succeeded}, {ID: "b", Dependencies: []ID{"a"}}, {ID: "c", Status: Blocked}}) + if err != nil { + t.Fatal(err) + } + waves, err := g.Waves(nil) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(waves, [][]ID{{"b"}}) { + t.Fatalf("waves = %v", waves) + } + if got := g.Ready(map[ID]bool{"a": true}); !reflect.DeepEqual(got, []ID{"b"}) { + t.Fatalf("ready = %v", got) + } +} diff --git a/v2/graph/reconcile.go b/v2/graph/reconcile.go new file mode 100644 index 0000000..69fa98c --- /dev/null +++ b/v2/graph/reconcile.go @@ -0,0 +1,386 @@ +package graph + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "sort" + "strings" +) + +// Action is a provider-neutral reconciliation request/decision. +type Action string + +const ( + Admit Action = "admit" + Defer Action = "defer" + Ask Action = "ask" + Cancel Action = "cancel" + Replan Action = "replan" + Complete Action = "complete" + Block Action = "block" + + ActionAdmit = Admit + ActionDefer = Defer + ActionAsk = Ask + ActionCancel = Cancel + ActionReplan = Replan + ActionComplete = Complete + ActionBlock = Block +) + +// Policy controls pure backpressure and feedback-loop guards. A zero limit +// means unlimited, except MaxActive where zero means unlimited as well. +type Policy struct { + MaxActive int + MaxCost uint64 + MaxSteps uint64 + MaxReplans int + RequireProgressForComplete bool +} + +// Budget records cumulative reservations. Admission consumes cost and steps; +// cancellation does not refund them, preserving a conservative budget. +type Budget struct { + UsedCost uint64 + UsedSteps uint64 +} + +// Snapshot is the complete input/output of reconciliation. It contains no +// handles to providers and can be copied or replayed safely. +type Snapshot struct { + Graph Graph + Budget Budget + Replans int + SeenPlans map[string]bool + Questions []string +} + +// State is an alias useful to callers that model reconciliation as a state +// machine. +type State = Snapshot + +// NewSnapshot returns a normalized snapshot and records the initial graph +// fingerprint, preventing a no-op replan from creating a feedback loop. +func NewSnapshot(g Graph) Snapshot { + s := Snapshot{Graph: cloneGraph(g), SeenPlans: map[string]bool{}} + s.SeenPlans[Fingerprint(g)] = true + return s +} + +// NewState is an alias for NewSnapshot. +func NewState(g Graph) Snapshot { return NewSnapshot(g) } + +// Clone makes all nested graph and slice data independent. +func (s Snapshot) Clone() Snapshot { + o := Snapshot{ + Graph: cloneGraph(s.Graph), Budget: s.Budget, Replans: s.Replans, + SeenPlans: map[string]bool{}, Questions: append([]string(nil), s.Questions...), + } + for p, yes := range s.SeenPlans { + o.SeenPlans[p] = yes + } + return o +} + +// Request asks reconciliation to perform one guarded lifecycle transition. +// ID is required for every action except Replan of an entire graph. +type Request struct { + Action Action + ID ID + Reason string + Progress float64 + Evidence []Evidence + Question string + Rewrite *Rewrite +} + +// Event is an alias for Request for event-oriented adapters. +type Event = Request + +// Decision is the deterministic result for an adapter/executor. +type Decision struct { + Action Action + ID ID + Reason string +} + +// Result carries both the new immutable snapshot and the proposal. +type Result struct { + State Snapshot + Decision Decision +} + +var ( + ErrUnknownNode = errors.New("unknown node") + ErrInvalidAction = errors.New("invalid action") + ErrFeedbackLoop = errors.New("replan feedback loop guarded") + ErrReplanLimit = errors.New("replan limit reached") +) + +// Reconcile applies one request without mutating the input snapshot. Budget +// exhaustion and incomplete evidence are represented as defer/ask decisions, +// not provider-specific errors. +func Reconcile(s Snapshot, req Request, p Policy) (Result, error) { + if err := s.Graph.Validate(); err != nil { + return Result{}, err + } + n := s.Clone() + if n.SeenPlans == nil { + n.SeenPlans = map[string]bool{Fingerprint(n.Graph): true} + } + if req.Action == "" { + return Result{}, ErrInvalidAction + } + if req.Action != Replan { + if req.ID == "" { + return Result{}, fmt.Errorf("%w: empty ID", ErrUnknownNode) + } + if _, ok := n.Graph.Node(req.ID); !ok { + return Result{}, fmt.Errorf("%w %q", ErrUnknownNode, req.ID) + } + } + decision := Decision{Action: req.Action, ID: req.ID, Reason: req.Reason} + + switch req.Action { + case Admit: + node, _ := n.Graph.Node(req.ID) + if node.Status == Running { + return Result{State: n, Decision: decision}, nil + } + if node.Status.terminal() { + return Result{}, fmt.Errorf("cannot admit terminal node %q", req.ID) + } + // An explicit admission resolves a prior clarification request. + if node.Status == Waiting { + node.Status = Pending + setNode(&n, node) + } + if !contains(n.Graph.Ready(nil), req.ID) { + node.Status = Deferred + setNode(&n, node) + decision.Action, decision.Reason = Defer, "dependencies are not complete" + return Result{State: n, Decision: decision}, nil + } + active, _ := activeUsage(n.Graph) + if p.MaxActive > 0 && active >= p.MaxActive { + node.Status = Deferred + setNode(&n, node) + decision.Action, decision.Reason = Defer, "active-task budget exhausted" + return Result{State: n, Decision: decision}, nil + } + if p.MaxCost > 0 && (n.Budget.UsedCost > p.MaxCost || node.Cost > p.MaxCost-n.Budget.UsedCost) { + node.Status = Deferred + setNode(&n, node) + decision.Action, decision.Reason = Defer, "cost budget exhausted" + return Result{State: n, Decision: decision}, nil + } + if p.MaxSteps > 0 && n.Budget.UsedSteps >= p.MaxSteps { + node.Status = Deferred + setNode(&n, node) + decision.Action, decision.Reason = Defer, "step budget exhausted" + return Result{State: n, Decision: decision}, nil + } + node.Status = Running + n.Budget.UsedCost += node.Cost + n.Budget.UsedSteps++ + setNode(&n, node) + return Result{State: n, Decision: decision}, nil + + case Defer: + node, _ := n.Graph.Node(req.ID) + if node.Status.terminal() { + return Result{}, fmt.Errorf("cannot defer terminal node %q", req.ID) + } + node.Status = Deferred + setNode(&n, node) + decision.Reason = defaultReason(req.Reason, "deferred by policy") + return Result{State: n, Decision: decision}, nil + + case Ask: + node, _ := n.Graph.Node(req.ID) + if node.Status.terminal() { + return Result{}, fmt.Errorf("cannot ask about terminal node %q", req.ID) + } + node.Status = Waiting + setNode(&n, node) + if req.Question != "" { + n.Questions = append(n.Questions, req.Question) + } + decision.Reason = defaultReason(req.Reason, req.Question) + if decision.Reason == "" { + decision.Reason = "clarification required" + } + return Result{State: n, Decision: decision}, nil + + case Cancel: + node, _ := n.Graph.Node(req.ID) + if node.Status == Succeeded { + return Result{}, fmt.Errorf("cannot cancel succeeded node %q", req.ID) + } + node.Status = Cancelled + setNode(&n, node) + decision.Reason = defaultReason(req.Reason, "cancelled") + return Result{State: n, Decision: decision}, nil + + case Block: + node, _ := n.Graph.Node(req.ID) + if node.Status == Succeeded || node.Status == Cancelled { + return Result{}, fmt.Errorf("cannot block terminal node %q", req.ID) + } + node.Status = Blocked + setNode(&n, node) + decision.Reason = defaultReason(req.Reason, "blocked") + return Result{State: n, Decision: decision}, nil + + case Complete: + node, _ := n.Graph.Node(req.ID) + if node.Status != Running { + return Result{}, fmt.Errorf("cannot complete node %q in status %q", req.ID, node.Status) + } + progress := req.Progress + if progress < node.Progress { + progress = node.Progress + } + minimum := node.MinProgress + if minimum == 0 { + minimum = 1 + } + if (p.RequireProgressForComplete || node.MinProgress > 0) && progress < minimum { + decision.Action, decision.Reason = Ask, "completion requires additional progress" + return Result{State: n, Decision: decision}, nil + } + if node.EvidenceRequired && len(req.Evidence) == 0 && len(node.Evidence) == 0 { + decision.Action, decision.Reason = Ask, "completion requires evidence" + return Result{State: n, Decision: decision}, nil + } + node.Progress = progress + node.Evidence = append(node.Evidence, req.Evidence...) + node.Status = Succeeded + setNode(&n, node) + return Result{State: n, Decision: decision}, nil + + case Replan: + if req.Rewrite == nil { + return Result{}, fmt.Errorf("replan requires a rewrite") + } + if p.MaxReplans > 0 && n.Replans >= p.MaxReplans { + return guardedReplanBlock(n, req, decision, ErrReplanLimit) + } + g, err := n.Graph.Rewrite(*req.Rewrite) + if err != nil { + return Result{}, err + } + fingerprint := Fingerprint(g) + if n.SeenPlans[fingerprint] { + return guardedReplanBlock(n, req, decision, ErrFeedbackLoop) + } + n.Graph = g + n.Replans++ + n.SeenPlans[fingerprint] = true + decision.Reason = defaultReason(req.Reason, "graph replanned") + return Result{State: n, Decision: decision}, nil + + default: + return Result{}, fmt.Errorf("%w %q", ErrInvalidAction, req.Action) + } +} + +// Apply is an alias for Reconcile. +func Apply(s Snapshot, req Request, p Policy) (Result, error) { + return Reconcile(s, req, p) +} + +func guardedReplanBlock(s Snapshot, req Request, d Decision, cause error) (Result, error) { + if req.ID != "" { + if node, ok := s.Graph.Node(req.ID); ok && !node.Status.terminal() { + node.Status = Blocked + setNode(&s, node) + d.Action, d.Reason = Block, cause.Error() + return Result{State: s, Decision: d}, nil + } + } + return Result{State: s, Decision: Decision{Action: Block, ID: req.ID, Reason: cause.Error()}}, nil +} + +func activeUsage(g Graph) (int, uint64) { + count := 0 + var cost uint64 + for _, n := range g.Nodes() { + if n.Status == Running { + count++ + cost += n.Cost + } + } + return count, cost +} + +func setNode(s *Snapshot, n Node) { + // Graph is deliberately encapsulated; replacing through Rewrite preserves + // validation and copy semantics. + g, err := s.Graph.Rewrite(Rewrite{ReplaceDependencies: map[ID][]ID{n.ID: n.Dependencies}}) + if err != nil { + return + } + // Rewrite only replaces edges, so overlay lifecycle fields from n. + clone := cloneGraph(g) + clone.nodes[n.ID] = cloneNode(n) + s.Graph = clone +} + +func contains(ids []ID, id ID) bool { + for _, x := range ids { + if x == id { + return true + } + } + return false +} + +func defaultReason(got, fallback string) string { + if got != "" { + return got + } + return fallback +} + +// Fingerprint returns a stable SHA-256 identity for graph structure. Lifecycle +// status is intentionally omitted so status-only feedback cannot evade the +// replan loop guard. +func Fingerprint(g Graph) string { + var b strings.Builder + for _, n := range g.Nodes() { + b.WriteString(string(n.ID)) + b.WriteByte('|') + b.WriteString(string(n.GoalID)) + b.WriteByte('|') + b.WriteString(n.Description) + b.WriteByte('|') + b.WriteString(n.CapabilityRef) + b.WriteByte('|') + b.WriteString(n.PolicyRef) + b.WriteByte('|') + b.WriteString(n.GrantRef) + b.WriteByte('|') + for _, d := range sortedDeps(n.Dependencies) { + b.WriteString(string(d)) + b.WriteByte(',') + } + b.WriteByte(';') + } + h := sha256.Sum256([]byte(b.String())) + return hex.EncodeToString(h[:]) +} + +// Active returns running node IDs in stable order. +func (s Snapshot) Active() []ID { + var ids []ID + for _, n := range s.Graph.Nodes() { + if n.Status == Running { + ids = append(ids, n.ID) + } + } + sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) + return ids +} diff --git a/v2/graph/reconcile_test.go b/v2/graph/reconcile_test.go new file mode 100644 index 0000000..885f978 --- /dev/null +++ b/v2/graph/reconcile_test.go @@ -0,0 +1,83 @@ +package graph + +import "testing" + +func mustGraph(t *testing.T, nodes ...Node) Graph { + t.Helper() + g, err := Build(nodes) + if err != nil { + t.Fatal(err) + } + return g +} + +func TestReconcileBudgetDefersAndPreservesInput(t *testing.T) { + g := mustGraph(t, Node{ID: "a", Cost: 3}, Node{ID: "b", Cost: 3}) + s := NewSnapshot(g) + res, err := Reconcile(s, Request{Action: Admit, ID: "a"}, Policy{MaxActive: 1, MaxCost: 3}) + if err != nil || res.Decision.Action != Admit { + t.Fatalf("admit: decision=%+v err=%v", res.Decision, err) + } + if res.State.Budget.UsedCost != 3 || res.State.Budget.UsedSteps != 1 { + t.Fatalf("budget = %+v", res.State.Budget) + } + // The original snapshot remains pending, demonstrating pure semantics. + if n, _ := s.Graph.Node("a"); n.Status != Pending { + t.Fatalf("input mutated: %q", n.Status) + } + res, err = Reconcile(res.State, Request{Action: Admit, ID: "b"}, Policy{MaxActive: 1, MaxCost: 3}) + if err != nil || res.Decision.Action != Defer { + t.Fatalf("expected defer, decision=%+v err=%v", res.Decision, err) + } + if n, _ := res.State.Graph.Node("b"); n.Status != Deferred { + t.Fatalf("b status = %q", n.Status) + } +} + +func TestReconcileProgressAndEvidenceGuards(t *testing.T) { + g := mustGraph(t, Node{ID: "a", EvidenceRequired: true, MinProgress: .8}) + s := NewSnapshot(g) + r, err := Reconcile(s, Request{Action: Admit, ID: "a"}, Policy{}) + if err != nil { + t.Fatal(err) + } + r, err = Reconcile(r.State, Request{Action: Complete, ID: "a", Progress: .7}, Policy{RequireProgressForComplete: true}) + if err != nil || r.Decision.Action != Ask { + t.Fatalf("progress guard: %+v err=%v", r.Decision, err) + } + r, err = Reconcile(r.State, Request{Action: Complete, ID: "a", Progress: .9}, Policy{RequireProgressForComplete: true}) + if err != nil || r.Decision.Action != Ask { + t.Fatalf("evidence guard: %+v err=%v", r.Decision, err) + } + r, err = Reconcile(r.State, Request{Action: Complete, ID: "a", Progress: .9, Evidence: []Evidence{{Kind: "test", Value: "passed"}}}, Policy{RequireProgressForComplete: true}) + if err != nil || r.Decision.Action != Complete { + t.Fatalf("complete: %+v err=%v", r.Decision, err) + } + if n, _ := r.State.Graph.Node("a"); n.Status != Succeeded || n.Progress != .9 || len(n.Evidence) != 1 { + t.Fatalf("completed node = %+v", n) + } +} + +func TestReconcileActionsAndGuardedFeedback(t *testing.T) { + g := mustGraph(t, Node{ID: "a"}) + s := NewSnapshot(g) + r, err := Reconcile(s, Request{Action: Ask, ID: "a", Question: "which target?"}, Policy{}) + if err != nil || r.Decision.Action != Ask || len(r.State.Questions) != 1 { + t.Fatalf("ask: %+v %+v", r.Decision, r.State.Questions) + } + r, err = Reconcile(r.State, Request{Action: Admit, ID: "a"}, Policy{}) + if err != nil || r.Decision.Action != Admit { + t.Fatalf("admit after ask: %+v err=%v", r.Decision, err) + } + // First replan changes the graph; replaying it is blocked instead of + // recursively feeding the same plan back into itself. + rewrite := &Rewrite{AddNodes: []Node{{ID: "b"}}} + r, err = Reconcile(r.State, Request{Action: Replan, Rewrite: rewrite}, Policy{MaxReplans: 2}) + if err != nil || r.Decision.Action != Replan { + t.Fatalf("replan: %+v err=%v", r.Decision, err) + } + r, err = Reconcile(r.State, Request{Action: Replan, Rewrite: &Rewrite{RemoveIDs: []ID{"b"}}}, Policy{MaxReplans: 2}) + if err != nil || r.Decision.Action != Block { + t.Fatalf("feedback guard: %+v err=%v", r.Decision, err) + } +} diff --git a/v2/runtime/doc.go b/v2/runtime/doc.go new file mode 100644 index 0000000..ba80ebf --- /dev/null +++ b/v2/runtime/doc.go @@ -0,0 +1,10 @@ +// Package runtime coordinates the provider-independent SCUD v2 kernel with +// adapter seams. It owns no provider SDK, persistence implementation, prompt +// format, or command-line behavior. +// +// Runtime persists core events as opaque JSON payloads in an adapters.EventStore +// and replays only those canonical events. Runner progress is copied to the +// same store as non-canonical envelopes and never affects core state. Step is +// deterministic up to adapter responses: it selects the next ready obligation, +// authorizes it, reserves budget, runs it, and records an observation. +package runtime diff --git a/v2/runtime/runtime.go b/v2/runtime/runtime.go new file mode 100644 index 0000000..e4a0f60 --- /dev/null +++ b/v2/runtime/runtime.go @@ -0,0 +1,416 @@ +package runtime + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "sync" + + "github.com/reuben/scud/v2/adapters" + "github.com/reuben/scud/v2/core" + "github.com/reuben/scud/v2/graph" +) + +var ( + ErrInvalidConfig = errors.New("invalid runtime config") + ErrNoWork = errors.New("no runnable work") + ErrStoreRequired = errors.New("event store is required") +) + +// Config describes one provider-blind run. Model, prompts, working directory, +// and context are passed through to Runner without interpretation. Costs are +// opaque scheduling estimates used only for deterministic budget accounting. +type Config struct { + RunID string + Goal core.Goal + Obligations []core.Obligation + Budget core.Budget + Costs map[core.ID]uint64 + EvidenceRequired map[core.ID]bool + Model string + WorkingDir string + SystemPrompt string + Limits adapters.Limits + Context map[string]any + Backpressure graph.Policy + Runner adapters.Runner + Policy adapters.Policy + Store adapters.EventStore +} + +// Runtime is a serialized coordinator. Step is safe for concurrent callers; +// adapter execution occurs while holding the run lock so two callers cannot +// admit the same obligation. +type Runtime struct { + mu sync.Mutex + cfg Config + state core.State + stream uint64 + ready bool +} + +// New validates configuration and creates an uninitialized runtime. Call +// Replay (or Step, which calls it lazily) before inspecting State. +func New(cfg Config) (*Runtime, error) { + if cfg.RunID == "" || cfg.Goal.ID == "" { + return nil, fmt.Errorf("%w: run ID and goal ID are required", ErrInvalidConfig) + } + if cfg.Runner == nil { + return nil, fmt.Errorf("%w: runner is required", ErrInvalidConfig) + } + if cfg.Store == nil { + return nil, ErrStoreRequired + } + cfg.Obligations = append([]core.Obligation(nil), cfg.Obligations...) + cfg.Costs = cloneCosts(cfg.Costs) + cfg.EvidenceRequired = cloneBools(cfg.EvidenceRequired) + cfg.Context = cloneContext(cfg.Context) + return &Runtime{cfg: cfg, state: core.NewStateWithBudget(cfg.Budget)}, nil +} + +// State returns an independent state snapshot. +func (r *Runtime) State() core.State { r.mu.Lock(); defer r.mu.Unlock(); return r.state.Clone() } + +// Replay loads the canonical core event stream. If the store is empty, the +// configured goal and DAG are initialized as canonical events. +func (r *Runtime) Replay(ctx context.Context) error { + r.mu.Lock() + defer r.mu.Unlock() + return r.replayLocked(ctx) +} + +func (r *Runtime) replayLocked(ctx context.Context) error { + events, err := r.cfg.Store.List(ctx, r.cfg.RunID, 0) + if err != nil { + return err + } + var canonical []core.Event + for _, e := range events { + if e.Sequence > r.stream { + r.stream = e.Sequence + } + if !strings.HasPrefix(e.Type, "core/") { + continue + } + var ce core.Event + if err := json.Unmarshal(e.Data, &ce); err != nil { + return fmt.Errorf("decode canonical event %d: %w", e.Sequence, err) + } + canonical = append(canonical, ce) + } + if len(canonical) == 0 { + r.state = core.NewStateWithBudget(r.cfg.Budget) + if err := r.initializeLocked(ctx); err != nil { + return err + } + } else { + s, err := core.Replay(canonical) + if err != nil { + return err + } + // Limits are run configuration, while usage is canonical event state. + // Reapply limits after replay so a resumed run cannot silently become + // unbounded merely because limits are not repeated in every event. + s.Budget.MaxSteps, s.Budget.MaxCost = r.cfg.Budget.MaxSteps, r.cfg.Budget.MaxCost + if err := s.Validate(); err != nil { + return err + } + r.state = s + } + r.ready = true + return nil +} + +func (r *Runtime) initializeLocked(ctx context.Context) error { + // Validate the complete plan before writing its first event; otherwise a + // malformed plan could leave a durable, partially initialized stream. + nodes := make([]graph.Node, 0, len(r.cfg.Obligations)) + for _, o := range r.cfg.Obligations { + if o.GoalID != r.cfg.Goal.ID { + return fmt.Errorf("%w: obligation %q references goal %q", ErrInvalidConfig, o.ID, o.GoalID) + } + nodes = append(nodes, graph.Node{ID: graph.ID(o.ID), Dependencies: ids(o.DependsOn), Status: graphStatus(o.Status)}) + } + if _, err := graph.Build(nodes); err != nil { + return fmt.Errorf("%w: %v", ErrInvalidConfig, err) + } + if _, err := r.appendCoreLocked(ctx, core.Event{ID: core.ID("goal:" + string(r.cfg.Goal.ID)), Kind: core.EventGoalCreated, Goal: r.cfg.Goal}); err != nil { + return err + } + byID := make(map[core.ID]core.Obligation, len(r.cfg.Obligations)) + for _, o := range r.cfg.Obligations { + if _, exists := byID[o.ID]; exists { + return fmt.Errorf("%w: duplicate obligation %q", ErrInvalidConfig, o.ID) + } + byID[o.ID] = o + } + ids := make([]core.ID, 0, len(byID)) + for id := range byID { + ids = append(ids, id) + } + sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) + added := map[core.ID]bool{} + for len(added) < len(ids) { + progress := false + for _, id := range ids { + if added[id] { + continue + } + o := byID[id] + depsReady := true + for _, dep := range o.DependsOn { + if !added[dep] { + depsReady = false + break + } + } + if !depsReady { + continue + } + if _, err := r.appendCoreLocked(ctx, core.Event{ID: core.ID("obligation:" + string(o.ID)), Kind: core.EventObligationAdded, Obligation: o}); err != nil { + return err + } + added[id], progress = true, true + } + if !progress { + return fmt.Errorf("%w: invalid obligation DAG", ErrInvalidConfig) + } + } + return nil +} + +// Step performs at most one bounded runner execution. A wait, block, or +// terminal decision is returned without invoking Runner. +func (r *Runtime) Step(ctx context.Context) (core.Decision, error) { + r.mu.Lock() + defer r.mu.Unlock() + if !r.ready { + if err := r.replayLocked(ctx); err != nil { + return core.Decision{}, err + } + } + decision, err := core.Reconcile(r.state) + if err != nil { + return core.Decision{}, err + } + if decision.Kind != core.DecisionExecute { + if decision.Kind != core.DecisionWait { + if _, err := r.appendCoreLocked(ctx, core.Event{ID: decision.ID, Kind: core.EventDecisionIssued, Decision: decision}); err != nil { + return core.Decision{}, err + } + } + return decision, nil + } + o := r.state.Obligations[decision.ObligationID] + if !r.allowedByGraph(o) { + return core.Decision{ID: decision.ID, GoalID: decision.GoalID, ObligationID: decision.ObligationID, Kind: core.DecisionWait, Reason: "backpressure deferred execution"}, nil + } + if r.cfg.Policy != nil { + input := adapters.PolicyInput{RunID: r.cfg.RunID, Action: string(core.DecisionExecute), Resource: string(o.ID), Attributes: map[string]string{"goal_id": string(o.GoalID), "capability_ref": o.CapabilityRef, "policy_ref": o.PolicyRef, "grant_ref": o.GrantRef}} + allowed, err := r.cfg.Policy.Authorize(ctx, input) + if err != nil { + return core.Decision{}, err + } + if !allowed.Allowed { + blocked := core.Decision{ID: decision.ID + ":blocked", GoalID: decision.GoalID, ObligationID: decision.ObligationID, Kind: core.DecisionBlock, Reason: defaultReason(allowed.Reason, "authorization denied")} + if _, err := r.appendCoreLocked(ctx, core.Event{ID: blocked.ID, Kind: core.EventDecisionIssued, Decision: blocked}); err != nil { + return core.Decision{}, err + } + return blocked, nil + } + } + cost := r.cfg.Costs[o.ID] + if !r.budgetAllows(cost) { + exhaust := core.Decision{ID: decision.ID + ":exhausted", GoalID: decision.GoalID, ObligationID: decision.ObligationID, Kind: core.DecisionExhaust, Reason: "budget exhausted"} + // Exhaustion is canonical state, not a transient scheduler answer. Record + // it so Run terminates and replay cannot silently retry over the limit. + exhaust.ObligationID = "" + if _, err := r.appendCoreLocked(ctx, core.Event{ID: exhaust.ID, Kind: core.EventDecisionIssued, Decision: exhaust}); err != nil { + return core.Decision{}, err + } + return exhaust, nil + } + decision.Cost = cost + if _, err := r.appendCoreLocked(ctx, core.Event{ID: decision.ID, Kind: core.EventDecisionIssued, Decision: decision}); err != nil { + return core.Decision{}, err + } + if _, err := r.appendCoreLocked(ctx, core.Event{ID: decision.ID + ":budget", Kind: core.EventBudgetConsumed, Budget: core.BudgetDelta{Steps: 1, Cost: cost}}); err != nil { + return core.Decision{}, err + } + runCtx, cancel := context.WithCancel(ctx) + defer cancel() + var progressErr error + result, runErr := r.cfg.Runner.Run(runCtx, adapters.RunRequest{RunID: r.cfg.RunID, Prompt: o.Description, SystemPrompt: r.cfg.SystemPrompt, Model: r.cfg.Model, WorkingDir: r.cfg.WorkingDir, Limits: r.cfg.Limits, Context: cloneContext(r.cfg.Context)}, func(e adapters.Event) { + if progressErr != nil { + return + } + if err := r.appendProgressLocked(ctx, e); err != nil { + progressErr = err + cancel() + } + }) + if progressErr != nil { + return core.Decision{}, fmt.Errorf("persist runner progress: %w", progressErr) + } + observation := observationFromResult(o, result, runErr) + if observation.Outcome == core.OutcomeSuccess && r.cfg.EvidenceRequired[o.ID] && len(observation.Evidence) == 0 { + observation.Outcome = core.OutcomeFailure + observation.Detail = "completion requires evidence" + } + if _, err := r.appendCoreLocked(ctx, core.Event{ID: observation.ID, Kind: core.EventObservationAdded, Observation: observation}); err != nil { + return core.Decision{}, err + } + return decision, nil +} + +// Run advances until the selected goal is terminal or reconciliation waits +// for an external observation. A wait is not an error and can be resumed by +// calling Run or Step after new events are appended. +func (r *Runtime) Run(ctx context.Context) error { + for { + if err := ctx.Err(); err != nil { + return err + } + d, err := r.Step(ctx) + if err != nil { + return err + } + if d.Kind == core.DecisionWait || r.Terminal() { + return nil + } + } +} + +// Terminal reports whether the configured goal is terminal. +func (r *Runtime) Terminal() bool { + r.mu.Lock() + defer r.mu.Unlock() + g, ok := r.state.Goals[r.cfg.Goal.ID] + return ok && g.Status.IsTerminal() +} + +func (r *Runtime) appendCoreLocked(ctx context.Context, event core.Event) (core.Event, error) { + n, err := r.state.Append(event) + if err != nil { + return core.Event{}, err + } + canonical := n.Events[len(n.Events)-1] + b, err := json.Marshal(canonical) + if err != nil { + return core.Event{}, err + } + r.stream++ + if err := r.cfg.Store.Append(ctx, adapters.Event{RunID: r.cfg.RunID, Sequence: r.stream, Type: "core/" + string(canonical.Kind), Data: b}); err != nil { + r.stream-- + return core.Event{}, err + } + r.state = n + return canonical, nil +} + +func (r *Runtime) appendProgressLocked(ctx context.Context, event adapters.Event) error { + r.stream++ + e := adapters.Event{RunID: r.cfg.RunID, Sequence: r.stream, Time: event.Time, Type: "progress/" + event.Type, Data: append([]byte(nil), event.Data...)} + if err := r.cfg.Store.Append(ctx, e); err != nil { + r.stream-- + return err + } + return nil +} + +func (r *Runtime) budgetAllows(cost uint64) bool { + b := r.state.Budget + if b.MaxSteps != 0 && b.UsedSteps >= b.MaxSteps { + return false + } + return b.MaxCost == 0 || cost <= b.MaxCost-b.UsedCost +} + +func (r *Runtime) allowedByGraph(o core.Obligation) bool { + nodes := make([]graph.Node, 0, len(r.state.Obligations)) + for _, x := range r.state.Obligations { + nodes = append(nodes, graph.Node{ID: graph.ID(x.ID), Dependencies: ids(x.DependsOn), Status: graphStatus(x.Status), Cost: r.cfg.Costs[x.ID], EvidenceRequired: r.cfg.EvidenceRequired[x.ID]}) + } + g, err := graph.Build(nodes) + if err != nil { + return false + } + s := graph.NewSnapshot(g) + res, err := graph.Reconcile(s, graph.Request{Action: graph.Admit, ID: graph.ID(o.ID)}, r.cfg.Backpressure) + return err == nil && res.Decision.Action == graph.Admit +} + +func observationFromResult(o core.Obligation, result adapters.RunResult, runErr error) core.Observation { + outcome := core.OutcomeSuccess + detail := result.Text + if runErr != nil { + outcome, detail = core.OutcomeFailure, runErr.Error() + } + if result.Failure != nil { + outcome, detail = core.OutcomeFailure, result.Failure.Message + } + if result.Outcome != "" && !strings.EqualFold(result.Outcome, "success") && !strings.EqualFold(result.Outcome, "completed") { + outcome = core.OutcomeFailure + } + obs := core.Observation{ID: core.ID("observation:" + string(o.ID) + ":" + string(outcome)), GoalID: o.GoalID, ObligationID: o.ID, Outcome: outcome, Detail: detail} + if result.Text != "" { + obs.Evidence = []core.Evidence{{Kind: "runner_output", Value: result.Text}} + } + return obs +} + +func graphStatus(s core.ObligationStatus) graph.Status { + switch s { + case core.ObligationRunning: + return graph.Running + case core.ObligationSucceeded: + return graph.Succeeded + case core.ObligationFailed: + return graph.Failed + case core.ObligationBlocked: + return graph.Blocked + case core.ObligationCancelled: + return graph.Cancelled + case core.ObligationReady: + return graph.Ready + default: + return graph.Pending + } +} +func ids(in []core.ID) []graph.ID { + out := make([]graph.ID, len(in)) + for i, id := range in { + out[i] = graph.ID(id) + } + return out +} +func cloneCosts(in map[core.ID]uint64) map[core.ID]uint64 { + out := map[core.ID]uint64{} + for k, v := range in { + out[k] = v + } + return out +} +func cloneBools(in map[core.ID]bool) map[core.ID]bool { + out := map[core.ID]bool{} + for k, v := range in { + out[k] = v + } + return out +} +func cloneContext(in map[string]any) map[string]any { + out := map[string]any{} + for k, v := range in { + out[k] = v + } + return out +} +func defaultReason(given, fallback string) string { + if given != "" { + return given + } + return fallback +} diff --git a/v2/runtime/runtime_test.go b/v2/runtime/runtime_test.go new file mode 100644 index 0000000..c5f8ca5 --- /dev/null +++ b/v2/runtime/runtime_test.go @@ -0,0 +1,158 @@ +package runtime + +import ( + "context" + "errors" + "strings" + "testing" + + "github.com/reuben/scud/v2/adapters" + "github.com/reuben/scud/v2/adapters/memory" + "github.com/reuben/scud/v2/core" +) + +type fakeRunner struct { + calls int + outcome string + err error +} + +func (f *fakeRunner) Run(_ context.Context, req adapters.RunRequest, sink adapters.EventSink) (adapters.RunResult, error) { + f.calls++ + sink(adapters.Event{Type: "progress", Data: []byte(req.Prompt)}) + return adapters.RunResult{RunID: req.RunID, Outcome: f.outcome, Text: "done"}, f.err +} + +type fakePolicy struct { + allowed bool + calls int +} + +type failProgressStore struct{ adapters.EventStore } + +func (s failProgressStore) Append(ctx context.Context, event adapters.Event) error { + if event.Type == "progress/progress" { + return errors.New("store unavailable") + } + return s.EventStore.Append(ctx, event) +} + +func (p *fakePolicy) Authorize(_ context.Context, _ adapters.PolicyInput) (adapters.Decision, error) { + p.calls++ + return adapters.Decision{Allowed: p.allowed, Reason: "test policy"}, nil +} + +func configFor(r adapters.Runner, p adapters.Policy, store adapters.EventStore) Config { + return Config{RunID: "run-1", Goal: core.Goal{ID: "goal-1", Title: "test"}, Obligations: []core.Obligation{{ID: "a", GoalID: "goal-1", Description: "first"}, {ID: "b", GoalID: "goal-1", Description: "second", DependsOn: []core.ID{"a"}}}, Runner: r, Policy: p, Store: store} +} + +func TestRunPersistsAndReplaysCanonicalState(t *testing.T) { + store := memory.NewEventStore() + runner := &fakeRunner{outcome: "success"} + policy := &fakePolicy{allowed: true} + r, err := New(configFor(runner, policy, store)) + if err != nil { + t.Fatal(err) + } + if err := r.Run(context.Background()); err != nil { + t.Fatal(err) + } + if !r.Terminal() || runner.calls != 2 || policy.calls != 2 { + t.Fatalf("terminal=%v calls=%d policy=%d", r.Terminal(), runner.calls, policy.calls) + } + state := r.State() + if state.Goals["goal-1"].Status != core.GoalSucceeded || state.Budget.UsedSteps != 2 { + t.Fatalf("unexpected state: %#v", state) + } + replay, err := New(configFor(&fakeRunner{outcome: "success"}, &fakePolicy{allowed: true}, store)) + if err != nil { + t.Fatal(err) + } + if err := replay.Replay(context.Background()); err != nil { + t.Fatal(err) + } + if replay.State().Revision != state.Revision || replay.State().Goals["goal-1"].Status != core.GoalSucceeded { + t.Fatalf("replay differs: %#v", replay.State()) + } + events, err := store.List(context.Background(), "run-1", 0) + if err != nil { + t.Fatal(err) + } + if len(events) < 8 { + t.Fatalf("expected canonical and progress events, got %d", len(events)) + } +} + +func TestAuthorizationDenialBlocksWithoutRunning(t *testing.T) { + store := memory.NewEventStore() + runner := &fakeRunner{outcome: "success"} + policy := &fakePolicy{allowed: false} + r, err := New(configFor(runner, policy, store)) + if err != nil { + t.Fatal(err) + } + d, err := r.Step(context.Background()) + if err != nil { + t.Fatal(err) + } + if d.Kind != core.DecisionBlock || runner.calls != 0 { + t.Fatalf("got %#v calls=%d", d, runner.calls) + } + if !r.Terminal() || r.State().Goals["goal-1"].Status != core.GoalBlocked { + t.Fatalf("expected blocked terminal state") + } +} + +func TestRunnerFailureBecomesObservation(t *testing.T) { + store := memory.NewEventStore() + runner := &fakeRunner{outcome: "failure", err: errors.New("runner unavailable")} + r, err := New(configFor(runner, nil, store)) + if err != nil { + t.Fatal(err) + } + if _, err := r.Step(context.Background()); err != nil { + t.Fatal(err) + } + if got := r.State().Obligations["a"].Status; got != core.ObligationFailed { + t.Fatalf("got %q", got) + } +} + +func TestConfigurationRejectsMissingStoreAndRunner(t *testing.T) { + _, err := New(Config{RunID: "r", Goal: core.Goal{ID: "g"}}) + if !errors.Is(err, ErrInvalidConfig) { + t.Fatalf("got %v", err) + } + _, err = New(Config{RunID: "r", Goal: core.Goal{ID: "g"}, Runner: &fakeRunner{}, Store: nil, Obligations: []core.Obligation{{ID: "o", GoalID: "g"}}}) + if !errors.Is(err, ErrStoreRequired) { + t.Fatalf("got %v", err) + } +} + +func TestRunPersistsBudgetExhaustionAndTerminates(t *testing.T) { + store := memory.NewEventStore() + cfg := configFor(&fakeRunner{outcome: "success"}, nil, store) + cfg.Budget.MaxCost = 1 + cfg.Costs = map[core.ID]uint64{"a": 2} + r, err := New(cfg) + if err != nil { + t.Fatal(err) + } + if err := r.Run(context.Background()); err != nil { + t.Fatal(err) + } + if got := r.State().Goals["goal-1"].Status; got != core.GoalExhausted { + t.Fatalf("goal status = %q, want exhausted", got) + } +} + +func TestProgressPersistenceFailureFailsTheStep(t *testing.T) { + base := memory.NewEventStore() + r, err := New(configFor(&fakeRunner{outcome: "success"}, nil, failProgressStore{base})) + if err != nil { + t.Fatal(err) + } + if _, err := r.Step(context.Background()); err == nil || !strings.Contains(err.Error(), "persist runner progress") { + t.Fatalf("got %v, want progress persistence error", err) + } +}