package server

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	_ "github.com/mattn/go-sqlite3"
	agsconfig "github.com/pingcap/agent-git-service/config"
	agsserver "github.com/pingcap/agent-git-service/server"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/api/roles"
	"github.com/tidbcloud/AgentTeam/internal/config"
	"github.com/tidbcloud/AgentTeam/internal/database"
	"github.com/tidbcloud/AgentTeam/internal/service/automation"
	"github.com/tidbcloud/AgentTeam/internal/service/automation/runtime"
	automationstore "github.com/tidbcloud/AgentTeam/internal/store/automation"
	"github.com/tidbcloud/AgentTeam/internal/store/content"
	"github.com/tidbcloud/AgentTeam/internal/store/kb"
	"gorm.io/gorm"
)

type seamWorkspaceChecker struct{ calls int }

func (checker *seamWorkspaceChecker) IsWorkspaceRoutableToGroup(context.Context, string, string) (bool, error) {
	checker.calls++
	return true, nil
}

func TestWorkspaceAGSSeamRemainsInsideReadinessGate(t *testing.T) {
	checker := &seamWorkspaceChecker{}
	baseHits := 0
	agsHits := 0
	server := &Server{
		cfg: config.ResourceRuntimeConfig{
			PublicBaseURL: "https://agentteam.example.test",
			Auth: config.ResourceRuntimeAuthConfig{
				Enabled: true, Header: config.DefaultTrustedJWTHeader,
				TrustedJWTClaims: true, TrustedJWTClaimsProtectedListener: true,
			},
		},
		baseHandler: http.HandlerFunc(func(http.ResponseWriter, *http.Request) { baseHits++ }),
		ags:         fakeAGS{handler: http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) { agsHits++; writer.WriteHeader(http.StatusNoContent) })},
		meshxDB:     &gorm.DB{},
	}
	server.handler = server.readinessGate(server.baseHandler)
	require.NoError(t, server.ConfigureWorkspaceAGSSeam("group_1", checker))
	request := httptest.NewRequest(http.MethodGet, "/api/v1/workspaces/wsp_1/ags/api/v3/user/repos", nil)
	request.Header.Set("X-AgentTeam-Route-Workspace-Id", "wsp_1")
	request.Header.Set("X-AgentTeam-Resolved-Workspace-Group-Id", "group_1")

	recorder := httptest.NewRecorder()
	server.Handler().ServeHTTP(recorder, request)
	require.Equal(t, http.StatusServiceUnavailable, recorder.Code)
	require.Zero(t, checker.calls)
	require.Zero(t, baseHits)
	require.Zero(t, agsHits)

	server.ready.Store(true)
	recorder = httptest.NewRecorder()
	server.Handler().ServeHTTP(recorder, request)
	require.Equal(t, http.StatusNoContent, recorder.Code)
	require.Equal(t, 1, checker.calls)
	require.Zero(t, baseHits)
	require.Equal(t, 1, agsHits)
}

func TestWorkspaceAGSSeamRequiresSharedProtectedIdentityHeader(t *testing.T) {
	t.Parallel()

	for _, test := range []struct {
		name string
		auth config.ResourceRuntimeAuthConfig
		want string
	}{
		{
			name: "custom trusted header",
			auth: config.ResourceRuntimeAuthConfig{
				Enabled: true, Header: "X-Client-Controlled-Identity",
				TrustedJWTClaims: true, TrustedJWTClaimsProtectedListener: true,
			},
			want: `configure workspace AGS seam: split deployment MESHX_AUTH_JWT_HEADER must be "X-AgentTeam-Verified-JWT"`,
		},
		{
			name: "claims are not trusted",
			auth: config.ResourceRuntimeAuthConfig{Enabled: true, Header: config.DefaultTrustedJWTHeader},
			want: "configure workspace AGS seam: trusted JWT claims on a protected listener are required",
		},
		{
			name: "listener is not protected",
			auth: config.ResourceRuntimeAuthConfig{
				Enabled: true, Header: config.DefaultTrustedJWTHeader, TrustedJWTClaims: true,
			},
			want: "configure workspace AGS seam: trusted JWT claims on a protected listener are required",
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			server := &Server{
				cfg:     config.ResourceRuntimeConfig{Auth: test.auth},
				ags:     fakeAGS{handler: http.HandlerFunc(func(http.ResponseWriter, *http.Request) { t.Fatal("untrusted identity reached AGS") })},
				meshxDB: &gorm.DB{},
			}
			require.EqualError(t, server.ConfigureWorkspaceAGSSeam("group_1", &seamWorkspaceChecker{}), test.want)
			require.Nil(t, server.handler, "failed split identity configuration must not install a handler")
		})
	}
}

type fakeAGS struct{ handler http.Handler }

func (f fakeAGS) Handler() http.Handler          { return f.handler }
func (f fakeAGS) Shutdown(context.Context) error { return nil }

type lifecycleFakeAGS struct {
	handler        http.Handler
	starts         int
	shutdowns      int
	shutdownErrors []error
}

func (f *lifecycleFakeAGS) Handler() http.Handler { return f.handler }
func (f *lifecycleFakeAGS) StartWorkers() error {
	f.starts++
	return nil
}
func (f *lifecycleFakeAGS) Shutdown(context.Context) error {
	f.shutdowns++
	if len(f.shutdownErrors) > 0 {
		err := f.shutdownErrors[0]
		f.shutdownErrors = f.shutdownErrors[1:]
		return err
	}
	return nil
}

func expectedRepoTestAGSHandler(t *testing.T) http.Handler {
	t.Helper()
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch {
		case r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, "/api/v3/repos/"):
			http.NotFound(w, r)
		case r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, "/api/v3/orgs/"):
			owner := strings.TrimPrefix(r.URL.Path, "/api/v3/orgs/")
			w.WriteHeader(http.StatusOK)
			_, _ = w.Write([]byte(`{"login":"` + owner + `"}`))
		case r.Method == http.MethodPost && strings.HasPrefix(r.URL.Path, "/api/v3/orgs/"):
			owner := strings.TrimPrefix(r.URL.Path, "/api/v3/orgs/")
			owner = strings.TrimSuffix(owner, "/repos")
			var payload map[string]any
			require.NoError(t, json.NewDecoder(r.Body).Decode(&payload))
			name, _ := payload["name"].(string)
			if name == "" {
				name = "repo"
			}
			w.WriteHeader(http.StatusCreated)
			_, _ = w.Write([]byte(`{"id":42,"name":"` + name + `","full_name":"` + owner + `/` + name + `"}`))
		case r.Method == http.MethodPut && strings.HasPrefix(r.URL.Path, "/api/v3/repos/") && strings.Contains(r.URL.Path, "/wiki/pages/"):
			w.WriteHeader(http.StatusOK)
			_, _ = w.Write([]byte(`{"slug":"fixture","sha":"sha-fixture"}`))
		default:
			http.NotFound(w, r)
		}
	})
}

func TestRoutesDelegateAGSEndpointsToEmbeddedHandler(t *testing.T) {
	called := false
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS:         config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
	}, Options{NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
		return fakeAGS{handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			called = true
			w.WriteHeader(http.StatusNoContent)
		})}, nil
	}})
	require.NoError(t, err)

	req := httptest.NewRequest(http.MethodGet, "/api/v3/meta", nil)
	req.Header.Set("Authorization", "Bearer "+jwt(map[string]any{"sub": "user_1", "preferred_username": "user-one"}))
	rec := httptest.NewRecorder()
	srv.handler.ServeHTTP(rec, req)

	require.Equal(t, http.StatusNoContent, rec.Code)
	require.True(t, called)
}

func TestContentStoreSwitchDoesNotRegisterGenericProductBinding(t *testing.T) {
	newServer := func(enabled bool) *Server {
		t.Helper()
		agentteamCoreDSN := "file:" + t.TempDir() + "/agentteam-core?mode=memory&cache=shared"
		srv, err := New(config.ResourceRuntimeConfig{
			HTTPAddress:  ":0",
			Auth:         config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
			Database:     config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
			AgentTeamCore:     config.AgentTeamCoreConfig{DatabaseDSN: agentteamCoreDSN},
			ContentStore: config.ContentStoreConfig{Enabled: enabled, FilesystemRoot: t.TempDir()},
			AGS:          config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
		}, Options{
			DisableBackgroundWorkers: true,
			NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
				return fakeAGS{handler: http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) { writer.WriteHeader(http.StatusTeapot) })}, nil
			},
		})
		require.NoError(t, err)
		srv.ready.Store(true)
		t.Cleanup(func() { require.NoError(t, srv.Shutdown(context.Background())) })
		return srv
	}
	request := func(srv *Server) *httptest.ResponseRecorder {
		body := bytes.NewBufferString(`{"domain":"test_tree","bindingKey":"fixture_1"}`)
		req := httptest.NewRequest(http.MethodPost, "/api/v1/content-store/contents", body)
		rec := httptest.NewRecorder()
		srv.Handler().ServeHTTP(rec, req)
		return rec
	}

	disabled := newServer(false)
	require.Nil(t, disabled.contentPayload)
	require.Equal(t, http.StatusNotFound, request(disabled).Code)

	enabled := newServer(true)
	require.NotNil(t, enabled.contentPayload)
	require.Equal(t, http.StatusNotFound, request(enabled).Code, "the switch opens storage but registers no generic product binding")
}

