feat(hub): multi-node broker dispatch — affinity, durable intent, lifecycle + message routing

[ptone]

Summary

Adds cross-node broker dispatch so API calls hitting any hub replica are correctly routed to the replica that holds the broker's control-channel WebSocket. Fixes the silent split-brain where messages sent to the wrong replica were dropped.

Depends on #304 (Postgres Core) — base branch set to main.

What's included

• B1 — Affinity — Per-process instanceID; connected_hub_id/session_id/connected_at columns on runtime_brokers; Claim/Release CAS methods; sessionID threading + disconnect-race fix (stale offline stamp on broker reconnect)
• B2 — State machine substrate — broker_dispatch durable intent table; messages.dispatch_state column; BrokerDispatch store methods with CAS claim; PostgresCommandBus LISTEN/NOTIFY signal listener on scion_broker_cmd; reconcile-on-connect drain
• B3 — Message dispatch — route() decision in HybridBrokerClient (local → forward → HTTP → undeliverable); transactional intent+signal for cross-node messages; owner-side drain with MarkMessageDispatched
• B4 — Lifecycle dispatch — Rolling-timeout wait helper (resets on any phase/activity change); start/stop/restart/delete cross-node dispatch; create-time data ops (finalize_env, check_prompt, gather)
• B5 — Hardening — Stale-affinity + stuck-dispatch reaper singleton; pending-message sweep; Cloud Monitoring dispatch metrics
• Fixes — OAuth session made replica-portable; JWT signing keys derived from shared SESSION_SECRET (eliminates cross-replica login loop)

Test plan

• go build ./... passes
• go test ./pkg/hub/... passes
• Two-hub integration: dispatch a start from Hub A for a broker on Hub B → agent reaches running
• Chaos gate: kill owning hub mid-start → Hub B completes via reconcile drain, no double-execution
• Broker flap: reconnecting broker has affinity updated, no stale offline stamp

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

[gemini-code-assist]

Code Review

This pull request introduces PostgreSQL support to Scion, enabling multi-node deployments through Postgres LISTEN/NOTIFY for event publishing and command dispatching. It also adds a database migration command to copy state from SQLite to Postgres, refactors tests to use the Ent-backed memory database, and updates design documentation. The review feedback highlights several critical issues, including potential Windows sharing violations during SQLite file deletion, a stale nil capture of the control channel manager, incorrect parsing of file URLs with triple slashes, and SQLite database initialization failures in Go subtests due to unescaped slashes.

[gemini-code-assist]

Change the deferred close to a nil-checked closure. This allows us to manually close the source connection and set src to nil before dropping the file, preventing sharing violations on Windows and ensuring clean resource cleanup.

	defer func() {
		if src != nil {
			_ = src.Close()
		}
	}()

[gemini-code-assist]

Explicitly close the source database connection and set src to nil before calling dropSQLiteFile. This prevents deleting a file that is still actively held open by the connection pool, which is especially critical on Windows where it would otherwise fail with a sharing violation.

	if migrateDropSource {
		_ = src.Close()
		src = nil
		fmt.Fprintf(out, "Dropping source SQLite file: %s\n", srcPath)
		if err := dropSQLiteFile(srcPath); err != nil {
			return fmt.Errorf("dropping source: %w", err)
		}
		fmt.Fprintln(out, "Source dropped.")
	} else {
		fmt.Fprintf(out, "Source left in place: %s\n", srcPath)
	}

[gemini-code-assist]

Evaluate hubSrv.GetControlChannelManager() inside the closure instead of capturing a local variable. If GetControlChannelManager() returns nil at the time newCommandBus is called but is initialized later, the closure would otherwise permanently capture the stale nil value, causing ownsLocally to always return false.

Suggested change

	mgr := hubSrv.GetControlChannelManager()
	ownsLocally := func(brokerID string) bool {
	if mgr == nil {
	return false
	}
	return mgr.IsConnected(brokerID)
	}
	ownsLocally := func(brokerID string) bool {
	mgr := hubSrv.GetControlChannelManager()
	if mgr == nil {
	return false
	}
	return mgr.IsConnected(brokerID)
	}

[gemini-code-assist]

Trim the file:// prefix first before checking for file:. Standard file URLs often use the file:///path/to/db format, and only trimming file: would result in an invalid path like ///path/to/db on Unix-like systems.

	case strings.HasPrefix(raw, "file://"):
		path = strings.TrimPrefix(raw, "file://")
	case strings.HasPrefix(raw, "file:"):
		path = strings.TrimPrefix(raw, "file:")

[gemini-code-assist]

Add a test case to verify that file URLs with triple slashes (e.g., file:///tmp/hub.db) are correctly parsed and normalized to /tmp/hub.db.

Suggested change

	},
	{
	name: "file url with query",
	in: "file:/tmp/hub.db?cache=shared",
	wantDSN: "file:/tmp/hub.db?cache=shared",
	wantPath: "/tmp/hub.db",
	{
	name: "file url with query",
	in: "file:/tmp/hub.db?cache=shared",
	wantDSN: "file:/tmp/hub.db?cache=shared",
	wantPath: "/tmp/hub.db",
	},
	{
	name: "file url with triple slashes",
	in: "file:///tmp/hub.db",
	wantDSN: "file:/tmp/hub.db?cache=shared",
	wantPath: "/tmp/hub.db",
	},

[gemini-code-assist]

Sanitize t.Name() by replacing slashes (/) with underscores. Subtests in Go contain slashes in their names (e.g., TestParent/Subtest), which SQLite interprets as directory paths even for in-memory databases, causing "unable to open database file" errors if the directory does not exist on disk. Note: you may need to import "strings" if it is not already present.

Suggested change

	client, err := entc.OpenSQLite("file:"+t.Name()+"?mode=memory&cache=shared", entc.PoolConfig{})
	dbName := strings.ReplaceAll(t.Name(), "/", "_")
	client, err := entc.OpenSQLite("file:"+dbName+"?mode=memory&cache=shared", entc.PoolConfig{})