feat(scum): wire platform schema probe jobs

2026-08-12 11:06:14 +08:00
parent 6d4319187f
commit 2638eda55b
15 changed files with 405 additions and 13 deletions
@@ -334,6 +334,18 @@ func validateExecutionResultForJob(job domain.Job, result domain.RunJobResult) e
 		if result.State == domain.JobStateSucceeded && result.ExecutionResult.Kind != "run.update.staged" {
 			return validationError("Run self-update result type is invalid")
 		}
+	case domain.JobCapabilityRemoteRunDBSQLiteProbe:
+		if result.ExecutionResult.Kind != "" && result.ExecutionResult.Kind != scumSchemaProbeExecutionKind {
+			return validationError("SQLite schema probe result type is invalid")
+		}
+		if result.State == domain.JobStateSucceeded {
+			if result.ExecutionResult.SQLiteSchemaProbe == nil {
+				return validationError("SQLite schema probe terminal result is required")
+			}
+			if err := validateSCUMSchemaProbeResultForJob(job, *result.ExecutionResult.SQLiteSchemaProbe); err != nil {
+				return err
+			}
+		}
 	case domain.JobCapabilityClientManagerDeploy:
 		if result.State == domain.JobStateSucceeded && result.ExecutionResult.Kind != "client-manager.deployed" {
 			return validationError("client-manager deploy result type is invalid")
@@ -44,6 +44,9 @@ func (svc *CoreService) RequestRemoteAdapterForSession(sessionID string, request
 	if err := validator.ValidateRemoteAdapterRequest(request); err != nil {
 		return domain.RemoteAdapterResult{}, err
 	}
+	if request.Capability == domain.JobCapabilityRemoteRunDBSQLiteProbe && !request.PlatformScheduled {
+		return domain.RemoteAdapterResult{}, forbiddenError("schema probe is scheduled by Platform and is not a direct remote-adapter request")
+	}
 	instance, err := svc.GetServerInstanceForSession(sessionID, request.ServerInstanceID)
 	if err != nil {
 		return domain.RemoteAdapterResult{}, err
@@ -129,7 +132,7 @@ func isRemoteAdapterCapability(capability string) bool {
 		domain.JobCapabilityRemoteRsyncRead, domain.JobCapabilityRemoteRsyncWrite,
 		domain.JobCapabilityRemoteRunFilesRead, domain.JobCapabilityRemoteRunFilesWrite,
 		domain.JobCapabilityRemoteRunProcessStart, domain.JobCapabilityRemoteRunProcessStop,
-		domain.JobCapabilityRemoteRunDBMySQLQuery, domain.JobCapabilityRemoteRunDBSQLiteQuery,
+		domain.JobCapabilityRemoteRunDBMySQLQuery, domain.JobCapabilityRemoteRunDBSQLiteProbe, domain.JobCapabilityRemoteRunDBSQLiteQuery,
 		domain.JobCapabilityRemoteRunLogsTransfer, domain.JobCapabilityRemoteRunRCONCommand:
 		return true
 	default:
@@ -166,7 +169,7 @@ func remoteAdapterKindForCapability(capability string) domain.RemoteAdapterKind
 		return domain.RemoteAdapterRunFile
 	case domain.JobCapabilityRemoteRunProcessStart, domain.JobCapabilityRemoteRunProcessStop:
 		return domain.RemoteAdapterRunProcess
-	case domain.JobCapabilityRemoteRunDBMySQLQuery, domain.JobCapabilityRemoteRunDBSQLiteQuery:
+	case domain.JobCapabilityRemoteRunDBMySQLQuery, domain.JobCapabilityRemoteRunDBSQLiteProbe, domain.JobCapabilityRemoteRunDBSQLiteQuery:
 		return domain.RemoteAdapterDatabase
 	case domain.JobCapabilityRemoteRunRCONCommand:
 		return domain.RemoteAdapterRCON
@@ -2,6 +2,7 @@ package service

 import (
 	"encoding/json"
+	"strings"
 	"testing"

 	"browser.local/platform/domain"
@@ -86,3 +87,97 @@ func TestRemoteAdapterRequestPropagatesTypedInputsToRunJob(t *testing.T) {
 		t.Fatal("real Run claim aliases persisted remote inputs")
 	}
 }
+
+func TestSCUMSchemaProbeDispatchIsPlatformScheduledAndFenced(t *testing.T) {
+	svc := newTestCoreService()
+	plugin, endpoint := createPluginAndRunEndpoint(t, svc)
+	plugin.Permissions.RemoteAccess = true
+	plugin.DeclaredPermissions = append(plugin.DeclaredPermissions, "server.remote.access")
+	plugin.RequiredRunCapabilities = append(plugin.RequiredRunCapabilities, domain.JobCapabilityRemoteRunDBSQLiteProbe)
+	plugin.RemoteAccess = domain.GamePluginRemoteAccess{Methods: []string{"run"}, RunCapabilities: []string{domain.JobCapabilityRemoteRunDBSQLiteProbe}, DatabaseEngines: []string{"sqlite"}}
+	plugin.RuntimeProfiles.TransportProfiles = append(plugin.RuntimeProfiles.TransportProfiles, domain.RuntimeTransportProfile{Key: "scum-database", Kind: "sqlite", TargetKey: "scum-database", Capabilities: []string{domain.JobCapabilityRemoteRunDBSQLiteProbe}})
+	plugin.RuntimeProfiles.LifecycleProfiles[0].TransportKeys = append(plugin.RuntimeProfiles.LifecycleProfiles[0].TransportKeys, "scum-database")
+	plugin.SCUMLiveData = domain.SCUMLiveDataManifest{SchemaVersion: "1", Probe: domain.SCUMSchemaProbeDeclaration{Capability: domain.JobCapabilityRemoteRunDBSQLiteProbe, TargetKey: "scum-database", Bounds: domain.DefaultSCUMSchemaProbeBounds()}, CapabilityGates: []domain.SCUMLiveDataCapabilityGateDeclaration{{Capability: domain.SCUMDataCapabilitySchemaProbe, Gate: domain.SCUMCapabilityGateDisabled, AdapterVersion: "scum-live-data-v0", EvidenceStatus: domain.SCUMCapabilityEvidenceMissing, SafeReason: "waiting for current service evidence"}}}
+	if err := svc.store.GamePlugins().Update(plugin); err != nil {
+		t.Fatal(err)
+	}
+	endpoint.Capabilities = append(endpoint.Capabilities, domain.JobCapabilityRemoteRunDBSQLiteProbe)
+	if err := svc.store.RunEndpoints().Update(endpoint); err != nil {
+		t.Fatal(err)
+	}
+	session := createServiceUserAndLogin(t, svc, domain.User{ID: "user-scum-probe-owner", DisplayName: "SCUM Probe Owner", Email: "scum-probe-owner@example.test", Roles: []string{"server-owner"}, PasswordHash: "secret-password"})
+	instance, err := svc.CreateServerInstanceForSession(session, domain.ServerInstance{ID: "server-scum-probe", PluginID: plugin.ID, RunEndpointID: endpoint.ID, Name: "SCUM Probe"})
+	if err != nil {
+		t.Fatalf("create server instance: %v", err)
+	}
+	createCompleteRuntimeBinding(t, svc, instance, "local")
+
+	_, err = svc.RequestRemoteAdapterForSession(session, domain.RemoteAdapterRequest{ServerInstanceID: instance.ID, DeclarationKey: "scum-database", TargetKey: "scum-database", Capability: domain.JobCapabilityRemoteRunDBSQLiteProbe, IdempotencyKey: "direct-probe-denied", InputRef: "input://scum-schema-probe/direct-probe-denied"})
+	if err == nil || !strings.Contains(err.Error(), "scheduled by Platform") {
+		t.Fatalf("expected public probe request denial, got %v", err)
+	}
+	endpoint.Capabilities = withoutCapability(endpoint.Capabilities, domain.JobCapabilityRemoteRunDBSQLiteProbe)
+	if err := svc.store.RunEndpoints().Update(endpoint); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err = svc.RequestSCUMSchemaProbeForSession(session, instance.ID, "probe-missing-run-capability"); err == nil || !strings.Contains(err.Error(), "does not expose") {
+		t.Fatalf("expected missing active Run capability, got %v", err)
+	}
+	jobs, err := svc.store.Jobs().List(domain.JobFilter{ServerInstanceID: instance.ID})
+	if err != nil || len(jobs) != 0 {
+		t.Fatalf("missing probe executor must not create jobs: len=%d err=%v", len(jobs), err)
+	}
+
+	endpoint.Capabilities = append(endpoint.Capabilities, domain.JobCapabilityRemoteRunDBSQLiteProbe)
+	if err := svc.store.RunEndpoints().Update(endpoint); err != nil {
+		t.Fatal(err)
+	}
+	probeRequest, queued, err := svc.RequestSCUMSchemaProbeForSession(session, instance.ID, "probe-current-schema")
+	if err != nil {
+		t.Fatalf("queue SCUM schema probe: %v", err)
+	}
+	job, err := svc.store.Jobs().Get(queued.RequestID)
+	if err != nil {
+		t.Fatalf("get probe job: %v", err)
+	}
+	if job.Capability != domain.JobCapabilityRemoteRunDBSQLiteProbe || job.TargetKey != "scum-database" || job.ExecutionInput.RemoteAdapterKind != string(domain.RemoteAdapterDatabase) || job.ExecutionInput.Inputs["databaseIdentity"] != "logical:scum-database" {
+		t.Fatalf("unexpected probe job envelope: %+v", job)
+	}
+	if job.ExecutionInput.Inputs["jobId"] != job.ID || job.ExecutionInput.Inputs["requestId"] != probeRequest.RequestID || job.ExecutionInput.Inputs["maxResultBytes"] != "524288" {
+		t.Fatalf("probe inputs are not fenced and bounded: %+v", job.ExecutionInput.Inputs)
+	}
+
+	helloRequest := validRunControlHello()
+	helloRequest.CapabilityReport.Capabilities = append(helloRequest.CapabilityReport.Capabilities, domain.JobCapabilityRemoteRunDBSQLiteProbe)
+	helloRequest.CapabilityReport.Fingerprint = "cap-scum-probe"
+	hello, err := svc.RegisterRunHello(helloRequest)
+	if err != nil {
+		t.Fatalf("register Run hello: %v", err)
+	}
+	claim, err := svc.ClaimRunJob(domain.RunJobClaim{RunEndpointID: endpoint.ID, SessionToken: hello.SessionToken, Capabilities: []string{domain.JobCapabilityRemoteRunDBSQLiteProbe}, Capacity: domain.RunCapacity{MaxJobs: 1}})
+	if err != nil {
+		t.Fatalf("claim probe job: %v", err)
+	}
+	if !claim.HasJob || claim.Job == nil || claim.Job.ExecutionInput.Inputs["adapterVersion"] != "scum-live-data-v0" || claim.Job.ExecutionInput.Inputs["targetKey"] != "scum-database" {
+		t.Fatalf("claimed probe job lost typed inputs: %+v", claim.Job)
+	}
+	badBinding := probeRequest.Binding
+	badBinding.RunBindingID = "runtime-binding-other"
+	badProbe := domain.SCUMSchemaProbeResult{RequestID: probeRequest.RequestID, JobID: probeRequest.JobID, Binding: badBinding, Status: domain.SCUMCapabilityEvidenceCompatible, SchemaFingerprint: "sha256:" + strings.Repeat("a", 64), ObservedAt: fixedTime, ResultDigest: "sha256:" + strings.Repeat("b", 64), Limits: probeRequest.Bounds}
+	_, err = svc.CompleteRunJob(domain.RunJobResult{RunEndpointID: endpoint.ID, SessionToken: hello.SessionToken, JobID: claim.Job.JobID, LeaseToken: claim.Job.LeaseToken, Attempt: claim.Job.Attempt, State: domain.JobStateSucceeded, Progress: domain.RunJobProgressReport{Percent: 100, Message: "probe complete"}, Message: "probe complete", ExecutionResult: domain.JobExecutionResult{Kind: scumSchemaProbeExecutionKind, SQLiteSchemaProbe: &badProbe, AuditSummary: "redacted schema probe"}})
+	if err == nil || !strings.Contains(err.Error(), "binding identity") {
+		t.Fatalf("expected binding mismatch rejection, got %v", err)
+	}
+	goodProbe := badProbe
+	goodProbe.Binding = probeRequest.Binding
+	if _, err := svc.CompleteRunJob(domain.RunJobResult{RunEndpointID: endpoint.ID, SessionToken: hello.SessionToken, JobID: claim.Job.JobID, LeaseToken: claim.Job.LeaseToken, Attempt: claim.Job.Attempt, State: domain.JobStateSucceeded, Progress: domain.RunJobProgressReport{Percent: 100, Message: "probe complete"}, Message: "probe complete", ExecutionResult: domain.JobExecutionResult{Kind: scumSchemaProbeExecutionKind, SQLiteSchemaProbe: &goodProbe, AuditSummary: "redacted schema probe"}}); err != nil {
+		t.Fatalf("complete fenced probe job: %v", err)
+	}
+	stored, err := svc.store.Jobs().Get(job.ID)
+	if err != nil {
+		t.Fatalf("get completed probe job: %v", err)
+	}
+	if stored.ExecutionResult.SQLiteSchemaProbe == nil || stored.ExecutionResult.SQLiteSchemaProbe.ResultDigest != goodProbe.ResultDigest || stored.ExecutionResult.SQLiteSchemaProbe.Binding.RunBindingID != probeRequest.Binding.RunBindingID {
+		t.Fatalf("typed probe result was not persisted safely: %+v", stored.ExecutionResult.SQLiteSchemaProbe)
+	}
+}
@@ -124,6 +124,7 @@ type Core interface {
 	ListBackupsForSession(string, domain.BackupFilter) ([]domain.BackupRecord, error)
 	ListRemoteAdapterDeclarationsForSession(string, string) ([]domain.RemoteAdapterDeclaration, error)
 	RequestRemoteAdapterForSession(string, domain.RemoteAdapterRequest) (domain.RemoteAdapterResult, error)
+	RequestSCUMSchemaProbeForSession(string, string, string) (domain.SCUMSchemaProbeRequest, domain.RemoteAdapterResult, error)
 	GetServerConfigForSession(string, string) (domain.ServerConfig, error)
 	GetDeclaredFileReadSnapshotForSession(string, string, string) (domain.DeclaredFileReadSnapshot, error)
 	PreviewServerConfigWriteForSession(string, domain.ServerConfigDiffRequest) (domain.ServerConfigDiffPreview, error)
@@ -1167,6 +1168,11 @@ func (svc *CoreService) executeBridgeRemoteAccessRequest(sessionID string, base
 		base.Error = &domain.PluginBridgeSafeError{Code: "capability_denied", Message: "requested remote capability is not declared by plugin"}
 		return base
 	}
+	if capability == domain.JobCapabilityRemoteRunDBSQLiteProbe {
+		base.Status = "denied"
+		base.Error = &domain.PluginBridgeSafeError{Code: "capability_denied", Message: "schema probe is scheduled by Platform and is not a plugin page action"}
+		return base
+	}
 	declarationKey := strings.TrimSpace(payload["declarationKey"])
 	if declarationKey == "" {
 		for _, profile := range plugin.RuntimeProfiles.TransportProfiles {
@@ -0,0 +1,171 @@
+package service
+
+import (
+	"errors"
+	"strconv"
+	"strings"
+
+	"browser.local/platform/domain"
+	"browser.local/platform/repo"
+	"browser.local/platform/validator"
+)
+
+const scumSchemaProbeExecutionKind = "sqlite.schema.probe"
+
+func (svc *CoreService) RequestSCUMSchemaProbeForSession(sessionID, serverInstanceID, idempotencyKey string) (domain.SCUMSchemaProbeRequest, domain.RemoteAdapterResult, error) {
+	idempotencyKey = strings.TrimSpace(idempotencyKey)
+	if idempotencyKey == "" {
+		return domain.SCUMSchemaProbeRequest{}, domain.RemoteAdapterResult{}, validationError("idempotencyKey is required")
+	}
+	instance, err := svc.GetServerInstanceForSession(sessionID, serverInstanceID)
+	if err != nil {
+		return domain.SCUMSchemaProbeRequest{}, domain.RemoteAdapterResult{}, err
+	}
+	plugin, err := svc.store.GamePlugins().Get(instance.PluginID)
+	if err != nil {
+		return domain.SCUMSchemaProbeRequest{}, domain.RemoteAdapterResult{}, err
+	}
+	probe := plugin.SCUMLiveData.Probe
+	if plugin.SCUMLiveData.SchemaVersion == "" || probe.Capability != domain.JobCapabilityRemoteRunDBSQLiteProbe || strings.TrimSpace(probe.TargetKey) == "" {
+		return domain.SCUMSchemaProbeRequest{}, domain.RemoteAdapterResult{}, validationError("SCUM schema probe is not declared by the plugin")
+	}
+	endpoint, err := svc.store.RunEndpoints().Get(instance.RunEndpointID)
+	if err != nil {
+		return domain.SCUMSchemaProbeRequest{}, domain.RemoteAdapterResult{}, err
+	}
+	if endpoint.Status != domain.RunEndpointStatusOnline && endpoint.Status != domain.RunEndpointStatusDegraded {
+		return domain.SCUMSchemaProbeRequest{}, domain.RemoteAdapterResult{}, validationError("bound Run is not online for SCUM schema probe")
+	}
+	if !containsString(endpoint.Capabilities, domain.JobCapabilityRemoteRunDBSQLiteProbe) {
+		return domain.SCUMSchemaProbeRequest{}, domain.RemoteAdapterResult{}, validationError("bound Run does not expose the generic SQLite schema-probe executor")
+	}
+	binding, err := svc.runtimeBindingForServer(instance.ID)
+	if errors.Is(err, repo.ErrNotFound) {
+		return domain.SCUMSchemaProbeRequest{}, domain.RemoteAdapterResult{}, validationError("runtime binding is required before SCUM schema probe")
+	}
+	if err != nil {
+		return domain.SCUMSchemaProbeRequest{}, domain.RemoteAdapterResult{}, err
+	}
+	adapterVersion := scumSchemaProbeAdapterVersion(plugin.SCUMLiveData)
+	if adapterVersion == "" {
+		return domain.SCUMSchemaProbeRequest{}, domain.RemoteAdapterResult{}, validationError("SCUM schema-probe adapter version is not declared")
+	}
+	bounds := probe.Bounds
+	if bounds.MaxObjects == 0 {
+		bounds = domain.DefaultSCUMSchemaProbeBounds()
+	}
+	jobID := jobIDFromParts("job-remote-adapter", instance.ID, idempotencyKey)
+	request := domain.SCUMSchemaProbeRequest{
+		RequestID: jobID,
+		JobID:     jobID,
+		Binding: domain.SCUMBindingIdentity{
+			ServerInstanceID: instance.ID,
+			RunBindingID:     binding.ID,
+			RunEndpointID:    instance.RunEndpointID,
+			PluginID:         plugin.ID,
+			PluginVersion:    plugin.Version,
+			AdapterVersion:   adapterVersion,
+			DatabaseIdentity: "logical:" + probe.TargetKey,
+		},
+		Bounds:      bounds,
+		RequestedAt: svc.now(),
+	}
+	if err := validator.ValidateSCUMSchemaProbeRequest(request); err != nil {
+		return domain.SCUMSchemaProbeRequest{}, domain.RemoteAdapterResult{}, err
+	}
+	result, err := svc.RequestRemoteAdapterForSession(sessionID, domain.RemoteAdapterRequest{
+		ServerInstanceID:  instance.ID,
+		DeclarationKey:    probe.TargetKey,
+		TargetKey:         probe.TargetKey,
+		Capability:        domain.JobCapabilityRemoteRunDBSQLiteProbe,
+		TimeoutSeconds:    scumSchemaProbeTimeoutSeconds(bounds),
+		MaxAttempts:       1,
+		IdempotencyKey:    idempotencyKey,
+		InputRef:          "input://scum-schema-probe/" + instance.ID + "/" + idempotencyKey,
+		Inputs:            scumSchemaProbeInputs(request, probe.TargetKey),
+		PlatformScheduled: true,
+	})
+	if err != nil {
+		return domain.SCUMSchemaProbeRequest{}, domain.RemoteAdapterResult{}, err
+	}
+	return request, result, nil
+}
+
+func scumSchemaProbeAdapterVersion(manifest domain.SCUMLiveDataManifest) string {
+	for _, gate := range manifest.CapabilityGates {
+		if gate.Capability == domain.SCUMDataCapabilitySchemaProbe {
+			return strings.TrimSpace(gate.AdapterVersion)
+		}
+	}
+	for _, gate := range manifest.CapabilityGates {
+		if strings.TrimSpace(gate.AdapterVersion) != "" {
+			return strings.TrimSpace(gate.AdapterVersion)
+		}
+	}
+	return ""
+}
+
+func scumSchemaProbeTimeoutSeconds(bounds domain.SCUMSchemaProbeBounds) int {
+	if bounds.TimeoutMS <= 0 {
+		return 1
+	}
+	seconds := (bounds.TimeoutMS + 999) / 1000
+	if seconds <= 0 {
+		return 1
+	}
+	return seconds
+}
+
+func scumSchemaProbeInputs(request domain.SCUMSchemaProbeRequest, targetKey string) map[string]string {
+	return map[string]string{
+		"requestId":           request.RequestID,
+		"jobId":               request.JobID,
+		"targetKey":           targetKey,
+		"serverInstanceId":    request.Binding.ServerInstanceID,
+		"runBindingId":        request.Binding.RunBindingID,
+		"runEndpointId":       request.Binding.RunEndpointID,
+		"pluginId":            request.Binding.PluginID,
+		"pluginVersion":       request.Binding.PluginVersion,
+		"adapterVersion":      request.Binding.AdapterVersion,
+		"gameVersion":         request.Binding.GameVersion,
+		"databaseIdentity":    request.Binding.DatabaseIdentity,
+		"maxObjects":          strconv.Itoa(request.Bounds.MaxObjects),
+		"maxColumnsPerObject": strconv.Itoa(request.Bounds.MaxColumnsPerObject),
+		"maxIndexesPerObject": strconv.Itoa(request.Bounds.MaxIndexesPerObject),
+		"maxForeignKeys":      strconv.Itoa(request.Bounds.MaxForeignKeys),
+		"maxCardinalityReads": strconv.Itoa(request.Bounds.MaxCardinalityReads),
+		"maxSampleRows":       strconv.Itoa(request.Bounds.MaxSampleRows),
+		"timeoutMs":           strconv.Itoa(request.Bounds.TimeoutMS),
+		"maxResultBytes":      strconv.Itoa(request.Bounds.MaxResultBytes),
+	}
+}
+
+func scumSchemaProbeBindingFromInputs(inputs map[string]string) domain.SCUMBindingIdentity {
+	return domain.SCUMBindingIdentity{
+		ServerInstanceID: inputs["serverInstanceId"],
+		RunBindingID:     inputs["runBindingId"],
+		RunEndpointID:    inputs["runEndpointId"],
+		PluginID:         inputs["pluginId"],
+		PluginVersion:    inputs["pluginVersion"],
+		AdapterVersion:   inputs["adapterVersion"],
+		GameVersion:      inputs["gameVersion"],
+		DatabaseIdentity: inputs["databaseIdentity"],
+	}
+}
+
+func validateSCUMSchemaProbeResultForJob(job domain.Job, result domain.SCUMSchemaProbeResult) error {
+	if err := validator.ValidateSCUMSchemaProbeResult(result); err != nil {
+		return err
+	}
+	if result.JobID != job.ID || result.JobID != job.ExecutionInput.Inputs["jobId"] || result.RequestID != job.ExecutionInput.Inputs["requestId"] {
+		return validationError("SQLite schema probe result does not match leased job identity")
+	}
+	if !sameSCUMSchemaProbeBinding(result.Binding, scumSchemaProbeBindingFromInputs(job.ExecutionInput.Inputs)) {
+		return validationError("SQLite schema probe result does not match leased binding identity")
+	}
+	return nil
+}
+
+func sameSCUMSchemaProbeBinding(a, b domain.SCUMBindingIdentity) bool {
+	return a.ServerInstanceID == b.ServerInstanceID && a.RunBindingID == b.RunBindingID && a.RunEndpointID == b.RunEndpointID && a.PluginID == b.PluginID && a.PluginVersion == b.PluginVersion && a.AdapterVersion == b.AdapterVersion && a.GameVersion == b.GameVersion && a.DatabaseIdentity == b.DatabaseIdentity
+}