func TestRestartRejectsDisabledContentStoreWithNativeAuthorityBeforeAnyAGSRoute(t *testing.T) {
	for _, domain := range []string{"kb_source", "memory", "automation"} {
		t.Run(domain, func(t *testing.T) {
			meshxDSN := "file:" + t.TempDir() + "/meshx.db?cache=shared&_busy_timeout=5000&_journal_mode=WAL"
			agentteamCoreDSN := "file:" + t.TempDir() + "/agentteam-core?mode=memory&cache=shared"
			root := t.TempDir()
			newConfig := func(enabled bool) config.ResourceRuntimeConfig {
				return config.ResourceRuntimeConfig{
					HTTPAddress: ":0", Auth: config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
					Database: config.ResourceDatabaseConfig{DSN: meshxDSN}, AgentTeamCore: config.AgentTeamCoreConfig{DatabaseDSN: agentteamCoreDSN},
					ContentStore: config.ContentStoreConfig{Enabled: enabled, FilesystemRoot: root},
					AGS:          config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
				}
			}
			first, err := New(newConfig(true), Options{DisableBackgroundWorkers: true, NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
				return fakeAGS{handler: http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) { writer.WriteHeader(http.StatusTeapot) })}, nil
			}})
			require.NoError(t, err)
			now := time.Now().UTC()
			require.NoError(t, first.meshxDB.Create(&content.Authority{
				ID: "cta_restart_native", WorkspaceID: "wsp_1", Domain: domain, BindingKey: "binding_1",
				State: content.AuthorityStateNativeActive, Generation: 1, Checkpoint: "kb1|ctv_1|", UpdatedBy: "owner",
				CreatedAt: now, UpdatedAt: now,
			}).Error)
			require.NoError(t, first.Shutdown(context.Background()))

			agsCalls := 0
			second, err := New(newConfig(false), Options{DisableBackgroundWorkers: true, NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
				return fakeAGS{handler: http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
					agsCalls++
					writer.WriteHeader(http.StatusTeapot)
				})}, nil
			}})
			require.Nil(t, second)
			require.ErrorContains(t, err, "cannot be disabled")
			require.Zero(t, agsCalls, "restart must fail before an activated product can route to AGS")
		})
	}
}

func TestStartKBPeriodicWorkersDoNotBlockMirrorTicks(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	codeWikiStarted := make(chan struct{})
	releaseCodeWiki := make(chan struct{})
	mirrorWorkerRan := make(chan struct{})
	mirrorSchedulerRan := make(chan struct{})
	var mirrorWorkerOnce sync.Once
	var mirrorSchedulerOnce sync.Once

	done := startKBPeriodicWorkers(ctx, nil,
		kbPeriodicWorker{
			interval: time.Millisecond,
			run: func(ctx context.Context) error {
				select {
				case <-codeWikiStarted:
				default:
					close(codeWikiStarted)
				}
				select {
				case <-releaseCodeWiki:
					return nil
				case <-ctx.Done():
					return nil
				}
			},
			errMsg: "code wiki",
		},
		kbPeriodicWorker{
			interval: 5 * time.Millisecond,
			run: func(context.Context) error {
				mirrorWorkerOnce.Do(func() { close(mirrorWorkerRan) })
				return nil
			},
			errMsg: "mirror worker",
		},
		kbPeriodicWorker{
			interval: 7 * time.Millisecond,
			run: func(context.Context) error {
				mirrorSchedulerOnce.Do(func() { close(mirrorSchedulerRan) })
				return nil
			},
			errMsg: "mirror scheduler",
		},
	)

	select {
	case <-codeWikiStarted:
	case <-time.After(250 * time.Millisecond):
		t.Fatal("code wiki worker did not start")
	}

	select {
	case <-mirrorWorkerRan:
	case <-time.After(250 * time.Millisecond):
		t.Fatal("mirror worker tick did not run while code wiki worker was blocked")
	}

	select {
	case <-mirrorSchedulerRan:
	case <-time.After(250 * time.Millisecond):
		t.Fatal("mirror scheduler tick did not run while code wiki worker was blocked")
	}

	close(releaseCodeWiki)
	cancel()

	select {
	case <-done:
	case <-time.After(250 * time.Millisecond):
		t.Fatal("workers did not stop after cancellation")
	}
}

func TestEmbeddedAGSRejectsAGSEndpointWithoutJWT(t *testing.T) {
	requireCGO(t)
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS: config.AGSConfig{
			DatabaseDSN: agsFileDatabaseDSN(t),
			GitRepoDir:  t.TempDir(),
			BaseURL:     "http://127.0.0.1:3002",
		},
	})
	require.NoError(t, err)

	req := httptest.NewRequest(http.MethodGet, "/api/v3/user", nil)
	rec := httptest.NewRecorder()
	srv.handler.ServeHTTP(rec, req)

	require.Equal(t, http.StatusUnauthorized, rec.Code)
}

func TestRoutesExposeHealthWithoutAuth(t *testing.T) {
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS:         config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
	}, Options{AutomationAuthorizer: automation.AllowAutomationAuthorizer{}, DeferBackgroundWorkers: true, NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
		return fakeAGS{handler: expectedRepoTestAGSHandler(t)}, nil
	}})
	require.NoError(t, err)

	assertStatus := func(path string, expected int) {
		req := httptest.NewRequest(http.MethodGet, path, nil)
		rec := httptest.NewRecorder()
		srv.handler.ServeHTTP(rec, req)
		require.Equal(t, expected, rec.Code)
	}
	assertStatus("/healthz", http.StatusOK)
	assertStatus("/readyz", http.StatusServiceUnavailable)
	require.NoError(t, srv.StartBackgroundWorkers())
	require.NoError(t, srv.Ready(context.Background()))
	assertStatus("/readyz", http.StatusOK)
	require.NoError(t, srv.QuiesceBackgroundWorkers(context.Background()))
	assertStatus("/readyz", http.StatusServiceUnavailable)
}

func TestProductionServerMountsProtectedRoleOperationRoutes(t *testing.T) {
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth: config.ResourceRuntimeAuthConfig{
			Enabled: true, Header: config.DefaultTrustedJWTHeader,
			TrustedJWTClaims: true, TrustedJWTClaimsProtectedListener: true,
		},
		Database: config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS:      config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
	}, Options{
		DeferBackgroundWorkers: true,
		NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
			return fakeAGS{handler: http.NotFoundHandler()}, nil
		},
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, srv.Shutdown(context.Background())) })
	require.NoError(t, srv.StartBackgroundWorkers())
	require.NoError(t, srv.Ready(context.Background()))

	paths := []string{
		roles.RoleOperationPreflightPath,
		roles.RoleOperationExecutePath,
		roles.RoleOperationRecoveryPath,
	}
	internalClaims := map[string]any{
		"sub":                          "usr_internal",
		"agentteam_internal_purpose":        roles.RoleOperationInternalPurpose,
		"agent_ownership_workspace_id": "wsp_1",
		"owned_agent_ids":              []string{"agt_1"},
	}
	serve := func(path, body string, claims map[string]any) *httptest.ResponseRecorder {
		request := httptest.NewRequest(http.MethodPost, path, strings.NewReader(body))
		request.Header.Set(config.DefaultTrustedJWTHeader, "Bearer "+jwt(claims))
		request.Header.Set("X-Workspace-Id", "wsp_1")
		recorder := httptest.NewRecorder()
		srv.handler.ServeHTTP(recorder, request)
		return recorder
	}

	preflight := serve(
		roles.RoleOperationPreflightPath,
		`{"create":{"name":"Reviewer","definition":"Review production changes"}}`,
		internalClaims,
	)
	assert.Equal(t, http.StatusNoContent, preflight.Code, preflight.Body.String())

	for _, path := range []string{
		roles.RoleOperationExecutePath,
		roles.RoleOperationRecoveryPath,
	} {
		t.Run("valid internal claim reaches "+path, func(t *testing.T) {
			response := serve(path, `{}`, internalClaims)
			assert.Equal(t, http.StatusBadRequest, response.Code, response.Body.String())
		})
	}

	for _, path := range paths {
		t.Run("ordinary claim rejected before body "+path, func(t *testing.T) {
			response := serve(path, `{`, map[string]any{"sub": "usr_browser"})
			assert.Equal(t, http.StatusForbidden, response.Code, response.Body.String())
		})
		t.Run("wrong workspace claim rejected before body "+path, func(t *testing.T) {
			response := serve(path, `{`, map[string]any{
				"sub":                          "usr_internal",
				"agentteam_internal_purpose":        roles.RoleOperationInternalPurpose,
				"agent_ownership_workspace_id": "wsp_other",
			})
			assert.Equal(t, http.StatusForbidden, response.Code, response.Body.String())
		})
	}
}

func TestProductionServerDoesNotMountRoleOperationRoutesWithUntrustedJWTClaims(t *testing.T) {
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS:         config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
	}, Options{
		DeferBackgroundWorkers: true,
		NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
			return fakeAGS{handler: http.NotFoundHandler()}, nil
		},
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, srv.Shutdown(context.Background())) })
	require.NoError(t, srv.StartBackgroundWorkers())
	require.NoError(t, srv.Ready(context.Background()))

	for _, path := range []string{
		roles.RoleOperationPreflightPath,
		roles.RoleOperationExecutePath,
		roles.RoleOperationRecoveryPath,
	} {
		request := httptest.NewRequest(http.MethodPost, path, strings.NewReader(`{}`))
		request.Header.Set("Authorization", "Bearer "+jwt(map[string]any{
			"sub":                          "usr_forged",
			"agentteam_internal_purpose":        roles.RoleOperationInternalPurpose,
			"agent_ownership_workspace_id": "wsp_1",
			"owned_agent_ids":              []string{"agt_1"},
		}))
		request.Header.Set("X-Workspace-Id", "wsp_1")
		recorder := httptest.NewRecorder()

		srv.handler.ServeHTTP(recorder, request)

		require.Equal(t, http.StatusNotFound, recorder.Code, path)
	}
}

