feat(tfc-job-agent): Init

apps/workspace-engine/package.json

@@ -2,7 +2,7 @@
   "name": "@ctrlplane/workspace-engine",
   "scripts": {
     "dev": "air",
-    "build": "go build -o ./bin/workspace-engine main.go",
+    "build": "go build -o ./bin/workspace-engine .",
     "lint": "bash -c 'golangci-lint run --allow-parallel-runners'",
     "lint:fix": "golangci-lint run --fix",
     "start": "./bin/workspace-engine",
@@ -11,4 +11,4 @@
     "format": "bash -c 'go fmt ./...'",
     "clean": "go clean ./..."
   }
-}
+}

apps/workspace-engine/pkg/reconcile/memory/memory.go

@@ -153,7 +153,7 @@ func (q *Queue) Enqueue(ctx context.Context, params reconcile.EnqueueParams) err
 		s.ClaimedBy = ""
 		s.ClaimedUntil = nil
 	}
-	if !(s.ClaimedUntil != nil && s.ClaimedUntil.After(now)) {
+	if s.ClaimedUntil == nil || !s.ClaimedUntil.After(now) {
 		s.UpdatedAt = now
 	}
 

apps/workspace-engine/pkg/workspace/jobagents/registry.go

@@ -3,6 +3,7 @@ package jobagents
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"workspace-engine/pkg/oapi"
 	"workspace-engine/pkg/workspace/jobagents/argo"
@@ -12,6 +13,8 @@ import (
 	"workspace-engine/pkg/workspace/jobagents/types"
 	"workspace-engine/pkg/workspace/releasemanager/verification"
 	"workspace-engine/pkg/workspace/store"
+
+	"github.com/charmbracelet/log"
 )
 
 type Registry struct {
@@ -37,6 +40,58 @@ func (r *Registry) Register(dispatcher types.Dispatchable) {
 	r.dispatchers[dispatcher.Type()] = dispatcher
 }
 
+// RestoreJobs resumes tracking for all in-processing jobs after an engine restart.
+// Dispatchers that implement Restorable get their jobs back; remaining orphaned
+// jobs are failed.
+func (r *Registry) RestoreJobs(ctx context.Context) {
+	allJobs := r.store.Jobs.Items()
+
+	// Collect in-processing jobs grouped by job agent type
+	jobsByType := make(map[string][]*oapi.Job)
+	for _, job := range allJobs {
+		if !job.IsInProcessingState() {
+			continue
+		}
+		agent, ok := r.store.JobAgents.Get(job.JobAgentId)
+		if !ok {
+			continue
+		}
+		jobsByType[agent.Type] = append(jobsByType[agent.Type], job)
+	}
+
+	if len(jobsByType) == 0 {
+		return
+	}
+
+	now := time.Now().UTC()
+	for agentType, jobs := range jobsByType {
+		dispatcher, ok := r.dispatchers[agentType]
+		if !ok {
+			continue
+		}
+
+		restorable, ok := dispatcher.(types.Restorable)
+		if ok {
+			log.Info("Restoring jobs for dispatcher", "type", agentType, "count", len(jobs))
+			if err := restorable.RestoreJobs(ctx, jobs); err != nil {
+				log.Error("Failed to restore jobs", "type", agentType, "error", err)
+			}
+			continue
+		}
+
+		// Non-restorable dispatcher: fail orphaned jobs
+		for _, job := range jobs {
+			message := "Job was in-progress when the engine restarted; dispatch goroutine lost"
+			job.Status = oapi.JobStatusFailure
+			job.Message = &message
+			job.CompletedAt = &now
+			job.UpdatedAt = now
+			r.store.Jobs.Upsert(ctx, job)
+			log.Info("Marked orphaned job as failed", "jobId", job.Id, "agentType", agentType)
+		}
+	}
+}
+
 func (r *Registry) Dispatch(ctx context.Context, job *oapi.Job) error {
 	jobAgent, ok := r.store.JobAgents.Get(job.JobAgentId)
 	if !ok {