feat(agent): Albacore lifecycle fixes plus an out-of-process plugin framework

examples/plugins/hello/main.go

@@ -0,0 +1,29 @@
+// Command hello is a sample FlatRun plugin: a standalone binary that serves an API the agent
+// reverse-proxies under /api/v1/plugins/hello/. Build it with:
+//
+//	go build -o plugin ./examples/plugins/hello
+//
+// and drop the resulting "plugin" binary in <deployments>/.flatrun/plugins/hello/.
+package main
+
+import (
+	"net/http"
+
+	"github.com/flatrun/agent/pkg/pluginapi"
+	"github.com/flatrun/agent/pkg/pluginsdk"
+)
+
+func main() {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/hello", func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"message":"hello from the flatrun plugin"}`))
+	})
+
+	_ = pluginsdk.Serve(pluginapi.Info{
+		Name:        "hello",
+		Version:     "0.1.0",
+		DisplayName: "Hello Plugin",
+		Description: "A sample out-of-process plugin.",
+	}, mux)
+}

internal/api/compose_validation_integration_test.go

@@ -1,9 +1,12 @@
 package api
 
 import (
+	"os"
+	"path/filepath"
 	"strings"
 	"testing"
 
+	"github.com/flatrun/agent/internal/docker"
 	"github.com/flatrun/agent/pkg/config"
 )
 
@@ -13,7 +16,7 @@ services:
   app:
     image: nginx:alpine
 `
-	err := validateComposeWithComposeGo(validCompose)
+	err := validateComposeWithComposeGo(validCompose, ".")
 	if err != nil {
 		t.Errorf("validateComposeWithComposeGo(valid compose) = %v, want nil", err)
 	}
@@ -25,7 +28,7 @@ services:
   app:
     image: 12345
 `
-	err := validateComposeWithComposeGo(invalidCompose)
+	err := validateComposeWithComposeGo(invalidCompose, ".")
 	if err == nil {
 		t.Error("validateComposeWithComposeGo(invalid compose) = nil, want error")
 	}