func TestWorkerOwnershipStartsGlobalMaintenanceOnOnlyDesignatedOwner(t *testing.T) {
	newServer := func(t *testing.T) (*Server, *lifecycleFakeAGS) {
		t.Helper()
		ags := &lifecycleFakeAGS{handler: expectedRepoTestAGSHandler(t)}
		srv, err := New(config.ResourceRuntimeConfig{
			HTTPAddress: ":0",
			Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
			Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
			AGS:         config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
		}, Options{
			AutomationAuthorizer:   automation.AllowAutomationAuthorizer{},
			DeferBackgroundWorkers: true,
			NewAGS:                 func(agsconfig.Config, ...agsserver.Option) (agsServer, error) { return ags, nil },
		})
		require.NoError(t, err)
		t.Cleanup(func() { require.NoError(t, srv.Shutdown(context.Background())) })
		return srv, ags
	}

	secondary, secondaryAGS := newServer(t)
	require.NoError(t, secondary.StartBackgroundWorkers(WorkerOwnership{GlobalMaintenance: false}))
	require.Zero(t, secondaryAGS.starts)

	owner, ownerAGS := newServer(t)
	require.NoError(t, owner.StartBackgroundWorkers(WorkerOwnership{GlobalMaintenance: true}))
	require.Equal(t, 1, ownerAGS.starts)
}

func TestReadyCachesSchemaCompatibilityAfterStartupGate(t *testing.T) {
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS:         config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
	}, Options{AutomationAuthorizer: automation.AllowAutomationAuthorizer{}, DeferBackgroundWorkers: true, NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
		return fakeAGS{handler: expectedRepoTestAGSHandler(t)}, nil
	}})
	require.NoError(t, err)
	checks := 0
	srv.verifySchema = func(context.Context) error { checks++; return nil }
	require.NoError(t, srv.StartBackgroundWorkers())
	require.NoError(t, srv.Ready(context.Background()))
	for range 10 {
		require.NoError(t, srv.Ready(context.Background()))
	}
	require.Equal(t, 1, checks, "readiness must not repeat manifest metadata probes")
	require.NoError(t, srv.Shutdown(context.Background()))
}

func TestReadyCanceledDuringPingDoesNotCacheSchema(t *testing.T) {
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0", Auth: config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database: config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)}, AGS: config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
	}, Options{AutomationAuthorizer: automation.AllowAutomationAuthorizer{}, DeferBackgroundWorkers: true, NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
		return fakeAGS{handler: expectedRepoTestAGSHandler(t)}, nil
	}})
	require.NoError(t, err)
	require.NoError(t, srv.StartBackgroundWorkers())
	checks := 0
	srv.verifySchema = func(context.Context) error { checks++; return nil }
	canceled, cancel := context.WithCancel(context.Background())
	cancel()
	require.ErrorIs(t, srv.Ready(canceled), context.Canceled)
	require.False(t, srv.schemaVerified)
	require.NoError(t, srv.Ready(context.Background()))
	require.True(t, srv.schemaVerified)
	require.Equal(t, 1, checks)
	require.NoError(t, srv.Shutdown(context.Background()))
}

func TestReadyFailureInsideSchemaGateDoesNotCacheAndCanRetry(t *testing.T) {
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0", Auth: config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database: config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)}, AGS: config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
	}, Options{AutomationAuthorizer: automation.AllowAutomationAuthorizer{}, DeferBackgroundWorkers: true, NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
		return fakeAGS{handler: expectedRepoTestAGSHandler(t)}, nil
	}})
	require.NoError(t, err)
	require.NoError(t, srv.StartBackgroundWorkers())
	checks := 0
	srv.verifySchema = func(context.Context) error { checks++; return context.DeadlineExceeded }
	require.ErrorIs(t, srv.Ready(context.Background()), context.DeadlineExceeded)
	require.False(t, srv.schemaVerified)
	srv.verifySchema = func(context.Context) error { checks++; return nil }
	require.NoError(t, srv.Ready(context.Background()))
	require.True(t, srv.schemaVerified)
	require.Equal(t, 2, checks)
	require.NoError(t, srv.Shutdown(context.Background()))
}

func TestReadySchemaGateWaiterCanCancelAndFailureCanRetry(t *testing.T) {
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0", Auth: config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database: config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)}, AGS: config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
	}, Options{AutomationAuthorizer: automation.AllowAutomationAuthorizer{}, DeferBackgroundWorkers: true, NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
		return fakeAGS{handler: expectedRepoTestAGSHandler(t)}, nil
	}})
	require.NoError(t, err)
	require.NoError(t, srv.StartBackgroundWorkers())
	started := make(chan struct{})
	release := make(chan struct{})
	checks := 0
	srv.verifySchema = func(context.Context) error {
		checks++
		close(started)
		<-release
		return errors.New("injected verify failure")
	}
	firstDone := make(chan error, 1)
	go func() { firstDone <- srv.Ready(context.Background()) }()
	<-started
	waitCtx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
	defer cancel()
	require.ErrorIs(t, srv.Ready(waitCtx), context.DeadlineExceeded)
	require.Equal(t, 1, checks)
	close(release)
	require.ErrorContains(t, <-firstDone, "injected verify failure")
	require.False(t, srv.schemaVerified)
	srv.verifySchema = func(context.Context) error { checks++; return nil }
	require.NoError(t, srv.Ready(context.Background()))
	require.Equal(t, 2, checks)
	require.NoError(t, srv.Shutdown(context.Background()))
}

func TestReadyAfterSchemaCacheUsesOnlyFixedDependencyPings(t *testing.T) {
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0", Auth: config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database: config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)}, AGS: config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
	}, Options{AutomationAuthorizer: automation.AllowAutomationAuthorizer{}, DeferBackgroundWorkers: true, NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
		return fakeAGS{handler: expectedRepoTestAGSHandler(t)}, nil
	}})
	require.NoError(t, err)
	require.NoError(t, srv.StartBackgroundWorkers())
	checks := 0
	srv.verifySchema = func(context.Context) error { checks++; return nil }
	require.NoError(t, srv.Ready(context.Background()))
	require.NoError(t, srv.Ready(context.Background()))
	require.NoError(t, srv.Ready(context.Background()))
	require.Equal(t, 1, checks, "cached readiness must not repeat manifest metadata queries")
	require.NoError(t, srv.Shutdown(context.Background()))
}

func TestNewRejectsTrustedClaimsWithoutProtectedListener(t *testing.T) {
	called := false
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: config.DefaultTrustedJWTHeader, TrustedJWTClaims: true},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS:         config.AGSConfig{DatabaseDSN: "file:ags.db"},
	}, Options{NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
		called = true
		return fakeAGS{handler: http.NotFoundHandler()}, nil
	}})
	require.Error(t, err)
	require.Nil(t, srv)
	require.Contains(t, err.Error(), "MESHX_AUTH_TRUSTED_JWT_CLAIMS_PROTECTED_LISTENER")
	require.False(t, called)
}

func TestRoutesExposeKnowledgeBaseControlPlaneBehindJWT(t *testing.T) {
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS:         config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
	}, Options{AutomationAuthorizer: automation.AllowAutomationAuthorizer{}, NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
		return fakeAGS{handler: expectedRepoTestAGSHandler(t)}, nil
	}})
	require.NoError(t, err)

	req := httptest.NewRequest(http.MethodPost, "/api/v1/kb", bytes.NewBufferString(`{"name":"Runbooks","description":"Ops knowledge"}`))
	req.Header.Set("Authorization", "Bearer "+jwt(map[string]any{"sub": "user-1"}))
	req.Header.Set("X-Workspace-Id", "wsp_server_kb")
	rec := httptest.NewRecorder()
	srv.handler.ServeHTTP(rec, req)

	require.Equal(t, http.StatusCreated, rec.Code)
	require.Contains(t, rec.Body.String(), `"name":"Runbooks"`)
	require.Contains(t, rec.Body.String(), `"workspaceId":"wsp_server_kb"`)
}

func TestRoutesCreateKnowledgeBaseSourceRepositoryWithEmbeddedAGSSeed(t *testing.T) {
	requireCGO(t)
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS: config.AGSConfig{
			DatabaseDSN: agsFileDatabaseDSN(t),
			GitRepoDir:  t.TempDir(),
			BaseURL:     "http://127.0.0.1:3002",
		},
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, srv.Shutdown(context.Background())) })

	workspaceID := "ws_with_underscore"
	authHeader := "Bearer " + jwt(map[string]any{"sub": "user_1", "preferred_username": "creator-one", "name": "Creator One"})
	createKB := httptest.NewRequest(http.MethodPost, "/api/v1/kb", bytes.NewBufferString(`{"name":"Runbooks","description":"Ops knowledge"}`))
	createKB.Header.Set("Authorization", authHeader)
	createKB.Header.Set("X-Workspace-Id", workspaceID)
	createKBRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(createKBRec, createKB)
	require.Equal(t, http.StatusCreated, createKBRec.Code, createKBRec.Body.String())

	var kbBody struct {
		ID string `json:"id"`
	}
	require.NoError(t, json.Unmarshal(createKBRec.Body.Bytes(), &kbBody))
	require.NotEmpty(t, kbBody.ID)

	sourceBody := `{"name":"Runbook files","sourceType":"file_repository","repoMode":"create","repo":{"name":"Runbook files","repoNameOverride":"runbook-files","description":"Runbook source files","defaultBranch":"main"},"defaultRef":"main","indexingMode":"repo_wiki_and_files"}`
	createSource := httptest.NewRequest(http.MethodPost, "/api/v1/kb/"+kbBody.ID+"/sources", bytes.NewBufferString(sourceBody))
	createSource.Header.Set("Authorization", authHeader)
	createSource.Header.Set("X-Workspace-Id", workspaceID)
	createSourceRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(createSourceRec, createSource)

	require.Equal(t, http.StatusCreated, createSourceRec.Code, createSourceRec.Body.String())

	getRepo := httptest.NewRequest(http.MethodGet, "/api/v3/repos/user_1/runbook-files", nil)
	getRepo.Header.Set("Authorization", authHeader)
	getRepoRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(getRepoRec, getRepo)
	require.Equal(t, http.StatusOK, getRepoRec.Code, getRepoRec.Body.String())
}

