diff --git a/integration/migration_upgrade_path_test.go b/integration/migration_upgrade_path_test.go index 150894971..a6ae22167 100644 --- a/integration/migration_upgrade_path_test.go +++ b/integration/migration_upgrade_path_test.go @@ -1261,3 +1261,50 @@ func TestMigration0043ExternalGitLineAppliesFreshAndUpgrades(t *testing.T) { _, e = pool.Exec(`DELETE FROM repository_operation_events WHERE operation_id=$1`, operation) wantPGError(t, e, "P0001") } + +// 0044 lets the attempt ledger record a WorkflowRun-owned binding without inventing an IssueRun. +// It takes 0040's discriminator shape: run_id becomes nullable, workflow_run_id is added, and each +// owner gets its own tenant-scoped run and binding reference. A fresh database and one that stopped +// at 0043 reach the same shape, and rerunning the migration changes nothing. +func TestMigration0044ModelAttemptWorkflowRunAppliesFreshAndUpgrades(t *testing.T) { + shape := func(pool *sql.DB) (nullables, ownerFKs, legacyFKs, ownerCheck, indexes, identities int) { + t.Helper() + must(t, pool.QueryRow(`SELECT count(*) FROM information_schema.columns WHERE table_schema=current_schema() + AND table_name='model_request_attempts' AND column_name IN ('run_id','workflow_run_id') AND is_nullable='YES'`).Scan(&nullables)) + must(t, pool.QueryRow(`SELECT count(*) FROM pg_constraint WHERE conrelid='model_request_attempts'::regclass AND contype='f' + AND conname IN ('model_attempt_issue_run_scope','model_attempt_workflow_run_scope','model_attempt_issue_binding','model_attempt_workflow_binding')`).Scan(&ownerFKs)) + must(t, pool.QueryRow(`SELECT count(*) FROM pg_constraint WHERE conrelid='model_request_attempts'::regclass AND contype='f' + AND conname IN ('model_request_attempts_run_id_tenant_id_fkey','model_request_attempts_binding_id_run_id_fkey')`).Scan(&legacyFKs)) + must(t, pool.QueryRow(`SELECT count(*) FROM pg_constraint WHERE conrelid='model_request_attempts'::regclass AND contype='c' + AND conname='model_attempt_owner_exclusive'`).Scan(&ownerCheck)) + must(t, pool.QueryRow(`SELECT count(*) FROM pg_indexes WHERE schemaname=current_schema() AND tablename='model_request_attempts' + AND indexname IN ('model_attempt_run_order','model_attempt_workflow_run_order')`).Scan(&indexes)) + must(t, pool.QueryRow(`SELECT count(*) FROM pg_constraint WHERE contype='u' + AND ((conrelid='workflow_runs'::regclass AND conname='workflow_run_tenant_identity') + OR (conrelid='run_model_bindings'::regclass AND conname='model_binding_workflow_run_identity'))`).Scan(&identities)) + return nullables, ownerFKs, legacyFKs, ownerCheck, indexes, identities + } + assertShape := func(t *testing.T, pool *sql.DB, stage string) { + t.Helper() + if n, fk, legacy, check, idx, id := shape(pool); n != 2 || fk != 4 || legacy != 0 || check != 1 || idx != 2 || id != 2 { + t.Fatalf("%s: nullable owners %d owner fks %d legacy fks %d exclusive check %d indexes %d identities %d", + stage, n, fk, legacy, check, idx, id) + } + } + + freshPool, _ := testSchema(t, "test_ma44f_") + migratedTwice(t, newStoreOnSchema(t, freshPool)) + assertShape(t, freshPool, "fresh") + + pool, _ := testSchema(t, "test_ma44_") + applyMigrationsUpTo(t, pool, migrationsBefore(t, "0044_model_attempt_workflow_run.sql")) + if n, fk, legacy, check, idx, id := shape(pool); n != 0 || fk != 0 || legacy != 2 || check != 0 || idx != 1 || id != 0 { + t.Fatalf("before 0044: nullable owners %d owner fks %d legacy fks %d exclusive check %d indexes %d identities %d", + n, fk, legacy, check, idx, id) + } + if columnNullable(t, pool, "model_request_attempts", "run_id") { + t.Fatal("before 0044 run_id must still be required") + } + migratedTwice(t, newStoreOnSchema(t, pool)) + assertShape(t, pool, "upgrade") +} diff --git a/integration/session_statistics_test.go b/integration/session_statistics_test.go index 2c9735f55..d315265bc 100644 --- a/integration/session_statistics_test.go +++ b/integration/session_statistics_test.go @@ -69,10 +69,15 @@ func TestMigration0035ModelAttemptsFreshAndUpgradeWithoutBackfill(t *testing.T) var rows, constraints, indexes int must(t, pool.QueryRow(`SELECT count(*) FROM model_request_attempts`).Scan(&rows)) must(t, pool.QueryRow(`SELECT count(*) FROM pg_constraint WHERE conrelid='model_request_attempts'::regclass AND contype='f'`).Scan(&constraints)) - must(t, pool.QueryRow(`SELECT count(*) FROM pg_indexes WHERE schemaname=current_schema() AND indexname='model_attempt_run_order'`).Scan(&indexes)) - if rows != 0 || constraints != 3 || indexes != 1 { + must(t, pool.QueryRow(`SELECT count(*) FROM pg_indexes WHERE schemaname=current_schema() AND indexname IN ('model_attempt_run_order','model_attempt_workflow_run_order')`).Scan(&indexes)) + // 0035's three foreign keys are replaced by one IssueRun/WorkflowRun pair each for the run + // and the binding; both owners are nullable and discriminated by exactly one check. + if rows != 0 || constraints != 5 || indexes != 2 { t.Fatalf("rows=%d scoped constraints=%d indexes=%d", rows, constraints, indexes) } + if !columnNullable(t, pool, "model_request_attempts", "run_id") || !columnNullable(t, pool, "model_request_attempts", "workflow_run_id") { + t.Fatal("attempt owners must both be nullable and discriminated by model_attempt_owner_exclusive") + } } for _, upgrade := range []bool{false, true} { name := "fresh" diff --git a/internal/core/migrations/0044_model_attempt_workflow_run.sql b/internal/core/migrations/0044_model_attempt_workflow_run.sql new file mode 100644 index 000000000..d306d66e7 --- /dev/null +++ b/internal/core/migrations/0044_model_attempt_workflow_run.sql @@ -0,0 +1,38 @@ +-- Model attempts record WorkflowRun-owned bindings (circus decision #262 option A; specs ADR +-- 20261016-recorded-http-attempt-statistics). 0040 gave run_model_bindings an exactly-one owner +-- discriminator, but 0035's attempt ledger still required an IssueRun, so every WorkflowRun +-- request failed its foreign key before reaching the provider. The ledger now mirrors 0040: exactly +-- one of run_id / workflow_run_id, each with its own tenant-scoped run reference and binding +-- reference. Registration is the only consumer; WorkflowRun statistics reads stay closed. +-- +-- Every existing row carries run_id: 0035 required it, and no WorkflowRun attempt could ever be +-- inserted. There is nothing to backfill. Additive and forward-only. + +ALTER TABLE model_request_attempts ALTER COLUMN run_id DROP NOT NULL; +ALTER TABLE model_request_attempts ADD COLUMN workflow_run_id uuid; +ALTER TABLE model_request_attempts ADD CONSTRAINT model_attempt_owner_exclusive + CHECK ((run_id IS NULL) <> (workflow_run_id IS NULL)); + +-- Tenant-scoped identity for the new owner, and the binding identity its composite reference +-- needs; 0035 already added the IssueRun halves. +ALTER TABLE workflow_runs ADD CONSTRAINT workflow_run_tenant_identity UNIQUE(id,tenant_id); +ALTER TABLE run_model_bindings ADD CONSTRAINT model_binding_workflow_run_identity UNIQUE(id,workflow_run_id); + +-- Replace each single-owner reference with one per owner. A composite foreign key is not enforced +-- when any referencing column is NULL, so the discriminator above decides which one applies. +ALTER TABLE model_request_attempts DROP CONSTRAINT model_request_attempts_run_id_tenant_id_fkey; +ALTER TABLE model_request_attempts ADD CONSTRAINT model_attempt_issue_run_scope + FOREIGN KEY(run_id,tenant_id) REFERENCES issue_runs(id,tenant_id); +ALTER TABLE model_request_attempts ADD CONSTRAINT model_attempt_workflow_run_scope + FOREIGN KEY(workflow_run_id,tenant_id) REFERENCES workflow_runs(id,tenant_id); + +ALTER TABLE model_request_attempts DROP CONSTRAINT model_request_attempts_binding_id_run_id_fkey; +ALTER TABLE model_request_attempts ADD CONSTRAINT model_attempt_issue_binding + FOREIGN KEY(binding_id,run_id) REFERENCES run_model_bindings(id,run_id); +ALTER TABLE model_request_attempts ADD CONSTRAINT model_attempt_workflow_binding + FOREIGN KEY(binding_id,workflow_run_id) REFERENCES run_model_bindings(id,workflow_run_id); + +-- The IssueRun read path keeps model_attempt_run_order; WorkflowRun reads are not opened yet, but +-- its rows still need an owner-ordered index once one is. +CREATE INDEX model_attempt_workflow_run_order + ON model_request_attempts(tenant_id,workflow_run_id,started_at,id); diff --git a/internal/core/migrations/README.en.md b/internal/core/migrations/README.en.md index ea77ae7cf..b11495954 100644 --- a/internal/core/migrations/README.en.md +++ b/internal/core/migrations/README.en.md @@ -176,3 +176,17 @@ The migration that lands the external Git line (executor root ADR, identity-acce Fresh-database and merge-baseline (through 0042) upgrade evidence — an upgraded database carrying real Projects, credential references and a continuity column — is `TestMigration0043ExternalGitLineAppliesFreshAndUpgrades` in `integration/migration_upgrade_path_test.go`, which also exercises those constraints and the append-only trigger against real rows. +## 0044: Model attempts record WorkflowRun bindings + +`0044_model_attempt_workflow_run.sql` lets `model_request_attempts` record a WorkflowRun binding, +using the same discriminator shape as 0040: `run_id` becomes nullable, `workflow_run_id` is added, +and exactly one of them is required. Each owner gets its own pair of references — the run to +`issue_runs` / `workflow_runs` on `(id, tenant_id)`, and the binding to `run_model_bindings` on +`(id, run_id)` / `(id, workflow_run_id)` — with the two uniqueness constraints the new references +need. Registration is the only consumer; WorkflowRun statistics reads stay closed. The IssueRun +reads keep filtering by `run_id`, which is NULL on WorkflowRun rows and therefore cannot leak into +them. Every existing row carries `run_id` (0035 required it, and the defect made every WorkflowRun +registration fail), so nothing is backfilled. Fresh and 0043-upgrade evidence: +`TestMigration0044ModelAttemptWorkflowRunAppliesFreshAndUpgrades`; real WorkflowRun registration: +`internal/core/model_attempt_workflow_db_test.go`. + diff --git a/internal/core/migrations/README.md b/internal/core/migrations/README.md index 48f688f09..4a479ac3f 100644 --- a/internal/core/migrations/README.md +++ b/internal/core/migrations/README.md @@ -165,3 +165,14 @@ Controller 契约的字段。只新增，不回填。 新库与合并时基线（含 0042）升级（带真实 Project、凭据引用与连续性列的库）证据见 `integration/migration_upgrade_path_test.go` 的 `TestMigration0043ExternalGitLineAppliesFreshAndUpgrades`，其中也对真实行验证了上述约束与 append-only 触发器。 +## 0044：模型尝试登记 WorkflowRun 绑定 + +`0044_model_attempt_workflow_run.sql` 让 `model_request_attempts` 记录 WorkflowRun 的绑定，采用与 +0040 相同的判别形状：`run_id` 改为可空，新增 `workflow_run_id`，两者恰有其一。运行与绑定各按所有者 +一对：分别引用 `issue_runs`/`workflow_runs` 的 `(id, tenant_id)`，以及 `run_model_bindings` 的 +`(id, run_id)`/`(id, workflow_run_id)`，并补上后者所需的两个唯一约束。首期只登记，不开放 WorkflowRun +的统计读取；IssueRun 的读取继续按 `run_id` 过滤，WorkflowRun 行的 `run_id` 为空，不会进入既有读取。 +既有行都带 `run_id`（0035 要求非空，且任何 WorkflowRun 登记都无法写入），无需回填。新库与 0043 升级证据见 +`TestMigration0044ModelAttemptWorkflowRunAppliesFreshAndUpgrades`，真实 WorkflowRun 绑定登记见 +`internal/core/model_attempt_workflow_db_test.go`。 + diff --git a/internal/core/model_attempt_workflow_db_test.go b/internal/core/model_attempt_workflow_db_test.go new file mode 100644 index 000000000..7cdf9178e --- /dev/null +++ b/internal/core/model_attempt_workflow_db_test.go @@ -0,0 +1,84 @@ +package core + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "testing" + + "github.com/google/uuid" +) + +// A WorkflowRun binding must be able to register its outbound attempt. The ledger's foreign keys +// originally required an IssueRun (0035), while 0040 made a WorkflowRun the owner of its bindings, +// so every WorkflowRun model request failed registration and the gateway answered 503 without +// contacting the provider. This drives the real grant and binding chain and the real PostgreSQL +// constraints; success here is what lets the gateway forward the request. +func TestWorkflowModelAttemptRegistersUnderWorkflowRun(t *testing.T) { + store := dispatcherDB(t) + runID, _, _, _, _ := seedWorkflowSessionScene(t, store) + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance 1: %v", err) + } + wsID := settleSessionWorkspace(t, store, runID) + nodeService, spec := startSessionNode(t, store, runID, wsID, "a") + execution := dispatchSession(t, store, runID, "a", nodeService, spec) + // The runtime confirms its binding once the Node has the dispatched session. + if _, err := store.Pool.Exec(`UPDATE runtime_controls SET binding_confirmed=true, + bound_sandbox_id=(SELECT sandbox_instance_id FROM node_instances WHERE workspace_id=$1 AND ended_at IS NULL) + WHERE workspace_id=$1`, wsID); err != nil { + t.Fatal(err) + } + var bindingID, tenant string + var generation int64 + if err := store.Pool.QueryRow(`SELECT id FROM run_model_bindings WHERE workflow_run_id=$1 AND slot_name='default'`, runID).Scan(&bindingID); err != nil { + t.Fatalf("read frozen binding: %v", err) + } + if err := store.Pool.QueryRow(`SELECT tenant_id, runtime_generation FROM workspaces WHERE id=$1`, wsID).Scan(&tenant, &generation); err != nil { + t.Fatal(err) + } + scope := ModelRuntimeScope{TenantID: tenant, WorkspaceID: wsID, Generation: generation} + digest := sha256.Sum256([]byte("synthetic workflow attempt")) + grant, err := store.CreateModelGrant(context.Background(), scope, bindingID, execution, hex.EncodeToString(digest[:])) + if err != nil { + t.Fatalf("create grant: %v", err) + } + // The grant names the binding's owner, which for a WorkflowRun is its run id. + if grant.RunID != runID || grant.BindingID != bindingID || grant.ExecutionID != execution { + t.Fatalf("grant = %+v; want run %s binding %s execution %s", grant, runID, bindingID, execution) + } + attemptID := uuid.NewString() + if err := store.BeginModelAttempt(context.Background(), &grant, attemptID, true); err != nil { + t.Fatalf("begin attempt: %v", err) + } + var issueRun, workflowRun sql.NullString + var storedBinding, storedExecution string + var streamed bool + if err := store.Pool.QueryRow(`SELECT run_id::text, workflow_run_id::text, binding_id, execution_id, streamed + FROM model_request_attempts WHERE id=$1`, attemptID).Scan(&issueRun, &workflowRun, &storedBinding, &storedExecution, &streamed); err != nil { + t.Fatalf("read attempt ledger: %v", err) + } + if issueRun.Valid || !workflowRun.Valid || workflowRun.String != runID { + t.Fatalf("attempt owner = run_id %v workflow_run_id %v; want workflow_run_id %s", issueRun, workflowRun, runID) + } + if storedBinding != bindingID || storedExecution != execution || !streamed { + t.Fatalf("attempt = binding %s execution %s streamed %v; want %s %s true", storedBinding, storedExecution, streamed, bindingID, execution) + } + // Finalization addresses the attempt by id and is independent of the owner discriminator. + firstOutput := int64(1) + if err := store.FinishModelAttempt(context.Background(), attemptID, &ModelAttemptFinish{ + Outcome: "succeeded", DurationMS: 2, FirstOutputMS: &firstOutput, TerminalObserved: true, + Usage: ModelAttemptUsage{InputSemantics: "unknown", Availability: "unavailable"}, + }); err != nil { + t.Fatalf("finish attempt: %v", err) + } + var outcome string + var finished bool + if err := store.Pool.QueryRow(`SELECT outcome, finished_at IS NOT NULL FROM model_request_attempts WHERE id=$1`, attemptID).Scan(&outcome, &finished); err != nil { + t.Fatal(err) + } + if outcome != "succeeded" || !finished { + t.Fatalf("attempt settled outcome=%q finished=%v; want succeeded/true", outcome, finished) + } +} diff --git a/internal/core/model_attempts.go b/internal/core/model_attempts.go index 8d4c3396d..9f4fae4b9 100644 --- a/internal/core/model_attempts.go +++ b/internal/core/model_attempts.go @@ -22,8 +22,16 @@ func (s *Store) BeginModelAttempt(ctx context.Context, grant *ModelGrant, attemp require(old.S("grantId") == grant.ID && old.B("streamed") == streamed && old.S("outcome") == "started", 409, "model_attempt_conflict") return nil } - t.exec(`INSERT INTO model_request_attempts(id,tenant_id,run_id,binding_id,grant_id,execution_id,model_id,protocol,streamed) - VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)`, attemptID, scope.TenantID, b.S("runId"), g.S("bindingId"), grant.ID, g.S("executionId"), b.O("model").S("id"), b.S("protocol"), streamed) + // The binding names its owner with exactly one of run_id / workflow_run_id (0040), and the + // ledger mirrors that discriminator (0044). liveModelBinding maps either owner into runId for + // the runtime gates, so the raw workflow_run_id column decides which foreign key this row + // satisfies — an IssueRun attempt must never be written under the WorkflowRun reference. + runID, workflowRunID := b.S("runId"), "" + if b.S("workflowRunId") != "" { + runID, workflowRunID = "", b.S("workflowRunId") + } + t.exec(`INSERT INTO model_request_attempts(id,tenant_id,run_id,workflow_run_id,binding_id,grant_id,execution_id,model_id,protocol,streamed) + VALUES($1,$2,NULLIF($3,'')::uuid,NULLIF($4,'')::uuid,$5,$6,$7,$8,$9,$10)`, attemptID, scope.TenantID, runID, workflowRunID, g.S("bindingId"), grant.ID, g.S("executionId"), b.O("model").S("id"), b.S("protocol"), streamed) return nil }) return err