fix(core): 按 WorkflowRun 所有者登记模型尝试

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")
+}

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"

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);

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`.
+

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`。
+

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)
+	}
+}

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