func TestRoutesCreateMemoryInitializesEmbeddedAGSRepository(t *testing.T) {
	requireCGO(t)
	baseURL := "http://embedded.meshx"

	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS: config.AGSConfig{
			DatabaseDSN: agsFileDatabaseDSN(t),
			GitRepoDir:  t.TempDir(),
			BaseURL:     "http://127.0.0.1:1",
		},
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, srv.Shutdown(context.Background())) })

	authHeader := "Bearer " + jwt(map[string]any{"sub": "user_1", "preferred_username": "creator-one", "name": "Creator One"})
	createMemory := httptest.NewRequest(http.MethodPost, baseURL+"/api/v1/memory", bytes.NewBufferString(`{"name":"Runbooks","description":"Ops memory","templateLocale":"en-US"}`))
	createMemory.Header.Set("Authorization", authHeader)
	createMemory.Header.Set("X-Workspace-Id", "ws_memory_seed")
	createMemoryRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(createMemoryRec, createMemory)
	require.Equal(t, http.StatusCreated, createMemoryRec.Code, createMemoryRec.Body.String())

	getTree := httptest.NewRequest(http.MethodGet, "/api/v3/repos/user_1/runbooks.mem/git/trees/main", nil)
	getTree.Header.Set("Authorization", authHeader)
	getTreeRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(getTreeRec, getTree)
	require.Equal(t, http.StatusOK, getTreeRec.Code, getTreeRec.Body.String())
	require.Contains(t, getTreeRec.Body.String(), "MEMORY.md")
}

func TestRoutesCreateRoleStoresPackageInEmbeddedAGSHiddenRepository(t *testing.T) {
	requireCGO(t)
	baseURL := "http://embedded.meshx"

	agDB := agsFileDatabaseDSN(t)
	meshxDB := "file:" + t.TempDir() + "/meshx.db?cache=shared&_busy_timeout=5000&_journal_mode=WAL&_txlock=immediate"
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDB},
		AGS: config.AGSConfig{
			DatabaseDSN: agDB,
			GitRepoDir:  t.TempDir(),
			BaseURL:     baseURL,
		},
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, srv.Shutdown(context.Background())) })

	authHeader := "Bearer " + jwt(map[string]any{"sub": "user_1", "preferred_username": "creator-one", "name": "Creator One"})
	workspaceID := "ws_role_pkg"
	createReq := httptest.NewRequest(http.MethodPost, baseURL+"/api/v1/roles", bytes.NewBufferString(`{"name":"Reviewer","definition":"review v1","rolePackageManifest":{"items":[{"type":"role_definition_asset","content":"review v1"},{"type":"portable_knowledge","name":"Rubric","content":"small diffs"}]}}`))
	createReq.Header.Set("Authorization", authHeader)
	createReq.Header.Set("X-Workspace-Id", workspaceID)
	createRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(createRec, createReq)
	require.Equal(t, http.StatusCreated, createRec.Code, createRec.Body.String())

	var created struct {
		ID                  string `json:"id"`
		RoleVersionID       string `json:"roleVersionId"`
		Definition          string `json:"definition"`
		RolePackageManifest struct {
			Items []struct {
				Type    string `json:"type"`
				Name    string `json:"name"`
				Content string `json:"content"`
			} `json:"items"`
		} `json:"rolePackageManifest"`
	}
	require.NoError(t, json.Unmarshal(createRec.Body.Bytes(), &created))
	require.Equal(t, "review v1", created.Definition)
	require.Len(t, created.RolePackageManifest.Items, 2)
	require.Equal(t, "review v1", created.RolePackageManifest.Items[0].Content)
	require.Equal(t, "small diffs", created.RolePackageManifest.Items[1].Content)

	updateReq := httptest.NewRequest(http.MethodPatch, baseURL+"/api/v1/roles/"+created.ID, bytes.NewBufferString(`{"name":"Reviewer","definition":"review v2","rolePackageManifest":{"items":[{"type":"role_definition_asset","content":"review v2"},{"type":"portable_knowledge","name":"Rubric","content":"larger diffs"}]}}`))
	updateReq.Header.Set("Authorization", authHeader)
	updateReq.Header.Set("X-Workspace-Id", workspaceID)
	updateReq.SetPathValue("role_id", created.ID)
	updateRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(updateRec, updateReq)
	require.Equal(t, http.StatusOK, updateRec.Code, updateRec.Body.String())

	var updated struct {
		RoleVersionID       string `json:"roleVersionId"`
		Definition          string `json:"definition"`
		RolePackageManifest struct {
			Items []struct {
				Type    string `json:"type"`
				Name    string `json:"name"`
				Content string `json:"content"`
			} `json:"items"`
		} `json:"rolePackageManifest"`
	}
	require.NoError(t, json.Unmarshal(updateRec.Body.Bytes(), &updated))
	require.NotEqual(t, created.RoleVersionID, updated.RoleVersionID)
	require.Equal(t, "review v2", updated.Definition)
	require.Equal(t, "larger diffs", updated.RolePackageManifest.Items[1].Content)

	getReq := httptest.NewRequest(http.MethodGet, baseURL+"/api/v1/roles/"+created.ID, nil)
	getReq.Header.Set("Authorization", authHeader)
	getReq.Header.Set("X-Workspace-Id", workspaceID)
	getReq.SetPathValue("role_id", created.ID)
	getRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(getRec, getReq)
	require.Equal(t, http.StatusOK, getRec.Code, getRec.Body.String())
	require.Contains(t, getRec.Body.String(), `"definition":"review v2"`)
	require.Contains(t, getRec.Body.String(), `"content":"larger diffs"`)

	var storedRoleVersionID string
	require.NoError(t, srv.meshxDB.Raw("SELECT active_version_id FROM roles WHERE workspace_id = ? AND id = ?", workspaceID, created.ID).Scan(&storedRoleVersionID).Error)
	require.Equal(t, updated.RoleVersionID, storedRoleVersionID)

	type manifestRow struct {
		ItemType    string
		Name        string
		Content     string
		ContentRef  string
		ContentHash string
	}
	var rows []manifestRow
	require.NoError(t, srv.meshxDB.Raw("SELECT item_type, name, content, content_ref, content_hash FROM role_package_manifest_items WHERE workspace_id = ? AND role_id = ? AND role_version_id = ? ORDER BY sort_order ASC", workspaceID, created.ID, updated.RoleVersionID).Scan(&rows).Error)
	require.Len(t, rows, 2)
	for _, row := range rows {
		require.Empty(t, row.Content)
		require.NotEmpty(t, row.ContentRef)
		require.NotEmpty(t, row.ContentHash)
		require.Contains(t, row.ContentRef, "/api/v3/repos/"+workspaceID+"/")
		require.Contains(t, row.ContentRef, ".rolepkg/contents/")
		require.Contains(t, row.ContentRef, "?ref=")
	}
	defRef := rows[0].ContentRef
	require.Contains(t, defRef, "/contents/IDENTITY.md?ref=")
	knowledgeRef := rows[1].ContentRef
	require.Contains(t, knowledgeRef, "/contents/.agents/knowledge/rubric.md?ref=")

	repoPath := rows[0].ContentRef
	repoPath = strings.Split(strings.TrimPrefix(repoPath, baseURL), "/contents/")[0]
	require.Contains(t, repoPath, "/api/v3/repos/"+workspaceID+"/")
	require.Contains(t, repoPath, ".rolepkg")

	realizeReq := httptest.NewRequest(http.MethodPost, baseURL+"/api/v1/roles/"+created.ID+"/realize", bytes.NewBufferString(`{"agentId":"agt_apply","apply":true}`))
	realizeReq.Header.Set("Authorization", authHeader)
	realizeReq.Header.Set("X-Workspace-Id", workspaceID)
	realizeReq.SetPathValue("role_id", created.ID)
	realizeRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(realizeRec, realizeReq)
	require.Equal(t, http.StatusOK, realizeRec.Code, realizeRec.Body.String())
	require.Contains(t, realizeRec.Body.String(), `"status":"realized"`)
	requireNoContentAuthorityRows(t, srv.meshxDB)
}

func TestRoutesCreateSkillInitializesEmbeddedAGSRepository(t *testing.T) {
	requireCGO(t)
	baseURL := "http://embedded.meshx"

	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS: config.AGSConfig{
			DatabaseDSN: agsFileDatabaseDSN(t),
			GitRepoDir:  t.TempDir(),
			BaseURL:     "http://127.0.0.1:1",
		},
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, srv.Shutdown(context.Background())) })

	authHeader := "Bearer " + jwt(map[string]any{"sub": "user_1", "preferred_username": "creator-one", "name": "Creator One"})
	createSkill := httptest.NewRequest(http.MethodPost, baseURL+"/api/v1/skills", bytes.NewBufferString(`{"name":"Reviewer","repoNameOverride":"reviewer","description":"Reviews code changes","templateLocale":"en"}`))
	createSkill.Header.Set("Authorization", authHeader)
	createSkill.Header.Set("X-Workspace-Id", "ws_skill_seed")
	createSkillRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(createSkillRec, createSkill)
	require.Equal(t, http.StatusCreated, createSkillRec.Code, createSkillRec.Body.String())
	var created struct {
		ID string `json:"id"`
	}
	require.NoError(t, json.Unmarshal(createSkillRec.Body.Bytes(), &created))
	require.NotEmpty(t, created.ID)

	getTree := httptest.NewRequest(http.MethodGet, "/api/v3/repos/user_1/reviewer.sk/git/trees/main?recursive=1", nil)
	getTree.Header.Set("Authorization", authHeader)
	getTreeRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(getTreeRec, getTree)
	require.Equal(t, http.StatusOK, getTreeRec.Code, getTreeRec.Body.String())
	require.Contains(t, getTreeRec.Body.String(), "AGENTS.md")
	require.Contains(t, getTreeRec.Body.String(), "skills/README.md")
	require.Contains(t, getTreeRec.Body.String(), "skills/reviewer/SKILL.md")

	deleteSkill := httptest.NewRequest(http.MethodDelete, baseURL+"/api/v1/skills/"+created.ID, nil)
	deleteSkill.Header.Set("Authorization", authHeader)
	deleteSkill.Header.Set("X-Workspace-Id", "ws_skill_seed")
	deleteSkill.SetPathValue("skill_id", created.ID)
	deleteSkillRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(deleteSkillRec, deleteSkill)
	require.Equal(t, http.StatusNoContent, deleteSkillRec.Code, deleteSkillRec.Body.String())
	requireNoContentAuthorityRows(t, srv.meshxDB)
}