@@ -40,7 +43,7 @@ services:
   app:
     image: [broken yaml
 `
-	err := validateComposeWithComposeGo(invalidCompose)
+	err := validateComposeWithComposeGo(invalidCompose, ".")
 	if err == nil {
 		t.Error("validateComposeWithComposeGo(invalid YAML) = nil, want error")
 	}
@@ -69,3 +72,40 @@ networks:
 		t.Errorf("validateComposeContent(valid) = %v, want nil", err)
 	}
 }
+
+// A compose with a relative env_file must validate against the deployment directory,
+// where the file lives, not the agent's working directory.
+func TestValidateComposeContent_RelativeEnvFile_ResolvesAgainstDeploymentDir(t *testing.T) {
+	base := t.TempDir()
+	const name = "envfileapp"
+	deployDir := filepath.Join(base, name)
+	if err := os.MkdirAll(deployDir, 0755); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(deployDir, ".env"), []byte("FOO=bar\n"), 0644); err != nil {
+		t.Fatal(err)
+	}
+
+	s := &Server{
+		config: &config.Config{
+			Infrastructure: config.InfrastructureConfig{DefaultProxyNetwork: "proxy"},
+		},
+		manager: docker.NewManager(base),
+	}
+
+	compose := `name: envfileapp
+services:
+  app:
+    image: nginx:alpine
+    env_file:
+      - ./.env
+    networks:
+      - proxy
+networks:
+  proxy:
+    external: true
+`
+	if err := s.validateComposeContent(compose, name); err != nil {
+		t.Errorf("validateComposeContent with relative env_file in deployment dir = %v, want nil", err)
+	}
+}

internal/api/jobs.go

@@ -30,6 +30,7 @@ type ActionJob struct {
 	deployment string
 	service    string
 	action     string
+	opts       actionOptions
 	activeKey  string
 
 	mu          sync.Mutex
@@ -163,14 +164,14 @@ func newJobRegistry() *jobRegistry {
 // create registers a new pending job for the deployment. If an action is
 // already in flight for that deployment it returns the existing job and false,
 // so the caller can reject the request and point the client at the live job.
-func (r *jobRegistry) create(deployment, action string) (*ActionJob, bool) {
-	return r.createScoped(deployment, "", action)
+func (r *jobRegistry) create(deployment, action string, opts actionOptions) (*ActionJob, bool) {
+	return r.createScoped(deployment, "", action, opts)
 }
 
 // createScoped registers a job serialized on the deployment, or on a single
 // service within it when service is set, so two different services can act
 // concurrently while the same service cannot.
-func (r *jobRegistry) createScoped(deployment, service, action string) (*ActionJob, bool) {
+func (r *jobRegistry) createScoped(deployment, service, action string, opts actionOptions) (*ActionJob, bool) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
@@ -193,6 +194,7 @@ func (r *jobRegistry) createScoped(deployment, service, action string) (*ActionJ
 		deployment:  deployment,
 		service:     service,
 		action:      action,
+		opts:        opts,
 		activeKey:   key,
 		status:      JobPending,
 		startedAt:   time.Now(),

internal/api/jobs_test.go

@@ -16,12 +16,16 @@ import (
 // fakeRunner stands in for the docker-backed action runner so tests exercise
 // the job lifecycle over HTTP without shelling out to compose.
 type fakeRunner struct {
-	lines []string
-	err   error
-	gate  chan struct{}
+	lines  []string
+	err    error
+	gate   chan struct{}
+	optsCh chan actionOptions
 }
 
-func (f *fakeRunner) run(_, _ string, emit func(string)) error {
+func (f *fakeRunner) run(_, _ string, opts actionOptions, emit func(string)) error {
+	if f.optsCh != nil {
+		f.optsCh <- opts
+	}
 	for _, l := range f.lines {
 		emit(l)
 	}
@@ -35,8 +39,8 @@ func newJobTestServer(runner *fakeRunner) *Server {
 	gin.SetMode(gin.TestMode)
 	s := &Server{jobs: newJobRegistry()}
 	s.runDeploymentAction = runner.run
-	s.runServiceAction = func(action, name, service string, emit func(string)) error {
-		return runner.run(action, name, emit)
+	s.runServiceAction = func(action, name, service string, opts actionOptions, emit func(string)) error {
+		return runner.run(action, name, opts, emit)
 	}
 	return s
 }
@@ -215,14 +219,45 @@ func TestServiceRestartJobIsScopedPerService(t *testing.T) {
 	}
 }
 
+// Effective-apply flags in the request body must reach the runner so updated env vars
+// and images take effect.
+func TestServiceJobThreadsEffectiveApplyOptions(t *testing.T) {
+	optsCh := make(chan actionOptions, 1)
+	s := newJobTestServer(&fakeRunner{lines: []string{"ok"}, optsCh: optsCh})
+	srv := newSkippableHTTPServer(t, newJobRouter(s))
+	defer srv.Close()
+
+	resp, err := http.Post(
+		srv.URL+"/api/deployments/app/services/web/job",
+		"application/json",
+		strings.NewReader(`{"action":"rebuild","force_recreate":true,"no_cache":true,"fresh_pull":true}`),
+	)
+	if err != nil {
+		t.Fatalf("service job request failed: %v", err)
+	}
+	resp.Body.Close()
+	if resp.StatusCode != http.StatusAccepted {
+		t.Fatalf("expected 202, got %d", resp.StatusCode)
+	}
+
+	select {
+	case got := <-optsCh:
+		if !got.ForceRecreate || !got.NoCache || !got.FreshPull {
+			t.Fatalf("runner received opts %+v, want all flags set", got)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("runner was not invoked with options")
+	}
+}
+
 func TestDeploymentJobStreamReplaysAndCompletes(t *testing.T) {
 	s := &Server{
 		jobs:           newJobRegistry(),
 		authMiddleware: auth.NewMiddleware(&config.AuthConfig{Enabled: false}),
 	}
 
 	// Pre-populate a finished job so the stream must replay buffered output.
-	job, _ := s.jobs.create("myapp", "start")
+	job, _ := s.jobs.create("myapp", "start", actionOptions{})
 	job.appendLine("Pulling nginx")
 	job.appendLine("Started")
 	s.jobs.finish(job, JobSucceeded, "")

internal/api/metadata_update_test.go

@@ -78,6 +78,33 @@ func TestMergeMetadata_PartialUpdatePreservesOtherFields(t *testing.T) {
 	}
 }
 
+// Pinning a primary service records the pin and syncs the default-domain upstream even
+// when the networking block is not part of the same update.
+func TestMergeMetadata_PrimaryServiceSyncsRoutingService(t *testing.T) {
+	existing := &models.ServiceMetadata{
+		Name: "shop",
+		Networking: models.NetworkingConfig{
+			Expose:        true,
+			Domain:        "shop.example.com",
+			Service:       "web",
+			ContainerPort: 80,
+		},
+	}
+
+	sentFields, incoming := parseTestJSON(t, `{"primary_service": "api"}`)
+	merged := mergeMetadata(existing, &incoming, sentFields)
+
+	if merged.PrimaryService != "api" {
+		t.Errorf("PrimaryService = %q, want %q", merged.PrimaryService, "api")
+	}
+	if merged.Networking.Service != "api" {
+		t.Errorf("default-domain Service should follow the pin: got %q, want %q", merged.Networking.Service, "api")
+	}
+	if merged.Networking.Domain != "shop.example.com" {
+		t.Error("unsent networking fields should be preserved")
+	}
+}
+
 func TestMergeMetadata_SentFieldOverwritesExisting(t *testing.T) {
 	existing := &models.ServiceMetadata{
 		Name:         "old-name",