diff --git a/go-api/cmd/importagents/main.go b/go-api/cmd/importagents/main.go index 0df24f7..6ee5071 100644 --- a/go-api/cmd/importagents/main.go +++ b/go-api/cmd/importagents/main.go @@ -85,24 +85,8 @@ func run(dir, skillDir, orgSlug string, dryRun bool, timeout time.Duration) erro return err } - // Parsed before anything is opened, so a malformed spec is a message rather - // than a half-finished import. Every spec, not the first failure: an - // operator fixing five typos should see five, not one per run. - var problems []string - for _, s := range specs { - if len(s.parsed.Errors) > 0 { - problems = append(problems, fmt.Sprintf(" %s: %s", - s.name, strings.Join(s.parsed.Errors, "; "))) - } - if s.parsed.Status != "published" { - problems = append(problems, fmt.Sprintf( - " %s: status is %q; only a published spec can be imported", - s.name, s.parsed.Status)) - } - } - if len(problems) > 0 { - return fmt.Errorf("%d spec(s) will not import:\n%s", - len(problems), strings.Join(problems, "\n")) + if err := validateSpecs(specs); err != nil { + return err } for _, s := range specs { @@ -116,20 +100,9 @@ func run(dir, skillDir, orgSlug string, dryRun bool, timeout time.Duration) erro // written. The runtime refuses to load an agent with a missing dependency, // so importing one without its skills produces an agent that exists and // cannot run — a failure that surfaces per request instead of here. - // §3: subagents must form a DAG, checked at PUBLISH. Every spec is in hand - // here, which is the only place that is cheaply true — so this is where the - // check belongs. The runtime depth cap still bounds a cycle that reaches - // run time by another route. - graph := make(map[string][]string, len(specs)) - for _, s := range specs { - graph[s.parsed.ID] = s.parsed.Subagents + if err := validateGraph(specs); err != nil { + return err } - if cycle := definition.FindSubagentCycle(graph); cycle != "" { - return fmt.Errorf("the subagent graph has a cycle: %s\n\n"+ - "Nothing was written. Delegation follows these edges, so a loop is a "+ - "run that delegates until it runs out of budget.", cycle) - } - if missing := missingSkills(specs, skills); len(missing) > 0 { return fmt.Errorf("%d skill(s) named by an agent are not in %s: %s", len(missing), skillDir, strings.Join(missing, ", ")) @@ -174,73 +147,9 @@ func run(dir, skillDir, orgSlug string, dryRun bool, timeout time.Duration) erro } defer tx.Rollback(ctx) //nolint:errcheck // rolled back unless committed below - // Skills first. An agent row that lands before its dependencies exist is - // briefly unloadable, and inside one transaction that is invisible — but - // ordering them correctly costs nothing and means a future non-transactional - // path is not silently broken. - // Versions are recorded through the same transaction, so the history and - // the definition it describes cannot disagree: either both land or neither - // does. - versions := repo.NewVersionsRepo(tx) - ident := authctx.Identity{OrgID: orgID, UserID: author} - - skillsWritten, skillVersions := 0, 0 - for _, sk := range skills { - if err := upsertSkill(ctx, tx, orgID, author, sk); err != nil { - return fmt.Errorf("%s: %w", sk.name, err) - } - skillsWritten++ - - // Skills are numbered by the server rather than by their author — they - // have no `version:` to read. See snapshotSkill in internal/service, - // which does the same for the authoring path. - recorded, err := snapshotSkill(ctx, versions, ident, sk) - if err != nil { - return fmt.Errorf("%s: record version: %w", sk.name, err) - } - if recorded { - skillVersions++ - } - } - - // - // Snapshot is what refuses a spec that changed without raising its - // `version:`. Every such spec is collected rather than the first one - // returned, for the same reason the parse errors above are — an operator - // who forgot to bump three files should see three. Collecting is safe here - // because that refusal comes from comparing a row this code read, not from - // a failed statement: the INSERT is ON CONFLICT DO NOTHING, so the - // transaction is still healthy and the remaining specs can be checked. - inserted, updated, versioned := 0, 0, 0 - var rewrites []string - for _, s := range specs { - wasNew, err := upsert(ctx, tx, orgID, author, s) - if err != nil { - return fmt.Errorf("%s: %w", s.name, err) - } - if wasNew { - inserted++ - } else { - updated++ - } - - recorded, conflict, err := snapshotAgent(ctx, versions, ident, s) - switch { - case err != nil: - return fmt.Errorf("%s: record version: %w", s.name, err) - case conflict != "": - rewrites = append(rewrites, fmt.Sprintf(" %s: %s", s.name, conflict)) - case recorded: - versioned++ - } - } - - if len(rewrites) > 0 { - return fmt.Errorf( - "%d spec(s) would rewrite a version that is already published:\n%s\n\n"+ - "Nothing was written. Raise `version:` in the frontmatter of each, or "+ - "restore the published text.", - len(rewrites), strings.Join(rewrites, "\n")) + out, err := importInto(ctx, tx, orgID, author, specs, skills) + if err != nil { + return err } if err := tx.Commit(ctx); err != nil { @@ -249,7 +158,145 @@ func run(dir, skillDir, orgSlug string, dryRun bool, timeout time.Duration) erro fmt.Printf("\n%d agent(s) published, %d updated, %d agent version(s) recorded, "+ "%d skill(s) written, %d skill version(s) recorded, into %s\n", - inserted, updated, versioned, skillsWritten, skillVersions, orgSlug) + out.inserted, out.updated, out.versioned, out.skillsWritten, + out.skillVersions, orgSlug) + return nil +} + +// importCounts is what one import did. +type importCounts struct { + inserted int + updated int + versioned int + skillsWritten int + skillVersions int +} + +// importInto writes one validated set of specs and skills through a +// transaction, and reports what it did. +// +// Separated from run() so it can be TESTED. run() loads configuration, opens +// its own pool and resolves a tenant from a slug — none of which a test can +// supply, which is why this command had no tests at all while carrying the +// rules that decide whether a deploy is allowed to change a published agent. +// Everything interesting lives here; run() is the wiring around it. +// +// The caller owns the transaction, and therefore the decision to commit. On any +// error, including a refused rewrite, nothing here has committed and the +// caller's deferred rollback undoes the writes that did happen. +func importInto(ctx context.Context, tx pgx.Tx, orgID, author string, + specs []spec, skills []skillSpec) (importCounts, error) { + + var out importCounts + + // Versions are recorded through the same transaction, so the history and + // the definition it describes cannot disagree: either both land or neither + // does. + versions := repo.NewVersionsRepo(tx) + ident := authctx.Identity{OrgID: orgID, UserID: author} + + // Skills first. An agent row that lands before its dependencies exist is + // briefly unloadable, and inside one transaction that is invisible — but + // ordering them correctly costs nothing and means a future + // non-transactional path is not silently broken. + for _, sk := range skills { + if err := upsertSkill(ctx, tx, orgID, author, sk); err != nil { + return out, fmt.Errorf("%s: %w", sk.name, err) + } + out.skillsWritten++ + + // Skills are numbered by the server rather than by their author — they + // have no `version:` to read. See snapshotSkill in internal/service, + // which does the same for the authoring path. + recorded, err := snapshotSkill(ctx, versions, ident, sk) + if err != nil { + return out, fmt.Errorf("%s: record version: %w", sk.name, err) + } + if recorded { + out.skillVersions++ + } + } + + // snapshotAgent is what refuses a spec that changed without raising its + // `version:`, or that lowers it. Every such spec is collected rather than + // the first one returned, for the same reason the parse errors are — an + // operator who forgot to bump three files should see three. Collecting is + // safe because that refusal comes from comparing rows this code read, not + // from a failed statement: the INSERT is ON CONFLICT DO NOTHING, so the + // transaction is still healthy and the remaining specs can be checked. + var rewrites []string + for _, s := range specs { + wasNew, err := upsert(ctx, tx, orgID, author, s) + if err != nil { + return out, fmt.Errorf("%s: %w", s.name, err) + } + if wasNew { + out.inserted++ + } else { + out.updated++ + } + + recorded, conflict, err := snapshotAgent(ctx, versions, ident, s) + switch { + case err != nil: + return out, fmt.Errorf("%s: record version: %w", s.name, err) + case conflict != "": + rewrites = append(rewrites, fmt.Sprintf(" %s: %s", s.name, conflict)) + case recorded: + out.versioned++ + } + } + + if len(rewrites) > 0 { + return out, fmt.Errorf( + "%d spec(s) would rewrite a version that is already published:\n%s\n\n"+ + "Nothing was written. Raise `version:` in the frontmatter of each, or "+ + "restore the published text.", + len(rewrites), strings.Join(rewrites, "\n")) + } + return out, nil +} + +// validateSpecs rejects specs that cannot be imported, reporting every one. +// +// Parsed before anything is opened, so a malformed spec is a message rather +// than a half-finished import. Every spec, not the first failure: an operator +// fixing five typos should see five, not one per run. +func validateSpecs(specs []spec) error { + var problems []string + for _, s := range specs { + if len(s.parsed.Errors) > 0 { + problems = append(problems, fmt.Sprintf(" %s: %s", + s.name, strings.Join(s.parsed.Errors, "; "))) + } + if s.parsed.Status != "published" { + problems = append(problems, fmt.Sprintf( + " %s: status is %q; only a published spec can be imported", + s.name, s.parsed.Status)) + } + } + if len(problems) > 0 { + return fmt.Errorf("%d spec(s) will not import:\n%s", + len(problems), strings.Join(problems, "\n")) + } + return nil +} + +// validateGraph enforces §3's DAG requirement across the whole set. +// +// Every spec is in hand here, which is the only place that is cheaply true — +// so this is where the check belongs. The runtime depth cap still bounds a +// cycle that reaches run time by another route. +func validateGraph(specs []spec) error { + graph := make(map[string][]string, len(specs)) + for _, s := range specs { + graph[s.parsed.ID] = s.parsed.Subagents + } + if cycle := definition.FindSubagentCycle(graph); cycle != "" { + return fmt.Errorf("the subagent graph has a cycle: %s\n\n"+ + "Nothing was written. Delegation follows these edges, so a loop is a "+ + "run that delegates until it runs out of budget.", cycle) + } return nil } diff --git a/go-api/cmd/importagents/main_test.go b/go-api/cmd/importagents/main_test.go new file mode 100644 index 0000000..fb91022 --- /dev/null +++ b/go-api/cmd/importagents/main_test.go @@ -0,0 +1,244 @@ +package main + +import ( + "context" + "fmt" + "strings" + "testing" + + "github.com/krow/krow-backend/go-api/internal/definition" + "github.com/krow/krow-backend/go-api/internal/testutil" +) + +// specFor builds a parsed spec the way loadSpecs would, without a file. +func specFor(t *testing.T, id string, version int, subagents ...string) spec { + t.Helper() + var sub string + if len(subagents) > 0 { + sub = "subagents:\n" + for _, s := range subagents { + sub += " - " + s + "\n" + } + } + raw := fmt.Sprintf(`--- +id: %s +name: %s +description: a spec built for a test +icon: layers +status: published +version: %d +reasoning: balanced +pages: + - talent-pool +%s--- + +## Instructions +Answer the question, version %d. +`, id, strings.ToUpper(id[:1])+id[1:], version, sub, version) + + parsed, err := definition.ParseAgent(raw, definition.Options{}) + if err != nil { + t.Fatalf("fixture %q does not parse: %v", id, err) + } + return spec{name: id + ".md", raw: raw, parsed: parsed} +} + +/* ── The pure checks, which need no database ─────────────────────────────── */ + +func TestValidateSpecsReportsEveryProblem(t *testing.T) { + draft := specFor(t, "draft-agent", 1) + draft.parsed.Status = "draft" + broken := specFor(t, "broken-agent", 1) + broken.parsed.Errors = []string{"something is wrong"} + + err := validateSpecs([]spec{draft, broken, specFor(t, "fine-agent", 1)}) + if err == nil { + t.Fatal("two bad specs were accepted") + } + // Both, not the first: an operator fixing two problems should see two. + for _, want := range []string{"draft-agent", "broken-agent"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("the report does not mention %q:\n%s", want, err) + } + } + if strings.Contains(err.Error(), "fine-agent") { + t.Errorf("a valid spec was reported as a problem:\n%s", err) + } + + if err := validateSpecs([]spec{specFor(t, "fine-agent", 1)}); err != nil { + t.Errorf("a valid spec was refused: %v", err) + } +} + +func TestValidateGraphRefusesACycle(t *testing.T) { + acyclic := []spec{ + specFor(t, "a-agent", 1, "b-agent"), + specFor(t, "b-agent", 1), + } + if err := validateGraph(acyclic); err != nil { + t.Errorf("a chain was called a cycle: %v", err) + } + + cyclic := []spec{ + specFor(t, "a-agent", 1, "b-agent"), + specFor(t, "b-agent", 1, "a-agent"), + } + err := validateGraph(cyclic) + if err == nil { + t.Fatal("a cycle was accepted") + } + if !strings.Contains(err.Error(), "a-agent") || !strings.Contains(err.Error(), "b-agent") { + t.Errorf("the message does not name the edge to cut:\n%s", err) + } +} + +/* ── The write phase, against a real database ────────────────────────────── */ + +// importOnce runs one import in its own transaction and commits it, the way +// run() does. +// +// The author comes from resolveAuthor rather than a literal, so this exercises +// the production path and fails loudly if an organization has nobody to +// attribute specs to — which is a real deployment condition, not a test +// detail. +func importOnce(t *testing.T, h *testutil.Harness, specs []spec) (importCounts, error) { + t.Helper() + ctx := context.Background() + + author, err := resolveAuthor(ctx, h.Pool, h.OrgID) + if err != nil { + t.Fatalf("resolve author: %v", err) + } + + tx, err := h.Pool.Begin(ctx) + if err != nil { + t.Fatalf("begin: %v", err) + } + defer tx.Rollback(ctx) //nolint:errcheck + + out, err := importInto(ctx, tx, h.OrgID, author, specs, nil) + if err != nil { + return out, err + } + if err := tx.Commit(ctx); err != nil { + t.Fatalf("commit: %v", err) + } + return out, nil +} + +func TestImportRecordsVersionsAndIsIdempotent(t *testing.T) { + h := testutil.New(t) + specs := []spec{specFor(t, "import-a", 1), specFor(t, "import-b", 1)} + + out, err := importOnce(t, h, specs) + if err != nil { + t.Fatalf("first import: %v", err) + } + if out.inserted != 2 || out.versioned != 2 { + t.Errorf("first import: inserted=%d versioned=%d, want 2 and 2", out.inserted, out.versioned) + } + + // Again, unchanged. Nothing new is recorded — the number must not report + // nine every deploy, which is what it used to do. + out, err = importOnce(t, h, specs) + if err != nil { + t.Fatalf("second import: %v", err) + } + if out.versioned != 0 { + t.Errorf("re-importing unchanged specs recorded %d version(s), want 0", out.versioned) + } + if out.updated != 2 { + t.Errorf("second import: updated=%d, want 2", out.updated) + } +} + +func TestImportRefusesRewritingAPublishedVersion(t *testing.T) { + h := testutil.New(t) + if _, err := importOnce(t, h, []spec{specFor(t, "rewrite-me", 1)}); err != nil { + t.Fatalf("first import: %v", err) + } + + // Same version, different body. + changed := specFor(t, "rewrite-me", 1) + changed.raw = strings.Replace(changed.raw, "version 1.", "something else entirely.", 1) + reparsed, err := definition.ParseAgent(changed.raw, definition.Options{}) + if err != nil { + t.Fatalf("fixture does not parse: %v", err) + } + changed.parsed = reparsed + + _, err = importOnce(t, h, []spec{changed}) + if err == nil { + t.Fatal("a changed spec republished at the same version was accepted") + } + if !strings.Contains(err.Error(), "rewrite") { + t.Errorf("unexpected error: %v", err) + } + + // And nothing landed: the live row still says what v1 said. + var live string + if err := h.Pool.QueryRow(context.Background(), + `SELECT markdown FROM agent_definitions WHERE org_id = $1::uuid AND definition_id = 'rewrite-me'`, + h.OrgID).Scan(&live); err != nil { + t.Fatalf("read back: %v", err) + } + if strings.Contains(live, "something else entirely") { + t.Error("the refused import was committed anyway") + } +} + +func TestImportRefusesAVersionGoingBackwards(t *testing.T) { + h := testutil.New(t) + if _, err := importOnce(t, h, []spec{specFor(t, "backwards", 1)}); err != nil { + t.Fatalf("v1: %v", err) + } + if _, err := importOnce(t, h, []spec{specFor(t, "backwards", 2)}); err != nil { + t.Fatalf("v2: %v", err) + } + + // Back to v1, byte-for-byte what v1 said. Nothing conflicts, which is why + // this used to succeed and silently revert the deployed agent. + _, err := importOnce(t, h, []spec{specFor(t, "backwards", 1)}) + if err == nil { + t.Fatal("a lowered version was accepted") + } + if !strings.Contains(err.Error(), "monotonic") { + t.Errorf("the message does not explain why: %v", err) + } + + var version int + if err := h.Pool.QueryRow(context.Background(), + `SELECT version FROM agent_definitions WHERE org_id = $1::uuid AND definition_id = 'backwards'`, + h.OrgID).Scan(&version); err != nil { + t.Fatalf("read back: %v", err) + } + if version != 2 { + t.Errorf("live version = %d, want 2 — the refused import rolled it back", version) + } +} + +// Raising the version is the supported way to change a published spec. +func TestImportAcceptsARaisedVersion(t *testing.T) { + h := testutil.New(t) + if _, err := importOnce(t, h, []spec{specFor(t, "raised", 1)}); err != nil { + t.Fatalf("v1: %v", err) + } + out, err := importOnce(t, h, []spec{specFor(t, "raised", 2)}) + if err != nil { + t.Fatalf("v2 was refused: %v", err) + } + if out.versioned != 1 { + t.Errorf("recorded %d version(s) for a raised version, want 1", out.versioned) + } + + var n int + if err := h.Pool.QueryRow(context.Background(), + `SELECT count(*) FROM definition_versions + WHERE org_id = $1::uuid AND kind = 'agent' AND definition_id = 'raised'`, + h.OrgID).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + if n != 2 { + t.Errorf("history holds %d versions, want 2 (v1 and v2)", n) + } +}