func requireNoContentAuthorityRows(t *testing.T, db *gorm.DB) {
	t.Helper()
	var count int64
	require.NoError(t, db.Table("content_authorities").Count(&count).Error)
	require.Zero(t, count, "disabled Content Store must not materialize authority rows")
	require.False(t, db.Migrator().HasTable("content_authority_writers"))
	require.False(t, db.Migrator().HasTable("content_authority_recovery_points"))
}

func TestRoutesExposeAgentAccessibleKnowledgeBasesByAgentPermission(t *testing.T) {
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS:         config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
	}, Options{AutomationAuthorizer: automation.AllowAutomationAuthorizer{}, NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
		return fakeAGS{handler: expectedRepoTestAGSHandler(t)}, nil
	}})
	require.NoError(t, err)

	now := time.Date(2026, 6, 8, 1, 0, 0, 0, time.UTC)
	require.NoError(t, srv.meshxDB.Create(&kb.KnowledgeBase{ID: "kb_agent_allowed", WorkspaceID: "wsp_agent_kb", Name: "Runbooks", Description: "Ops", IsActive: true, CreatedBy: "owner", UpdatedBy: "owner", CreatedAt: now, UpdatedAt: now}).Error)
	require.NoError(t, srv.meshxDB.Create(&kb.KnowledgeBase{ID: "kb_agent_denied", WorkspaceID: "wsp_agent_kb", Name: "Secrets", Description: "No", IsActive: true, CreatedBy: "owner", UpdatedBy: "owner", CreatedAt: now, UpdatedAt: now}).Error)
	require.NoError(t, srv.meshxDB.Create(&kb.KnowledgeBaseGrant{ID: "kbg_agent_reader", WorkspaceID: "wsp_agent_kb", KnowledgeBaseID: "kb_agent_allowed", SubjectType: "agent", SubjectID: "agt_reader", Role: "view", CreatedBy: "admin", UpdatedBy: "admin", CreatedAt: now, UpdatedAt: now}).Error)

	req := httptest.NewRequest(http.MethodGet, "/api/v1/kb/agents/agt_reader/accessible", nil)
	req.Header.Set("Authorization", "Bearer "+jwt(map[string]any{"sub": "acting-user", "agent_ownership_workspace_id": "wsp_agent_kb", "owned_agent_ids": []string{"agt_reader"}}))
	req.Header.Set("X-Workspace-Id", "wsp_agent_kb")
	rec := httptest.NewRecorder()
	srv.handler.ServeHTTP(rec, req)

	require.Equal(t, http.StatusOK, rec.Code)
	require.Contains(t, rec.Body.String(), `"id":"kb_agent_allowed"`)
	require.NotContains(t, rec.Body.String(), "kb_agent_denied")
	require.Contains(t, rec.Body.String(), `"accessLevel":"read"`)
}

func TestToAGSConfigMapsMeshXConfig(t *testing.T) {
	cfg := toAGSConfig(config.ResourceRuntimeConfig{
		HTTPAddress: ":3002",
		AGS: config.AGSConfig{
			DatabaseDSN:         "root@tcp(127.0.0.1:4000)/meshx_ags",
			GitRepoDir:          ".data/repos",
			BaseURL:             "http://127.0.0.1:3002",
			AdminLogin:          "admin",
			AdminToken:          "token",
			AllowAnyToken:       true,
			EmbeddingAPIKey:     "test-key",
			EmbeddingBaseURL:    "https://embedding.example/v1",
			EmbeddingModel:      "text-embedding-3-small",
			EmbeddingDimensions: 1536,
		},
	})
	require.Equal(t, "3002", cfg.Port)
	require.Equal(t, "root@tcp(127.0.0.1:4000)/meshx_ags", cfg.DBdsn)
	require.Equal(t, ".data/repos", cfg.GitRepoDir)
	require.Equal(t, "http://127.0.0.1:3002", cfg.BaseURL)
	require.Equal(t, "http://127.0.0.1:3002/login/device", cfg.OAuthDeviceVerificationURL)
	require.Equal(t, "production", cfg.ListenMode)
	require.Equal(t, "development", cfg.Environment)
	require.True(t, cfg.AllowAnyToken)
	require.Equal(t, "test-key", cfg.EmbeddingAPIKey)
	require.Equal(t, "https://embedding.example/v1", cfg.EmbeddingBaseURL)
	require.Equal(t, "text-embedding-3-small", cfg.EmbeddingModel)
	require.Equal(t, 1536, cfg.EmbeddingDimensions)
}

func TestToAGSConfigExtractsPortFromHostAddress(t *testing.T) {
	cfg := toAGSConfig(config.ResourceRuntimeConfig{
		HTTPAddress: "127.0.0.1:3002",
		AGS: config.AGSConfig{
			DatabaseDSN: "root@tcp(127.0.0.1:4000)/meshx_ags",
			GitRepoDir:  ".data/repos",
			BaseURL:     "http://127.0.0.1:3002",
		},
	})
	require.Equal(t, "3002", cfg.Port)
}

func TestStartBackgroundWorkersIsIdempotent(t *testing.T) {
	server := &Server{}
	require.NoError(t, server.StartBackgroundWorkers())
	require.NoError(t, server.StartBackgroundWorkers())
}

func TestStartBackgroundWorkersRejectsStartAfterQuiesce(t *testing.T) {
	server := &Server{}
	require.NoError(t, server.QuiesceBackgroundWorkers(context.Background()))
	require.EqualError(t, server.StartBackgroundWorkers(), "start background workers: workers were already stopped")
}

func TestStartBackgroundWorkersLeavesServerUnstartedAfterFailure(t *testing.T) {
	attempts := 0
	server := &Server{startBackgroundWorkers: func() error {
		attempts++
		return errors.New("bridge start failed")
	}}
	require.EqualError(t, server.StartBackgroundWorkers(), "start background workers: bridge start failed")
	require.False(t, server.workersStarted)
	require.False(t, server.ready.Load())
	require.EqualError(t, server.StartBackgroundWorkers(), "start background workers: workers were already stopped")
	require.Equal(t, 1, attempts)
}

func TestStartWorkerStagesRollBackPartialStartupInReverseOrder(t *testing.T) {
	for _, failAt := range []int{1, 2, 3} {
		t.Run(fmt.Sprintf("stage_%d", failAt), func(t *testing.T) {
			var started, rolledBack []string
			stages := make([]workerStartStage, 4)
			for i := range stages {
				name := fmt.Sprintf("stage-%d", i)
				stages[i] = workerStartStage{
					name:  name,
					start: func() error { started = append(started, name); return nil },
					rollback: func(context.Context) error {
						rolledBack = append(rolledBack, name)
						return nil
					},
				}
			}
			err := startWorkerStages(stages, func(name string) error {
				if name == fmt.Sprintf("stage-%d", failAt) {
					return errors.New("injected failure")
				}
				return nil
			})
			require.ErrorContains(t, err, "injected failure")
			require.Len(t, started, failAt)
			for i := 0; i < failAt; i++ {
				require.Equal(t, fmt.Sprintf("stage-%d", failAt-1-i), rolledBack[i])
			}
		})
	}
}

func TestServerWorkerStartupFailureMatrixRollsBackOwnedComponents(t *testing.T) {
	requireCGO(t)
	for _, failStage := range []string{"automation-runtime", "kb-periodic-workers", "content-outbox"} {
		t.Run(failStage, func(t *testing.T) {
			dir := t.TempDir()
			agsDSN := "file:" + dir + "/ags.db?cache=shared&_busy_timeout=5000&_journal_mode=WAL&_txlock=immediate"
			meshxDSN := "file:" + dir + "/meshx.db?cache=shared&_busy_timeout=5000&_journal_mode=WAL&_txlock=immediate"
			agentteamCoreDSN := "file:" + dir + "/agentteam-core.db?cache=shared&_busy_timeout=5000&_journal_mode=WAL&_txlock=immediate"
			require.NoError(t, agsserver.MigrateDatabase(agsDSN, 0))
			meshxSeed, err := database.Open(database.Config{DSN: meshxDSN})
			require.NoError(t, err)
			require.NoError(t, database.Close(meshxSeed))
			agentteamCoreSeed, err := sql.Open("sqlite3", agentteamCoreDSN)
			require.NoError(t, err)
			require.NoError(t, agentteamCoreSeed.PingContext(context.Background()))
			require.NoError(t, agentteamCoreSeed.Close())

			srv, err := New(config.ResourceRuntimeConfig{
				HTTPAddress: ":0",
				Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
				Database:    config.ResourceDatabaseConfig{DSN: meshxDSN, DisableAutoMigrate: true},
				AgentTeamCore:    config.AgentTeamCoreConfig{DatabaseDSN: agentteamCoreDSN},
				AGS: config.AGSConfig{
					DatabaseDSN: agsDSN,
					GitRepoDir:  dir + "/repos",
					BaseURL:     "http://127.0.0.1:3002",
				},
			}, Options{
				AutomationAuthorizer:   automation.AllowAutomationAuthorizer{},
				DeferBackgroundWorkers: true,
				workerStartStageHook: func(stage string) error {
					if stage == failStage {
						return errors.New("injected " + stage + " failure")
					}
					return nil
				},
			})
			require.NoError(t, err)
			agsRuntime, ok := srv.ags.(*agsserver.Server)
			require.True(t, ok)
			meshxSQL, err := srv.meshxDB.DB()
			require.NoError(t, err)
			agentteamCoreSQL, err := srv.agentteamCoreDB.DB()
			require.NoError(t, err)
			err = srv.StartBackgroundWorkers()
			require.ErrorContains(t, err, "injected "+failStage+" failure")
			require.False(t, srv.workersStarted)
			require.True(t, srv.workersStopped)
			require.False(t, srv.ready.Load())
			require.Error(t, agsRuntime.Ready(context.Background()), "real AGS database remained open after rollback")
			require.NoError(t, meshxSQL.PingContext(context.Background()), "MeshX DB closes only with the public runtime")
			require.NoError(t, agentteamCoreSQL.PingContext(context.Background()), "AgentTeam-core DB closes only with the public runtime")
			if failStage != "automation-runtime" {
				if srv.runtimeDone != nil {
					select {
					case <-srv.runtimeDone:
					default:
						t.Fatal("automation runtime goroutines remained after rollback")
					}
				}
			}
			require.NoError(t, srv.Shutdown(context.Background()))
			require.Error(t, meshxSQL.PingContext(context.Background()), "MeshX database remained open after public shutdown")
			require.Error(t, agentteamCoreSQL.PingContext(context.Background()), "AgentTeam-core database remained open after public shutdown")
			require.Error(t, agsRuntime.Ready(context.Background()), "AGS database reopened after public shutdown")
		})
	}
}

