feat(KLEF-204): Namespace aware provisioning

cmd/kleffctl/main.go

@@ -178,7 +178,8 @@ func ensureProjectNetwork(ctx context.Context, cli *client.Client, name, project
 
 	netLabels := map[string]string{
 		labels.ManagedBy: labels.ManagedByValue,
-		labels.ProjectID: projectID,
+				labels.ProjectID: projectID,
+				labels.EnvironmentID: projectID,
 	}
 	if projectSlug != "" {
 		netLabels[labels.ProjectSlug] = projectSlug

cmd/kleffd/main.go

@@ -70,16 +70,18 @@ func main() {
 	// --- Repository ---
 	repo := memrepo.NewServerRepository()
 
-	// Reseed the in-memory repository from any workloads that were running
-	// before this daemon process started (e.g. after a daemon restart).
-	if existing, err := runtime.ListRunning(context.Background()); err != nil {
-		daemonLog.Warn("Failed to recover running workloads on startup", "error", err)
+	// Re-adopt containers from a previous daemon run so the tailer and scraper
+	// resume without waiting for the next provision job.
+	if existing, discErr := runtime.Discover(context.Background()); discErr != nil {
+		daemonLog.Warn("Failed to discover existing workloads on startup", "error", discErr)
 	} else {
-		for _, srv := range existing {
-			_ = repo.Save(context.Background(), srv)
+		for _, rec := range existing {
+			if saveErr := repo.Save(context.Background(), rec); saveErr != nil {
+				daemonLog.Warn("Failed to restore workload record", "workload_id", rec.ID, "error", saveErr)
+			}
 		}
 		if len(existing) > 0 {
-			daemonLog.Info("Recovered running workloads", "count", len(existing))
+			daemonLog.Info("Restored workload records from runtime", "count", len(existing))
 		}
 	}
 
@@ -91,6 +93,7 @@ func main() {
 		}
 	}()
 
+
 	// --- Platform registration + status reporting ---
 	platformClient := platformadapter.NewClient(cfg.PlatformURL, cfg.SharedSecret, cfg.NodeID, cfg.FileAPIURL, daemonLog)
 	if err := platformClient.RegisterNode(context.Background()); err != nil {
@@ -155,7 +158,7 @@ func detectRuntime(cfg *config.Config, logger ports.Logger) (ports.RuntimeAdapte
 	}
 
 	// No Kubernetes — check if Docker is actually reachable before using it.
-	dockerAdapter, err := dockeradapter.New(cfg.NodeID, cfg.StoragePath)
+	dockerAdapter, err := dockeradapter.New(cfg.NodeID, cfg.StoragePath, cfg.NodeNetwork)
 	if err != nil {
 		return nil, fmt.Errorf("no runtime available: kubernetes not detected, docker client failed: %w", err)
 	}

cmd/testprovision/main.go

@@ -79,7 +79,7 @@ func main() {
 }
 
 func runDocker(cleanup bool, serverID string, spec ports.WorkloadSpec) {
-	adapter, err := dockeradapter.New("test-node", "/var/lib/kleffd/servers")
+	adapter, err := dockeradapter.New("test-node", "/var/lib/kleffd/servers", "")
 	if err != nil {
 		log.Fatalf("failed to create docker adapter: %v", err)
 	}

internal/adapters/out/platform/client.go

@@ -95,7 +95,7 @@ func (c *Client) RegisterNode(ctx context.Context) error {
 	return nil
 }
 
-func (c *Client) ShipLogs(ctx context.Context, workloadID, projectID string, lines []ports.LogEntry) error {
+func (c *Client) ShipLogs(ctx context.Context, workloadID, projectID, environmentID string, lines []ports.LogEntry) error {
 	if len(lines) == 0 {
 		return nil
 	}
@@ -108,7 +108,7 @@ func (c *Client) ShipLogs(ctx context.Context, workloadID, projectID string, lin
 	for i, l := range lines {
 		dtos[i] = lineDTO{Ts: l.Ts.UTC().Format(time.RFC3339Nano), Stream: l.Stream, Line: l.Line}
 	}
-	payload := map[string]any{"project_id": projectID, "lines": dtos}
+	payload := map[string]any{"project_id": projectID, "environment_id": environmentID, "lines": dtos}
 	body, err := json.Marshal(payload)
 	if err != nil {
 		return fmt.Errorf("marshal log payload: %w", err)

internal/adapters/out/repository/sqlite/server_repository.go

@@ -0,0 +1,83 @@
+package sqlite
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/kleffio/kleff-daemon/internal/application/ports"
+)
+
+type ServerRepository struct {
+	db *sql.DB
+}
+
+func NewServerRepository(db *sql.DB) *ServerRepository {
+	return &ServerRepository{db: db}
+}
+
+func (r *ServerRepository) Save(ctx context.Context, s *ports.ServerRecord) error {
+	_, err := r.db.ExecContext(ctx, `
+		INSERT INTO servers (id, name, address, port, status, node_id, runtime, runtime_ref, project_id, environment_id)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+		ON CONFLICT(id) DO UPDATE SET
+			name            = excluded.name,
+			address         = excluded.address,
+			port            = excluded.port,
+			status          = excluded.status,
+			node_id         = excluded.node_id,
+			runtime         = excluded.runtime,
+			runtime_ref     = excluded.runtime_ref,
+			project_id      = excluded.project_id,
+			environment_id  = excluded.environment_id,
+			updated_at      = CURRENT_TIMESTAMP
+	`, s.ID, s.Name, s.Address, s.Port, s.Status, s.NodeID, s.Runtime, s.RuntimeRef, s.ProjectID, s.EnvironmentID)
+	return err
+}
+
+func (r *ServerRepository) FindByID(ctx context.Context, id string) (*ports.ServerRecord, error) {
+	row := r.db.QueryRowContext(ctx, `
+		SELECT id, name, address, port, status, node_id, runtime, runtime_ref, project_id, environment_id
+		FROM servers WHERE id = ?
+	`, id)
+	s := &ports.ServerRecord{}
+	err := row.Scan(&s.ID, &s.Name, &s.Address, &s.Port, &s.Status, &s.NodeID, &s.Runtime, &s.RuntimeRef, &s.ProjectID, &s.EnvironmentID)
+	if err == sql.ErrNoRows {
+		return nil, fmt.Errorf("server not found: %s", id)
+	}
+	return s, err
+}
+
+func (r *ServerRepository) UpdateStatus(ctx context.Context, id string, status string) error {
+	res, err := r.db.ExecContext(ctx, `
+		UPDATE servers SET status = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?
+	`, status, id)
+	if err != nil {
+		return err
+	}
+	n, _ := res.RowsAffected()
+	if n == 0 {
+		return fmt.Errorf("server not found: %s", id)
+	}
+	return nil
+}
+
+func (r *ServerRepository) ListAll(ctx context.Context) ([]*ports.ServerRecord, error) {
+	rows, err := r.db.QueryContext(ctx, `
+		SELECT id, name, address, port, status, node_id, runtime, runtime_ref, project_id, environment_id
+		FROM servers
+	`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []*ports.ServerRecord
+	for rows.Next() {
+		s := &ports.ServerRecord{}
+		if err := rows.Scan(&s.ID, &s.Name, &s.Address, &s.Port, &s.Status, &s.NodeID, &s.Runtime, &s.RuntimeRef, &s.ProjectID, &s.EnvironmentID); err != nil {
+			return nil, err
+		}
+		out = append(out, s)
+	}
+	return out, rows.Err()
+}