func TestShutdownDoesNotCloseThreeDatabasesBeforeDrainAndRetryClosesThem(t *testing.T) {
	requireCGO(t)
	dir := t.TempDir()
	agsDSN := "file:" + dir + "/ags.db?cache=shared&_busy_timeout=5000&_journal_mode=WAL&_txlock=immediate"
	meshxDSN := "file:" + dir + "/meshx.db?cache=shared&_busy_timeout=5000&_journal_mode=WAL&_txlock=immediate"
	agentteamCoreDSN := "file:" + dir + "/agentteam-core.db?cache=shared&_busy_timeout=5000&_journal_mode=WAL&_txlock=immediate"
	require.NoError(t, agsserver.MigrateDatabase(agsDSN, 0))
	meshxSeed, err := database.Open(database.Config{DSN: meshxDSN})
	require.NoError(t, err)
	require.NoError(t, database.Close(meshxSeed))
	agentteamCoreSeed, err := sql.Open("sqlite3", agentteamCoreDSN)
	require.NoError(t, err)
	require.NoError(t, agentteamCoreSeed.PingContext(context.Background()))
	require.NoError(t, agentteamCoreSeed.Close())

	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDSN, DisableAutoMigrate: true},
		AgentTeamCore:    config.AgentTeamCoreConfig{DatabaseDSN: agentteamCoreDSN},
		AGS: config.AGSConfig{
			DatabaseDSN: agsDSN,
			GitRepoDir:  dir + "/repos",
			BaseURL:     "http://127.0.0.1:3002",
		},
	}, Options{AutomationAuthorizer: automation.AllowAutomationAuthorizer{}, DeferBackgroundWorkers: true})
	require.NoError(t, err)
	require.NoError(t, srv.StartBackgroundWorkers())
	agsRuntime, ok := srv.ags.(*agsserver.Server)
	require.True(t, ok)
	meshxSQL, err := srv.meshxDB.DB()
	require.NoError(t, err)
	agentteamCoreSQL, err := srv.agentteamCoreDB.DB()
	require.NoError(t, err)

	started, release := installBlockingRuntimeDrain(t, srv)
	originalUnregister := srv.unregisterGit
	unregisters := 0
	srv.unregisterGit = func() { unregisters++; originalUnregister() }
	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond)
	defer cancel()
	require.Error(t, srv.Shutdown(ctx))
	<-started
	require.Zero(t, unregisters)
	require.False(t, srv.workersDrained)
	require.NoError(t, agsRuntime.Ready(context.Background()))
	require.NoError(t, meshxSQL.PingContext(context.Background()))
	require.NoError(t, agentteamCoreSQL.PingContext(context.Background()))
	close(release)
	require.NoError(t, srv.Shutdown(context.Background()))
	require.Equal(t, 1, unregisters)
	require.True(t, srv.workersDrained)
	require.Error(t, agsRuntime.Ready(context.Background()))
	require.Error(t, meshxSQL.PingContext(context.Background()))
	require.Error(t, agentteamCoreSQL.PingContext(context.Background()))
	require.NoError(t, srv.Shutdown(context.Background()))
	require.Equal(t, 1, unregisters, "repeated shutdown must not close dependencies twice")
}

func TestShutdownRetiresHandlerBeforeDependencyCloseAndDoesNotRestoreItOnRetry(t *testing.T) {
	ags := &lifecycleFakeAGS{
		handler:        http.NotFoundHandler(),
		shutdownErrors: []error{errors.New("injected AGS shutdown failure")},
	}
	srv := &Server{ags: ags, workersDrained: true}
	srv.handler = srv.readinessGate(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
		writer.WriteHeader(http.StatusNoContent)
	}))
	srv.ready.Store(true)
	captured := srv.Handler()
	require.NotNil(t, captured)

	recorder := httptest.NewRecorder()
	captured.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/api/test", nil))
	require.Equal(t, http.StatusNoContent, recorder.Code)

	require.ErrorContains(t, srv.Shutdown(context.Background()), "injected AGS shutdown failure")
	require.Nil(t, srv.Handler())
	require.ErrorContains(t, srv.Ready(context.Background()), "handler is not configured")
	for _, path := range []string{"/api/test", "/readyz"} {
		recorder = httptest.NewRecorder()
		captured.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, path, nil))
		require.Equal(t, http.StatusServiceUnavailable, recorder.Code, path)
	}

	require.NoError(t, srv.Shutdown(context.Background()))
	require.Nil(t, srv.Handler())
	require.NoError(t, srv.Shutdown(context.Background()))
	require.Nil(t, srv.Handler())
	require.Equal(t, 2, ags.shutdowns, "completed shutdown must not close dependencies twice")
}

func TestReadyCannotRepublishAfterShutdownStartsDraining(t *testing.T) {
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS:         config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
	}, Options{
		AutomationAuthorizer:   automation.AllowAutomationAuthorizer{},
		DeferBackgroundWorkers: true,
		NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
			return fakeAGS{handler: http.NotFoundHandler()}, nil
		},
	})
	require.NoError(t, err)
	defer func() { _ = srv.Shutdown(context.Background()) }()

	verifyStarted := make(chan struct{})
	releaseVerify := make(chan struct{})
	var releaseVerifyOnce sync.Once
	releaseVerifyFn := func() { releaseVerifyOnce.Do(func() { close(releaseVerify) }) }
	defer releaseVerifyFn()
	srv.verifySchema = func(context.Context) error {
		close(verifyStarted)
		<-releaseVerify
		return nil
	}

	require.NoError(t, srv.StartBackgroundWorkers())
	drainStarted, releaseDrain := installBlockingRuntimeDrain(t, srv)
	var releaseDrainOnce sync.Once
	releaseDrainFn := func() { releaseDrainOnce.Do(func() { close(releaseDrain) }) }
	defer releaseDrainFn()
	captured := srv.Handler()
	require.NotNil(t, captured)

	readyDone := make(chan error, 1)
	go func() { readyDone <- srv.Ready(context.Background()) }()
	<-verifyStarted
	shutdownDone := make(chan error, 1)
	go func() { shutdownDone <- srv.Shutdown(context.Background()) }()
	<-drainStarted
	require.Eventually(t, func() bool {
		srv.workersMu.Lock()
		defer srv.workersMu.Unlock()
		return srv.retiring && srv.workersStopped
	}, time.Second, time.Millisecond)

	assertCapturedUnavailable := func(path string) {
		recorder := httptest.NewRecorder()
		captured.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, path, nil))
		require.Equal(t, http.StatusServiceUnavailable, recorder.Code, path)
	}
	assertCapturedUnavailable("/api/test")
	assertCapturedUnavailable("/readyz")

	releaseVerifyFn()
	require.ErrorContains(t, <-readyDone, "workers are not running")
	assertCapturedUnavailable("/api/test")
	assertCapturedUnavailable("/readyz")

	releaseDrainFn()
	require.NoError(t, <-shutdownDone)
	require.Nil(t, srv.Handler())
	assertCapturedUnavailable("/api/test")
	assertCapturedUnavailable("/readyz")
	require.NoError(t, srv.Shutdown(context.Background()))
}

func installBlockingRuntimeDrain(t *testing.T, srv *Server) (<-chan struct{}, chan struct{}) {
	t.Helper()
	if srv.runtimeCancel != nil {
		srv.runtimeCancel()
		if srv.runtimeDone != nil {
			<-srv.runtimeDone
		}
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	started := make(chan struct{})
	release := make(chan struct{})
	srv.runtimeCancel, srv.runtimeDone = cancel, done
	go func() {
		<-ctx.Done()
		close(started)
		<-release
		close(done)
	}()
	return started, release
}

func TestAutomationServerRunnerDockerFailsStartupWithoutImages(t *testing.T) {
	_, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS:         config.AGSConfig{DatabaseDSN: "file:ags.db"},
		Automation: config.AutomationConfig{
			Roles:               []string{config.AutomationRoleRunnerDispatch},
			ServerRunnerEnabled: true,
			ServerRunnerWorkDir: t.TempDir(),
			Executor:            config.AutomationExecutorDocker,
		},
	}, Options{NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
		return fakeAGS{handler: http.NotFoundHandler()}, nil
	}})
	require.Error(t, err)
	require.Contains(t, err.Error(), "MESHX_AUTOMATION_DOCKER_IMAGES")
}

func TestAutomationServerRunnerSkipsWhenDockerUnavailable(t *testing.T) {
	runner, err := automationServerRunnerOption(
		config.AutomationConfig{
			Roles:               []string{config.AutomationRoleRunnerDispatch},
			ServerRunnerEnabled: true,
		},
		nil,
		nil,
		nil,
		nil,
		nil,
		nil,
		[]Options{{
			AutomationJobExecutorResolver: func(config.AutomationConfig, []Options) (runtime.JobExecutor, []string, error) {
				return nil, nil, fmt.Errorf("%w: connect to Docker: daemon unavailable", runtime.ErrDockerUnavailable)
			},
		}},
	)

	require.NoError(t, err)
	require.Nil(t, runner)
}

func TestAutomationServerRunnerKeepsNonDockerStartupErrorsFatal(t *testing.T) {
	runner, err := automationServerRunnerOption(
		config.AutomationConfig{
			Roles:               []string{config.AutomationRoleRunnerDispatch},
			ServerRunnerEnabled: true,
		},
		nil,
		nil,
		nil,
		nil,
		nil,
		nil,
		[]Options{{
			AutomationJobExecutorResolver: func(config.AutomationConfig, []Options) (runtime.JobExecutor, []string, error) {
				return nil, nil, errors.New("bad runner config")
			},
		}},
	)

	require.Error(t, err)
	require.Nil(t, runner)
	require.Contains(t, err.Error(), "bad runner config")
}

func TestAutomationServerRunnerAcceptsInjectedExecutorForTests(t *testing.T) {
	executor := &serverFakeJobExecutor{}
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS:         config.AGSConfig{DatabaseDSN: "file:ags.db", GitRepoDir: t.TempDir()},
		Automation: config.AutomationConfig{
			Roles:               []string{config.AutomationRoleRunnerDispatch},
			ServerRunnerEnabled: true,
			ServerRunnerWorkDir: t.TempDir(),
			ServerRunnerLabels:  []string{"ubuntu-latest"},
		},
	}, Options{
		NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
			return fakeAGS{handler: http.NotFoundHandler()}, nil
		},
		AutomationJobExecutor: executor,
	})
	require.NoError(t, err)
	require.False(t, executor.closed)
	require.NoError(t, srv.Shutdown(context.Background()))
	require.True(t, executor.closed)
}

type serverFakeJobExecutor struct {
	closed bool
}

func (*serverFakeJobExecutor) ExecuteJob(context.Context, runtime.ExecuteJobRequest) (runtime.ExecuteJobResult, error) {
	return runtime.ExecuteJobResult{}, nil
}

func (e *serverFakeJobExecutor) Close() error {
	e.closed = true
	return nil
}

func TestEmbeddedAGSAcceptsAgentTeamJWTThroughDefaultResolver(t *testing.T) {
	requireCGO(t)
	dsn := agsFileDatabaseDSN(t)
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS: config.AGSConfig{
			DatabaseDSN: dsn,
			GitRepoDir:  t.TempDir(),
			BaseURL:     "http://127.0.0.1:3002",
		},
	})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, srv.Shutdown(context.Background())) })

	req := httptest.NewRequest(http.MethodGet, "/api/v3/user", nil)
	req.Header.Set("Authorization", "Bearer "+jwt(map[string]any{
		"sub":                "casdoor|123",
		"email":              "meshx-user@example.com",
		"name":               "MeshX User",
		"preferred_username": "meshx-user",
	}))
	rec := httptest.NewRecorder()
	srv.handler.ServeHTTP(rec, req)

	require.Equal(t, http.StatusOK, rec.Code, rec.Body.String())
	require.Contains(t, rec.Body.String(), `"email":"meshx-user@example.com"`)
}

func requireCGO(t *testing.T) {
	t.Helper()
	if !cgoEnabled {
		t.Skip("embedded AGS SQLite integration requires cgo-backed go-sqlite3")
	}
}

func meshxDatabaseDSN(t *testing.T) string {
	t.Helper()
	// -count reruns use the same test name in one process. A name-derived shared
	// in-memory DSN can therefore retain rows through a still-closing connection
	// and contaminate the next run. A per-call temp namespace keeps the shared
	// connection semantics inside one server while isolating repeated runs.
	return "file:" + t.TempDir() + "/meshx?mode=memory&cache=shared"
}

// agsFileDatabaseDSN returns a file-backed SQLite DSN tuned for the embedded
// AGS integration tests, where the AGS HTTP server and the test harness race
// to write labels/repos into the same database. Adding _busy_timeout,
// _journal_mode=WAL, and _txlock=immediate prevents SQLITE_BUSY/LOCKED errors
// from turning the tests flaky on slow CI I/O. cache=shared stays so multiple
// go-sqlite3 connections in the same process keep working.
func agsFileDatabaseDSN(t *testing.T) string {
	t.Helper()
	return "file:" + t.TempDir() + "/ags.db?cache=shared&_busy_timeout=5000&_journal_mode=WAL&_txlock=immediate"
}

func prepareResolverDB(t *testing.T) string {
	t.Helper()
	requireCGO(t)
	dsn := "file:" + t.TempDir() + "/bridge.db?cache=shared&_busy_timeout=5000&_journal_mode=WAL&_txlock=immediate"
	db, err := sql.Open("sqlite3", dsn)
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, db.Close()) })
	_, err = db.Exec(`
CREATE TABLE users (
  id integer PRIMARY KEY AUTOINCREMENT,
  login text UNIQUE NOT NULL,
  name text,
  email text,
  type text,
  status text,
  user_kind text,
  site_admin boolean,
  default_repository_permission text,
  is_anonymous boolean,
  hide_presence boolean,
  created_at datetime,
  updated_at datetime
);
CREATE TABLE tokens (
  id integer PRIMARY KEY AUTOINCREMENT,
  user_id integer,
  name text,
  value text UNIQUE NOT NULL,
  created_at datetime,
  last_used_at datetime,
  expires_at datetime
);`)
	require.NoError(t, err)
	return dsn
}

func sqliteTableExists(t *testing.T, dsn string, table string) bool {
	t.Helper()
	db, err := sql.Open("sqlite3", dsn)
	require.NoError(t, err)
	defer func() { require.NoError(t, db.Close()) }()
	var count int
	require.NoError(t, db.QueryRow(`SELECT count(*) FROM sqlite_master WHERE type = 'table' AND name = ?`, table).Scan(&count))
	return count > 0
}

func jwt(claims map[string]any) string {
	header := base64.RawURLEncoding.EncodeToString([]byte(`{"alg":"none"}`))
	payload, _ := json.Marshal(claims)
	return header + "." + base64.RawURLEncoding.EncodeToString(payload) + "."
}

func TestAutomationsAPIReadsRunLedger(t *testing.T) {
	srv, err := New(config.ResourceRuntimeConfig{
		HTTPAddress: ":0",
		Auth:        config.ResourceRuntimeAuthConfig{Enabled: true, Header: "Authorization"},
		Database:    config.ResourceDatabaseConfig{DSN: meshxDatabaseDSN(t)},
		AGS:         config.AGSConfig{DatabaseDSN: prepareResolverDB(t), GitRepoDir: t.TempDir()},
	}, Options{AutomationAuthorizer: automation.AllowAutomationAuthorizer{}, NewAGS: func(agsconfig.Config, ...agsserver.Option) (agsServer, error) {
		return fakeAGS{handler: expectedRepoTestAGSHandler(t)}, nil
	}})
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, srv.Shutdown(context.Background())) })

	now := time.Date(2026, 6, 3, 0, 0, 0, 0, time.UTC)
	started := now
	created := automationstore.AutomationRun{ID: "wfr_ledger", WorkspaceID: "ws-automations", SourceKind: "agent_git_service_repo", RepositoryOwner: "pingcap", RepositoryName: "agentteam", RepositoryFullName: "pingcap/agentteam", DefinitionPath: ".agentteam/review.yml", GitRef: "refs/heads/main", CommitSHA: "0123456789abcdef0123456789abcdef01234567", DefinitionID: "ags:pingcap/agentteam:.agentteam/review.yml", Name: "Review automation", Status: automationstore.AutomationRunStatusRunning, TriggerKind: "task", TriggerRef: "task:#agentteam-in-agentteam:1601", TargetKind: "pr", TargetRef: "pingcap/agentteam#1539", OwnerPrincipalID: "user-1", CreatedByPrincipalID: "user-1", CurrentJobID: "plan", Summary: "MeshX ledger started", Metadata: `{"source":"ags"}`, StartedAt: &started, CreatedAt: now, UpdatedAt: now}
	_, err = automationstore.New(srv.meshxDB).CreateRun(context.Background(), created)
	require.NoError(t, err)
	require.True(t, srv.meshxDB.Migrator().HasTable(&automationstore.AutomationRun{}))
	require.True(t, srv.meshxDB.Migrator().HasTable(&automationstore.AutomationEvent{}))
	require.False(t, srv.meshxDB.Migrator().HasTable("automation_run_events"))

	auth := "Bearer " + jwt(map[string]any{"sub": "user-1"})
	getReq := httptest.NewRequest(http.MethodGet, "/api/v1/automations/runs/"+created.ID, nil)
	getReq.Header.Set("Authorization", auth)
	getReq.Header.Set("X-Workspace-Id", "ws-automations")
	getRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(getRec, getReq)
	require.Equal(t, http.StatusOK, getRec.Code, getRec.Body.String())
	require.Contains(t, getRec.Body.String(), `"repositoryFullName":"pingcap/agentteam"`)
	require.Contains(t, getRec.Body.String(), `"currentJobId":"plan"`)

	ledgerStore := automationstore.New(srv.meshxDB)
	require.NoError(t, ledgerStore.CreateJobRuns(context.Background(), []automationstore.AutomationJobRun{{ID: "wfj_plan", WorkspaceID: "ws-automations", RunID: created.ID, JobID: "plan", JobType: "github_actions_job", Status: automationstore.AutomationJobRunStatusSucceeded, Attempt: 1, MaxAttempts: 3, InputJSON: `{"steps":[{"run":"echo plan"}]}`, OutputJSON: `{"accepted":true}`, ScheduledAt: now, StartedAt: &started, CompletedAt: &started, CreatedAt: now, UpdatedAt: now}}))
	_, err = ledgerStore.CreateEvent(context.Background(), automationstore.AutomationEvent{ID: "wfe_ledger", WorkspaceID: "ws-automations", RunID: created.ID, JobRunID: "wfj_plan", EventType: "job_completed", ActorPrincipalID: "runner-1", ResourceType: "task", ResourceID: "task:#agentteam-in-agentteam:1601", Summary: "Plan accepted", MetadataJSON: `{"job_id":"plan"}`, CreatedAt: now.Add(time.Second)})
	require.NoError(t, err)

	jobRunsReq := httptest.NewRequest(http.MethodGet, "/api/v1/automations/runs/"+created.ID+"/job-runs", nil)
	jobRunsReq.Header.Set("Authorization", auth)
	jobRunsReq.Header.Set("X-Workspace-Id", "ws-automations")
	jobRunsRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(jobRunsRec, jobRunsReq)
	require.Equal(t, http.StatusOK, jobRunsRec.Code, jobRunsRec.Body.String())
	var jobRuns []map[string]any
	require.NoError(t, json.Unmarshal(jobRunsRec.Body.Bytes(), &jobRuns))
	require.Len(t, jobRuns, 1)
	require.Equal(t, "plan", jobRuns[0]["jobId"])
	require.Equal(t, "github_actions_job", jobRuns[0]["jobType"])
	require.Equal(t, "succeeded", jobRuns[0]["status"])

	listReq := httptest.NewRequest(http.MethodGet, "/api/v1/automations/runs?status=running&q=review", nil)
	listReq.Header.Set("Authorization", auth)
	listReq.Header.Set("X-Workspace-Id", "ws-automations")
	listRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(listRec, listReq)
	require.Equal(t, http.StatusOK, listRec.Code, listRec.Body.String())
	var runs struct {
		Items   []map[string]any `json:"items"`
		HasMore bool             `json:"has_more"`
	}
	require.NoError(t, json.Unmarshal(listRec.Body.Bytes(), &runs))
	require.Len(t, runs.Items, 1)
	require.False(t, runs.HasMore)
	require.Equal(t, created.ID, runs.Items[0]["id"])

	eventsReq := httptest.NewRequest(http.MethodGet, "/api/v1/automations/runs/"+created.ID+"/events", nil)
	eventsReq.Header.Set("Authorization", auth)
	eventsReq.Header.Set("X-Workspace-Id", "ws-automations")
	eventsRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(eventsRec, eventsReq)
	require.Equal(t, http.StatusOK, eventsRec.Code, eventsRec.Body.String())
	var events []map[string]any
	require.NoError(t, json.Unmarshal(eventsRec.Body.Bytes(), &events))
	require.Len(t, events, 1)
	require.Equal(t, "Plan accepted", events[0]["summary"])
	require.Equal(t, "job_completed", events[0]["eventType"])

	waitStore := automationstore.New(srv.meshxDB)
	agentActiveKey := "ws-automations:" + created.ID + ":wfj_agent_task"
	require.NoError(t, waitStore.CreateJobRuns(context.Background(), []automationstore.AutomationJobRun{{ID: "wfj_agent_task", WorkspaceID: "ws-automations", RunID: created.ID, JobID: "agent_task", JobType: "github_actions_job", Status: automationstore.AutomationJobRunStatusWaiting, InputJSON: `{"steps":[{"uses":"agentteam/agent-task"}]}`, ScheduledAt: now, CreatedAt: now, UpdatedAt: now}}))
	_, err = waitStore.CreateWait(context.Background(), automationstore.AutomationWait{ID: "wfw_agent_task", WorkspaceID: "ws-automations", RunID: created.ID, JobRunID: "wfj_agent_task", WaitKind: automationstore.AutomationWaitKindAgentTask, Status: automationstore.AutomationWaitStatusActive, ActiveKey: &agentActiveKey, ResourceType: "agent_task", ResourceID: "tsk_123", CreatedAt: now, UpdatedAt: now}, automationstore.AutomationJobRun{ID: "wfj_agent_task", WorkspaceID: "ws-automations", RunID: created.ID, JobID: "agent_task", JobType: "github_actions_job", Status: automationstore.AutomationJobRunStatusWaiting, InputJSON: `{"steps":[{"uses":"agentteam/agent-task"}]}`, ScheduledAt: now, CreatedAt: now, UpdatedAt: now}, automationstore.AutomationEvent{ID: "wfe_agent_wait_started", WorkspaceID: "ws-automations", RunID: created.ID, JobRunID: "wfj_agent_task", EventType: "wait_started", CreatedAt: now})
	require.NoError(t, err)
	taskReq := httptest.NewRequest(http.MethodPost, "/api/v1/automations/runs/"+created.ID+"/waits/wfw_agent_task/complete-task", bytes.NewBufferString(`{"taskId":"tsk_123","status":"done"}`))
	taskReq.Header.Set("Authorization", auth)
	taskReq.Header.Set("X-Workspace-Id", "ws-automations")
	taskRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(taskRec, taskReq)
	require.Equal(t, http.StatusOK, taskRec.Code, taskRec.Body.String())
	require.Contains(t, taskRec.Body.String(), `"status":"resolved"`)
	require.Contains(t, taskRec.Body.String(), `\"task_id\":\"tsk_123\"`)

	eventActiveKey := "ws-automations:" + created.ID + ":wfj_event"
	require.NoError(t, waitStore.CreateJobRuns(context.Background(), []automationstore.AutomationJobRun{{ID: "wfj_event", WorkspaceID: "ws-automations", RunID: created.ID, JobID: "event", JobType: "github_actions_job", Status: automationstore.AutomationJobRunStatusWaiting, InputJSON: `{"steps":[{"uses":"agentteam/wait-event"}]}`, ScheduledAt: now, CreatedAt: now, UpdatedAt: now}}))
	_, err = waitStore.CreateWait(context.Background(), automationstore.AutomationWait{ID: "wfw_event", WorkspaceID: "ws-automations", RunID: created.ID, JobRunID: "wfj_event", WaitKind: automationstore.AutomationWaitKindWaitEvent, Status: automationstore.AutomationWaitStatusActive, ActiveKey: &eventActiveKey, ResourceType: "event", ResourceID: "topic-1", ResumeEventType: "automation_event_received", CreatedAt: now, UpdatedAt: now}, automationstore.AutomationJobRun{ID: "wfj_event", WorkspaceID: "ws-automations", RunID: created.ID, JobID: "event", JobType: "github_actions_job", Status: automationstore.AutomationJobRunStatusWaiting, InputJSON: `{"steps":[{"uses":"agentteam/wait-event"}]}`, ScheduledAt: now, CreatedAt: now, UpdatedAt: now}, automationstore.AutomationEvent{ID: "wfe_event_wait_started", WorkspaceID: "ws-automations", RunID: created.ID, JobRunID: "wfj_event", EventType: "wait_started", CreatedAt: now})
	require.NoError(t, err)
	emptyEventReq := httptest.NewRequest(http.MethodPost, "/api/v1/automations/runs/"+created.ID+"/waits/wfw_event/signal-event", bytes.NewBufferString(`{}`))
	emptyEventReq.Header.Set("Authorization", auth)
	emptyEventReq.Header.Set("X-Workspace-Id", "ws-automations")
	emptyEventRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(emptyEventRec, emptyEventReq)
	require.Equal(t, http.StatusBadRequest, emptyEventRec.Code, emptyEventRec.Body.String())
	require.Contains(t, emptyEventRec.Body.String(), "event_type is required")

	partialEventReq := httptest.NewRequest(http.MethodPost, "/api/v1/automations/runs/"+created.ID+"/waits/wfw_event/signal-event", bytes.NewBufferString(`{"eventType":"automation_event_received"}`))
	partialEventReq.Header.Set("Authorization", auth)
	partialEventReq.Header.Set("X-Workspace-Id", "ws-automations")
	partialEventRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(partialEventRec, partialEventReq)
	require.Equal(t, http.StatusBadRequest, partialEventRec.Code, partialEventRec.Body.String())
	require.Contains(t, partialEventRec.Body.String(), "resource_type is required")

	mismatchEventReq := httptest.NewRequest(http.MethodPost, "/api/v1/automations/runs/"+created.ID+"/waits/wfw_event/signal-event", bytes.NewBufferString(`{"eventType":"automation_event_received","resourceType":"event","resourceId":"wrong-topic"}`))
	mismatchEventReq.Header.Set("Authorization", auth)
	mismatchEventReq.Header.Set("X-Workspace-Id", "ws-automations")
	mismatchEventRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(mismatchEventRec, mismatchEventReq)
	require.Equal(t, http.StatusBadRequest, mismatchEventRec.Code, mismatchEventRec.Body.String())
	require.Contains(t, mismatchEventRec.Body.String(), "resource_id does not match")

	eventReq := httptest.NewRequest(http.MethodPost, "/api/v1/automations/runs/"+created.ID+"/waits/wfw_event/signal-event", bytes.NewBufferString(`{"eventType":"automation_event_received","resourceType":"event","resourceId":"topic-1","payloadJson":"{\"ok\":true}"}`))
	eventReq.Header.Set("Authorization", auth)
	eventReq.Header.Set("X-Workspace-Id", "ws-automations")
	eventRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(eventRec, eventReq)
	require.Equal(t, http.StatusOK, eventRec.Code, eventRec.Body.String())
	require.Contains(t, eventRec.Body.String(), `"status":"resolved"`)
	require.Contains(t, eventRec.Body.String(), `\"resource_id\":\"topic-1\"`)

	otherWorkspaceReq := httptest.NewRequest(http.MethodGet, "/api/v1/automations/runs/"+created.ID, nil)
	otherWorkspaceReq.Header.Set("Authorization", auth)
	otherWorkspaceReq.Header.Set("X-Workspace-Id", "ws-other")
	otherWorkspaceRec := httptest.NewRecorder()
	srv.handler.ServeHTTP(otherWorkspaceRec, otherWorkspaceReq)
	require.Equal(t, http.StatusNotFound, otherWorkspaceRec.Code, otherWorkspaceRec.Body.String())
}