feat(P0c): Job/Service/DaemonSet schemas + emitter interface + systemd stub (REQ-074)

P0c — Kind-specific schema validators + Layer 4 emitter interface.

Schemas (internal/spec/schema/schema.go, REQ-074):
- Validator interface with JobValidator, ServiceValidator, DaemonSetValidator.
  JobValidator: count=1, no service block, optional schedule/timeout.
  ServiceValidator: ports required, count>=1, restart+update+runtime required.
  DaemonSetValidator: schedule mode required, no ports (D-175), no count.
  ValidatorFor(kind) dispatcher. 96.2% coverage.

Emitter interface (internal/emitter/emitter.go, REQ-074, I-B-002):
- File{Path,Content,Mode}, Emitter interface { Render(spec,node) []File },
  Registry keyed by kind:runtime, Register + Render lookup. 100% coverage.

Systemd stub (internal/emitter/systemd.go):
- SystemdEmitter for process runtime. Renders minimal [Service] unit at
  /etc/systemd/system/orca-v1-alloc-<name>.service (orca-v1- prefix per
  dual-write window REQ-090 — no overlap with v0.8 daemon's orca-<job>).

Flock test fix: TestFlock_concurrentBlocks rewritten to use non-blocking
tryFlockEx (LOCK_NB) instead of a leaked blocking goroutine. Eliminates
the temp-dir cleanup race.

20 packages pass, 20 bats pass, gofmt clean, verify-reqs 90 consistent.

---ci---
project: orca
phase: P0c
milestone: v0.9
status: execute
---/ci---

internal/emitter/emitter.go

@@ -0,0 +1,94 @@
+// Package emitter defines the Layer-4 emitter interface (REQ-074,
+// I-B-002): the bridge between the declarative *jobspec.WorkloadSpec
+// and the server-side files. An Emitter renders a *WorkloadSpec into a
+// slice of File artifacts that the SSH-push transport SCPs to peers.
+//
+// Emitters are registered per workload kind + runtime (e.g.
+// "service:wasm", "job:process", "daemonset:wasm"). The Registry looks
+// up the right emitter by "<kind>:<runtime>" and delegates. Unknown
+// combinations return an error so the caller can fail fast before any
+// file is written.
+//
+// P0c only ships the interface, the Registry, a stub SystemdEmitter
+// (process runtime), and the File/Node value types. The full emitter
+// implementations (systemd lifecycle hooks, Traefik, Syncthing,
+// sockets) land in later phases (P02 Traefik, P04 lifecycle, P08
+// sockets, P09 Syncthing, v0.10-P03 secrets).
+package emitter
+
+import (
+	"fmt"
+	"strings"
+
+	"git.cloudinit.dev/coreci/orca/internal/jobspec"
+)
+
+// File is a single rendered artifact destined for a peer. The SSH-push
+// transport writes Content to Path atomically (write-to-tmp + rename)
+// with the given Mode (an octal string like "0644").
+type File struct {
+	Path    string
+	Content string
+	Mode    string
+}
+
+// Node is the minimal peer description an emitter needs to render
+// node-specific paths. It carries the hostname, the runtimes available
+// on the node (so emitters can branch), and the node tags (used by
+// DaemonSet matching and affinity in P05).
+type Node struct {
+	Hostname string
+	Runtime  []string
+	Tags     []string
+}
+
+// Emitter renders a *jobspec.WorkloadSpec for a given Node into a slice
+// of File artifacts. Implementations are registered with a Registry
+// keyed by "<kind>:<runtime>".
+type Emitter interface {
+	Render(spec *jobspec.WorkloadSpec, node *Node) ([]File, error)
+}
+
+// Registry holds emitters keyed by "<kind>:<runtime>" (e.g.
+// "service:process", "job:wasm"). The zero-value Registry is not
+// usable; construct one with NewRegistry.
+type Registry struct {
+	emitters map[string]Emitter
+}
+
+// NewRegistry returns an empty Registry ready for Register calls.
+func NewRegistry() *Registry {
+	return &Registry{emitters: make(map[string]Emitter)}
+}
+
+// Register registers an Emitter under the given key. The key is
+// "<kind>:<runtime>" (e.g. "job:process"). Registering twice under the
+// same key overwrites the prior registration (last-wins) to keep the
+// surface simple; callers are responsible for not double-registering.
+func (r *Registry) Register(key string, e Emitter) {
+	r.emitters[key] = e
+}
+
+// Render looks up the emitter for "<kind>:<runtime>" in the registry and
+// delegates to it. The kind is lowercased so the canonical spec kinds
+// (Job, Service, DaemonSet) map to the lowercase registry keys
+// ("job:process", "service:wasm", "daemonset:process"). Returns an error
+// if the spec is nil, the spec is missing its Kind, the runtime is
+// missing, or no emitter is registered for the combination.
+func (r *Registry) Render(spec *jobspec.WorkloadSpec, node *Node) ([]File, error) {
+	if spec == nil {
+		return nil, fmt.Errorf("emitter: spec is nil")
+	}
+	if strings.TrimSpace(spec.Kind) == "" {
+		return nil, fmt.Errorf("emitter: spec kind is empty")
+	}
+	if spec.Runtime == nil {
+		return nil, fmt.Errorf("emitter: spec runtime is nil")
+	}
+	key := strings.ToLower(spec.Kind) + ":" + spec.Runtime.OneOf
+	e, ok := r.emitters[key]
+	if !ok {
+		return nil, fmt.Errorf("emitter: no emitter registered for %q (kind:runtime)", key)
+	}
+	return e.Render(spec, node)
+}

internal/emitter/emitter_test.go

@@ -0,0 +1,163 @@
+package emitter
+
+import (
+	"errors"
+	"strings"
+	"testing"
+
+	"git.cloudinit.dev/coreci/orca/internal/jobspec"
+)
+
+// mockEmitter is a test-only Emitter that returns a fixed File slice
+// (or an error) so the Registry tests do not depend on the
+// SystemdEmitter. Implements Emitter via value receiver.
+type mockEmitter struct {
+	files []File
+	err   error
+}
+
+func (m mockEmitter) Render(spec *jobspec.WorkloadSpec, node *Node) ([]File, error) {
+	if m.err != nil {
+		return nil, m.err
+	}
+	out := make([]File, len(m.files))
+	copy(out, m.files)
+	return out, nil
+}
+
+func TestRegistry_RegisterAndRender(t *testing.T) {
+	r := NewRegistry()
+	want := []File{{Path: "/tmp/a", Content: "alpha", Mode: "0644"}}
+	r.Register("job:process", mockEmitter{files: want})
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "demo",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"},
+	}
+	node := &Node{Hostname: "node-1", Runtime: []string{"process"}}
+	got, err := r.Render(spec, node)
+	if err != nil {
+		t.Fatalf("Render: %v", err)
+	}
+	if len(got) != 1 {
+		t.Fatalf("got %d files, want 1", len(got))
+	}
+	if got[0] != want[0] {
+		t.Errorf("file = %+v, want %+v", got[0], want[0])
+	}
+}
+
+func TestRegistry_UnknownKindRuntime(t *testing.T) {
+	r := NewRegistry()
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "wasm"},
+	}
+	_, err := r.Render(spec, &Node{})
+	if err == nil {
+		t.Fatal("expected error for unknown kind:runtime, got nil")
+	}
+	if !strings.Contains(err.Error(), "no emitter registered") {
+		t.Errorf("error = %q, want 'no emitter registered'", err.Error())
+	}
+	if !strings.Contains(err.Error(), "service:wasm") {
+		t.Errorf("error = %q, want it to mention 'service:wasm'", err.Error())
+	}
+}
+
+func TestRegistry_MultipleEmittersCorrectSelected(t *testing.T) {
+	r := NewRegistry()
+	jobFiles := []File{{Path: "/tmp/job", Content: "job", Mode: "0644"}}
+	svcFiles := []File{{Path: "/tmp/svc", Content: "svc", Mode: "0644"}}
+	dsFiles := []File{{Path: "/tmp/ds", Content: "ds", Mode: "0644"}}
+	r.Register("job:process", mockEmitter{files: jobFiles})
+	r.Register("service:process", mockEmitter{files: svcFiles})
+	r.Register("daemonset:process", mockEmitter{files: dsFiles})
+
+	cases := []struct {
+		kind     string
+		runtime  string
+		wantPath string
+	}{
+		{"Job", "process", "/tmp/job"},
+		{"Service", "process", "/tmp/svc"},
+		{"DaemonSet", "process", "/tmp/ds"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.kind+":"+tc.runtime, func(t *testing.T) {
+			spec := &jobspec.WorkloadSpec{
+				Kind:    tc.kind,
+				Name:    "x",
+				Runtime: &jobspec.RuntimeBlock{OneOf: tc.runtime, Command: "/bin/x"},
+			}
+			got, err := r.Render(spec, &Node{Hostname: "n"})
+			if err != nil {
+				t.Fatalf("Render: %v", err)
+			}
+			if len(got) != 1 {
+				t.Fatalf("got %d files, want 1", len(got))
+			}
+			if got[0].Path != tc.wantPath {
+				t.Errorf("path = %q, want %q", got[0].Path, tc.wantPath)
+			}
+		})
+	}
+}
+
+func TestRegistry_NilSpec(t *testing.T) {
+	r := NewRegistry()
+	_, err := r.Render(nil, &Node{})
+	if err == nil {
+		t.Fatal("expected error for nil spec")
+	}
+	if !strings.Contains(err.Error(), "spec is nil") {
+		t.Errorf("error = %q, want 'spec is nil'", err.Error())
+	}
+}
+
+func TestRegistry_EmptyKind(t *testing.T) {
+	r := NewRegistry()
+	spec := &jobspec.WorkloadSpec{Runtime: &jobspec.RuntimeBlock{OneOf: "process"}}
+	_, err := r.Render(spec, &Node{})
+	if err == nil {
+		t.Fatal("expected error for empty kind")
+	}
+	if !strings.Contains(err.Error(), "kind is empty") {
+		t.Errorf("error = %q, want 'kind is empty'", err.Error())
+	}
+}
+
+func TestRegistry_NilRuntime(t *testing.T) {
+	r := NewRegistry()
+	spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "x"}
+	_, err := r.Render(spec, &Node{})
+	if err == nil {
+		t.Fatal("expected error for nil runtime")
+	}
+	if !strings.Contains(err.Error(), "runtime is nil") {
+		t.Errorf("error = %q, want 'runtime is nil'", err.Error())
+	}
+}
+
+func TestRegistry_EmitterErrorPropagates(t *testing.T) {
+	r := NewRegistry()
+	wantErr := errors.New("boom")
+	r.Register("job:process", mockEmitter{err: wantErr})
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "x",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/x"},
+	}
+	_, err := r.Render(spec, &Node{})
+	if !errors.Is(err, wantErr) {
+		t.Errorf("err = %v, want %v", err, wantErr)
+	}
+}
+
+// Compile-time assertion that mockEmitter and SystemdEmitter implement
+// Emitter.
+var (
+	_ Emitter = mockEmitter{}
+	_ Emitter = SystemdEmitter{}
+)

internal/emitter/systemd.go

@@ -0,0 +1,60 @@
+package emitter
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"git.cloudinit.dev/coreci/orca/internal/jobspec"
+)
+
+// SystemdEmitter is a stub Emitter implementation for the "process"
+// runtime. It renders a minimal systemd unit file for the workload.
+//
+// This is a STUB — the full systemd emitter (with lifecycle hooks,
+// sockets, EnvironmentFile, LoadCredential) lands in later phases:
+//
+//   - P04: lifecycle hooks (ExecStop, ExecStartPre/Post, timeouts)
+//   - P08: socket plumbing (R-007)
+//   - v0.10-P03: secrets via EnvironmentFile= + LoadCredential=
+//
+// P0c ships only the minimal [Service]\nExecStart=... shape to prove
+// the Emitter interface end-to-end. The unit name carries the
+// `orca-v1-` prefix per the dual-write window (REQ-090) so the v0.9
+// SSH-push path does not collide with the v0.8 daemon's
+// `orca-<job>.service` units during the migration window.
+type SystemdEmitter struct{}
+
+// unitNamePrefix is the v0.9 SSH-push unit-name prefix. The v0.8
+// daemon uses `orca-<job>.service`; the v0.9 path uses
+// `orca-v1-<spec.Name>.service` so the two never overlap (REQ-090,
+// I-C-006). The prefix is load-bearing — do not change it without
+// updating the dual-write window contract.
+const unitNamePrefix = "orca-v1-"
+
+// Render renders a minimal systemd unit file for a process-runtime
+// workload. The unit name is `/etc/systemd/system/<unitNamePrefix><spec.Name>.service`
+// and the content is a minimal `[Service]` block with the runtime
+// command as ExecStart. Mode is 0644 (the lead applier chmods after
+// atomic rename).
+//
+// Returns an error if the spec is nil, the spec is missing its name,
+// or the runtime command is empty (a workload with no command has
+// nothing to ExecStart).
+func (SystemdEmitter) Render(spec *jobspec.WorkloadSpec, node *Node) ([]File, error) {
+	if spec == nil {
+		return nil, errors.New("emitter/systemd: spec is nil")
+	}
+	if strings.TrimSpace(spec.Name) == "" {
+		return nil, errors.New("emitter/systemd: spec name is empty")
+	}
+	if spec.Runtime == nil {
+		return nil, errors.New("emitter/systemd: runtime block is nil")
+	}
+	if strings.TrimSpace(spec.Runtime.Command) == "" {
+		return nil, errors.New("emitter/systemd: runtime command is empty")
+	}
+	path := fmt.Sprintf("/etc/systemd/system/%s%s.service", unitNamePrefix, spec.Name)
+	content := fmt.Sprintf("[Service]\nExecStart=%s\n", spec.Runtime.Command)
+	return []File{{Path: path, Content: content, Mode: "0644"}}, nil
+}

internal/emitter/systemd_test.go

@@ -0,0 +1,156 @@
+package emitter
+
+import (
+	"strings"
+	"testing"
+
+	"git.cloudinit.dev/coreci/orca/internal/jobspec"
+)
+
+func TestSystemdEmitter_RenderJob(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "backup",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/usr/bin/rsync -a /src /dst"},
+	}
+	node := &Node{Hostname: "node-1", Runtime: []string{"process"}}
+	files, err := SystemdEmitter{}.Render(spec, node)
+	if err != nil {
+		t.Fatalf("Render: %v", err)
+	}
+	if len(files) != 1 {
+		t.Fatalf("got %d files, want 1", len(files))
+	}
+	f := files[0]
+	wantPath := "/etc/systemd/system/orca-v1-backup.service"
+	if f.Path != wantPath {
+		t.Errorf("Path = %q, want %q", f.Path, wantPath)
+	}
+	wantContent := "[Service]\nExecStart=/usr/bin/rsync -a /src /dst\n"
+	if f.Content != wantContent {
+		t.Errorf("Content = %q, want %q", f.Content, wantContent)
+	}
+	if f.Mode != "0644" {
+		t.Errorf("Mode = %q, want 0644", f.Mode)
+	}
+}
+
+func TestSystemdEmitter_RenderService(t *testing.T) {
+	// The full service emitter (Traefik route + health checks) lands in
+	// P02; here we only prove the systemd side renders for a Service
+	// kind with a process runtime.
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/usr/local/bin/httpd -f"},
+	}
+	node := &Node{Hostname: "node-1", Runtime: []string{"process"}}
+	files, err := SystemdEmitter{}.Render(spec, node)
+	if err != nil {
+		t.Fatalf("Render: %v", err)
+	}
+	if len(files) != 1 {
+		t.Fatalf("got %d files, want 1", len(files))
+	}
+	if files[0].Path != "/etc/systemd/system/orca-v1-web.service" {
+		t.Errorf("Path = %q, want /etc/systemd/system/orca-v1-web.service", files[0].Path)
+	}
+	if !strings.Contains(files[0].Content, "ExecStart=/usr/local/bin/httpd -f") {
+		t.Errorf("Content = %q, want it to contain the ExecStart line", files[0].Content)
+	}
+}
+
+func TestSystemdEmitter_EmptyCommandError(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "x",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: ""},
+	}
+	_, err := SystemdEmitter{}.Render(spec, &Node{})
+	if err == nil {
+		t.Fatal("expected error for empty command, got nil")
+	}
+	if !strings.Contains(err.Error(), "command is empty") {
+		t.Errorf("error = %q, want 'command is empty'", err.Error())
+	}
+}
+
+func TestSystemdEmitter_WhitespaceCommandError(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "x",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "   "},
+	}
+	_, err := SystemdEmitter{}.Render(spec, &Node{})
+	if err == nil {
+		t.Fatal("expected error for whitespace-only command, got nil")
+	}
+	if !strings.Contains(err.Error(), "command is empty") {
+		t.Errorf("error = %q, want 'command is empty'", err.Error())
+	}
+}
+
+func TestSystemdEmitter_NilSpec(t *testing.T) {
+	v := SystemdEmitter{}
+	if _, err := v.Render(nil, &Node{}); err == nil {
+		t.Fatal("expected error for nil spec")
+	}
+}
+
+func TestSystemdEmitter_EmptyName(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "  ",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/x"},
+	}
+	_, err := SystemdEmitter{}.Render(spec, &Node{})
+	if err == nil {
+		t.Fatal("expected error for empty name")
+	}
+	if !strings.Contains(err.Error(), "name is empty") {
+		t.Errorf("error = %q, want 'name is empty'", err.Error())
+	}
+}
+
+func TestSystemdEmitter_NilRuntime(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "x"}
+	_, err := SystemdEmitter{}.Render(spec, &Node{})
+	if err == nil {
+		t.Fatal("expected error for nil runtime")
+	}
+	if !strings.Contains(err.Error(), "runtime block is nil") {
+		t.Errorf("error = %q, want 'runtime block is nil'", err.Error())
+	}
+}
+
+func TestSystemdEmitter_UnitNamePrefix(t *testing.T) {
+	// The orca-v1- prefix is load-bearing for the dual-write window
+	// (REQ-090, I-C-006): the v0.8 daemon writes `orca-<job>.service`
+	// and the v0.9 SSH-push path writes `orca-v1-<spec.Name>.service`,
+	// so the two never collide. This test guards against accidental
+	// removal of the prefix.
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "dual-write-safety",
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"},
+	}
+	files, err := SystemdEmitter{}.Render(spec, &Node{Hostname: "n"})
+	if err != nil {
+		t.Fatalf("Render: %v", err)
+	}
+	if !strings.HasPrefix(files[0].Path, "/etc/systemd/system/orca-v1-") {
+		t.Errorf("Path = %q, want it to start with /etc/systemd/system/orca-v1- (REQ-090)", files[0].Path)
+	}
+	if !strings.HasSuffix(files[0].Path, ".service") {
+		t.Errorf("Path = %q, want it to end with .service", files[0].Path)
+	}
+	// Explicitly assert the full expected unit name to lock the contract.
+	want := "/etc/systemd/system/orca-v1-dual-write-safety.service"
+	if files[0].Path != want {
+		t.Errorf("Path = %q, want %q", files[0].Path, want)
+	}
+	// Sanity: the prefix is exactly "orca-v1-", not "orca-v0" or "orca".
+	if unitNamePrefix != "orca-v1-" {
+		t.Errorf("unitNamePrefix = %q, want orca-v1-", unitNamePrefix)
+	}
+}

internal/jobspec/markdown.go

@@ -24,6 +24,37 @@ type WorkloadSpec struct {
 	Secrets     []string
 	Volumes     []VolumeSpec
 	Body        string
+
+	// Kind-specific blocks consumed by the P0c schema validators
+	// (internal/spec/schema). The Markdown parser does not populate
+	// these yet; later phases (P02 service block, P03 update stanza,
+	// P04 lifecycle hooks) extend the parser. P0c only defines the
+	// struct shape so validators can reference the fields.
+
+	// Restart is the restart policy block. Required for Service and
+	// DaemonSet; optional for Job (defaults to never/on-failure).
+	// Populated by the P04 lifecycle phase.
+	Restart *RestartBlock
+
+	// Schedule is the schedule block. For Job it carries an optional
+	// cron string; for DaemonSet it carries the placement mode
+	// (every-node/matching/mandatory). Populated by P05 (scheduler
+	// skeleton) and the DaemonSet phase.
+	Schedule *ScheduleBlock
+
+	// Update is the rolling/canary update stanza. Required for
+	// Service. Populated by the P03 update-stanza phase.
+	Update *UpdateBlock
+
+	// Service is the service block (Traefik route definition). For
+	// Service kind it is implied; Job and DaemonSet do not carry a
+	// Traefik route by default (D-175). Populated by the P02 service
+	// block phase.
+	Service *ServiceBlock
+
+	// Timeout is an optional execution timeout (duration string) for
+	// Job. Populated by P04.
+	Timeout string
 }
 
 // RuntimeBlock is a minimal runtime abstraction surface populated by the
@@ -35,6 +66,37 @@ type RuntimeBlock struct {
 	Command string
 }
 
+// RestartBlock is the restart policy block. Mode is one of never,
+// on-failure, service (REQ-074 schema validators). Populated by P04.
+type RestartBlock struct {
+	Mode       string
+	MaxRetries int
+	Delay      string
+}
+
+// ScheduleBlock is the scheduling block. For Job, Cron is an optional
+// cron expression. For DaemonSet, Mode is one of every-node, matching,
+// mandatory (REQ-074). Populated by P05 and the DaemonSet phase.
+type ScheduleBlock struct {
+	Mode string
+	Cron string
+}
+
+// UpdateBlock is the rolling/canary update stanza. Required for Service.
+// Populated by P03.
+type UpdateBlock struct {
+	Strategy string
+	MaxSurge int
+}
+
+// ServiceBlock is the Traefik route definition. For Service it is
+// implied (Traefik route YES); Job and DaemonSet do not carry one by
+// default (D-175). Populated by P02.
+type ServiceBlock struct {
+	Host    string
+	RouteID string
+}
+
 // PortSpec is a minimal port binding entry. HostIP is optional.
 type PortSpec struct {
 	Name     string

internal/security/flock.go

@@ -25,3 +25,11 @@ func Flock(path string) (release func(), err error) {
 		_ = f.Close()
 	}, nil
 }
+
+func tryFlockEx(fd int) error {
+	return syscall.Flock(fd, syscall.LOCK_EX|syscall.LOCK_NB)
+}
+
+func releaseFlock(fd int) error {
+	return syscall.Flock(fd, syscall.LOCK_UN)
+}

internal/security/flock_test.go

@@ -4,6 +4,7 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 )
 
 func TestFlock_acquireAndRelease(t *testing.T) {
@@ -45,17 +46,37 @@ func TestFlock_concurrentBlocks(t *testing.T) {
 	if err != nil {
 		t.Fatalf("first Flock: %v", err)
 	}
-	defer r1()
 
-	done := make(chan error, 1)
+	// Give the blocking goroutine a chance to start and block.
+	time.Sleep(50 * time.Millisecond)
+
+	// Verify the second lock is blocked by checking it hasn't acquired after a short window.
+	// Use a non-blocking attempt: open the file and try LOCK_EX|LOCK_NB.
+	blocked := make(chan bool, 1)
 	go func() {
-		_, err := Flock(path)
-		done <- err
+		f, err := os.OpenFile(path, os.O_RDWR, 0o600)
+		if err != nil {
+			blocked <- false
+			return
+		}
+		defer f.Close()
+		// LOCK_NB = non-blocking; returns EWOULDBLOCK if locked.
+		if err := tryFlockEx(int(f.Fd())); err != nil {
+			blocked <- true // got EWOULDBLOCK = the lock is held by r1
+			return
+		}
+		releaseFlock(int(f.Fd()))
+		blocked <- false // acquired = r1 didn't hold the lock (bug)
 	}()
 
 	select {
-	case <-done:
-		t.Fatal("second Flock should block while first holds the lock")
-	default:
+	case b := <-blocked:
+		if !b {
+			t.Fatal("second lock acquired while first holds it — lock not working")
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("non-blocking try-lock timed out")
 	}
+
+	r1()
 }

internal/spec/schema/schema.go

@@ -0,0 +1,164 @@
+// Package schema provides kind-specific validators for the unified
+// *jobspec.WorkloadSpec introduced in P0b (REQ-064). Each workload kind
+// (Job, Service, DaemonSet per R-012) has different required fields;
+// this package exposes a Validator interface and a ValidatorFor
+// dispatcher so the emitter layer (REQ-074) and the lint engine
+// (REQ-084) can reject invalid specs before rendering.
+//
+// The validators operate purely on the *WorkloadSpec shape; they do no
+// I/O. Required-field violations return a structured error listing
+// every problem found (missing required fields, invalid combinations).
+package schema
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"git.cloudinit.dev/coreci/orca/internal/jobspec"
+)
+
+// Validator validates a *jobspec.WorkloadSpec against a kind-specific
+// schema. Implementations are pure (no I/O) and return a clear error
+// listing every violation found.
+type Validator interface {
+	Validate(spec *jobspec.WorkloadSpec) error
+}
+
+// JobValidator validates the Job workload kind (R-012).
+//
+// Rules:
+//   - no service block required (Job has no Traefik route by default D-175)
+//   - restart optional (defaults to never/on-failure when omitted)
+//   - schedule optional (cron string)
+//   - timeout optional
+//   - ports optional
+//   - count must be 1 (or unset → 1); count > 1 is an error for Job
+//     (use a Service for replicas)
+//   - no Traefik route (a ServiceBlock is rejected)
+type JobValidator struct{}
+
+// ServiceValidator validates the Service workload kind (R-012).
+//
+// Rules:
+//   - ports required (at least one)
+//   - count ≥ 1
+//   - restart required (mode must be service)
+//   - update required
+//   - runtime required
+//   - service block implied (Traefik route YES)
+type ServiceValidator struct{}
+
+// DaemonSetValidator validates the DaemonSet workload kind (R-012).
+//
+// Rules:
+//   - schedule block with mode (every-node/matching/mandatory) required
+//   - no ports (no Traefik route by default D-175)
+//   - no count (implicit = nodes matching condition)
+//   - restart required
+type DaemonSetValidator struct{}
+
+// ValidatorFor returns the Validator for the given workload kind, or an
+// error for an unknown kind. kind must be one of Job, Service,
+// DaemonSet (R-012).
+func ValidatorFor(kind string) (Validator, error) {
+	switch kind {
+	case "Job":
+		return JobValidator{}, nil
+	case "Service":
+		return ServiceValidator{}, nil
+	case "DaemonSet":
+		return DaemonSetValidator{}, nil
+	default:
+		return nil, fmt.Errorf("schema: unknown kind %q (want one of Job, Service, DaemonSet)", kind)
+	}
+}
+
+// Validate validates a Job spec. See JobValidator for the rules.
+func (JobValidator) Validate(spec *jobspec.WorkloadSpec) error {
+	if spec == nil {
+		return errors.New("schema/Job: spec is nil")
+	}
+	var errs []string
+	if strings.TrimSpace(spec.Name) == "" {
+		errs = append(errs, "name is required")
+	}
+	if spec.Count != 0 && spec.Count != 1 {
+		errs = append(errs, fmt.Sprintf("count must be 1 (or unset) for Job, got %d (use Service for replicas)", spec.Count))
+	}
+	if spec.Service != nil {
+		errs = append(errs, "service block (Traefik route) is not allowed for Job (D-175)")
+	}
+	return composeErrors("schema/Job", errs)
+}
+
+// Validate validates a Service spec. See ServiceValidator for the rules.
+func (ServiceValidator) Validate(spec *jobspec.WorkloadSpec) error {
+	if spec == nil {
+		return errors.New("schema/Service: spec is nil")
+	}
+	var errs []string
+	if strings.TrimSpace(spec.Name) == "" {
+		errs = append(errs, "name is required")
+	}
+	if len(spec.Ports) == 0 {
+		errs = append(errs, "ports required (at least one)")
+	}
+	if spec.Count < 1 {
+		errs = append(errs, fmt.Sprintf("count must be ≥ 1 for Service, got %d", spec.Count))
+	}
+	if spec.Restart == nil {
+		errs = append(errs, "restart block required for Service")
+	} else if spec.Restart.Mode != "service" {
+		errs = append(errs, fmt.Sprintf("restart mode must be %q for Service, got %q", "service", spec.Restart.Mode))
+	}
+	if spec.Update == nil {
+		errs = append(errs, "update block required for Service")
+	}
+	if spec.Runtime == nil {
+		errs = append(errs, "runtime block required for Service")
+	}
+	return composeErrors("schema/Service", errs)
+}
+
+// Validate validates a DaemonSet spec. See DaemonSetValidator for the rules.
+func (DaemonSetValidator) Validate(spec *jobspec.WorkloadSpec) error {
+	if spec == nil {
+		return errors.New("schema/DaemonSet: spec is nil")
+	}
+	var errs []string
+	if strings.TrimSpace(spec.Name) == "" {
+		errs = append(errs, "name is required")
+	}
+	if spec.Schedule == nil {
+		errs = append(errs, "schedule block required for DaemonSet")
+	} else {
+		switch spec.Schedule.Mode {
+		case "every-node", "matching", "mandatory":
+		case "":
+			errs = append(errs, "schedule mode required for DaemonSet (one of every-node, matching, mandatory)")
+		default:
+			errs = append(errs, fmt.Sprintf("schedule mode %q invalid (want one of every-node, matching, mandatory)", spec.Schedule.Mode))
+		}
+	}
+	if len(spec.Ports) > 0 {
+		errs = append(errs, "ports not allowed for DaemonSet (no Traefik route by default D-175)")
+	}
+	if spec.Count != 0 {
+		errs = append(errs, fmt.Sprintf("count not allowed for DaemonSet (implicit = nodes matching condition), got %d", spec.Count))
+	}
+	if spec.Restart == nil {
+		errs = append(errs, "restart block required for DaemonSet")
+	}
+	return composeErrors("schema/DaemonSet", errs)
+}
+
+// composeErrors joins the per-field errors into a single error prefixed
+// by the validator name. Returns nil when there are no errors so the
+// caller can return the result directly.
+func composeErrors(name string, errs []string) error {
+	if len(errs) == 0 {
+		return nil
+	}
+	return fmt.Errorf("%s: %s", name, strings.Join(errs, "; "))
+}

internal/spec/schema/schema_test.go

@@ -0,0 +1,436 @@
+package schema
+
+import (
+	"strings"
+	"testing"
+
+	"git.cloudinit.dev/coreci/orca/internal/jobspec"
+)
+
+func TestJobValidator_ValidMinimal(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "backup", Count: 1}
+	v := JobValidator{}
+	if err := v.Validate(spec); err != nil {
+		t.Fatalf("expected nil, got %v", err)
+	}
+}
+
+func TestJobValidator_ValidWithSchedule(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "Job",
+		Name:     "backup",
+		Count:    1,
+		Schedule: &jobspec.ScheduleBlock{Cron: "0 2 * * *"},
+		Timeout:  "1h",
+	}
+	v := JobValidator{}
+	if err := v.Validate(spec); err != nil {
+		t.Fatalf("expected nil, got %v", err)
+	}
+}
+
+func TestJobValidator_ValidUnsetCount(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "one-shot"}
+	v := JobValidator{}
+	if err := v.Validate(spec); err != nil {
+		t.Fatalf("unset count should default-accept, got %v", err)
+	}
+}
+
+func TestJobValidator_MissingName(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{Kind: "Job", Count: 1}
+	err := JobValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing name, got nil")
+	}
+	if !strings.Contains(err.Error(), "name is required") {
+		t.Errorf("error = %q, want 'name is required'", err.Error())
+	}
+}
+
+func TestJobValidator_CountGreaterThanOne(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "batch", Count: 3}
+	err := JobValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for count > 1, got nil")
+	}
+	if !strings.Contains(err.Error(), "count must be 1") {
+		t.Errorf("error = %q, want 'count must be 1'", err.Error())
+	}
+}
+
+func TestJobValidator_ServiceBlockRejected(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Job",
+		Name:    "x",
+		Count:   1,
+		Service: &jobspec.ServiceBlock{Host: "x.example"},
+	}
+	err := JobValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for service block on Job, got nil")
+	}
+	if !strings.Contains(err.Error(), "service block") {
+		t.Errorf("error = %q, want 'service block'", err.Error())
+	}
+}
+
+func TestJobValidator_NilSpec(t *testing.T) {
+	v := JobValidator{}
+	if err := v.Validate(nil); err == nil {
+		t.Fatal("expected error for nil spec")
+	}
+}
+
+func TestServiceValidator_ValidFull(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Count:   3,
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/http"},
+		Restart: &jobspec.RestartBlock{Mode: "service"},
+		Update:  &jobspec.UpdateBlock{Strategy: "rolling", MaxSurge: 1},
+		Ports:   []jobspec.PortSpec{{Name: "http", Port: 8080}},
+	}
+	v := ServiceValidator{}
+	if err := v.Validate(spec); err != nil {
+		t.Fatalf("expected nil, got %v", err)
+	}
+}
+
+func TestServiceValidator_MissingPorts(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Count:   2,
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process"},
+		Restart: &jobspec.RestartBlock{Mode: "service"},
+		Update:  &jobspec.UpdateBlock{Strategy: "rolling"},
+	}
+	err := ServiceValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing ports, got nil")
+	}
+	if !strings.Contains(err.Error(), "ports required") {
+		t.Errorf("error = %q, want 'ports required'", err.Error())
+	}
+}
+
+func TestServiceValidator_CountZero(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Count:   0,
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process"},
+		Restart: &jobspec.RestartBlock{Mode: "service"},
+		Update:  &jobspec.UpdateBlock{Strategy: "rolling"},
+		Ports:   []jobspec.PortSpec{{Name: "http", Port: 8080}},
+	}
+	err := ServiceValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for count 0, got nil")
+	}
+	if !strings.Contains(err.Error(), "count must be") {
+		t.Errorf("error = %q, want 'count must be'", err.Error())
+	}
+}
+
+func TestServiceValidator_MissingRestart(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Count:   1,
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process"},
+		Update:  &jobspec.UpdateBlock{Strategy: "rolling"},
+		Ports:   []jobspec.PortSpec{{Name: "http", Port: 8080}},
+	}
+	err := ServiceValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing restart, got nil")
+	}
+	if !strings.Contains(err.Error(), "restart block required") {
+		t.Errorf("error = %q, want 'restart block required'", err.Error())
+	}
+}
+
+func TestServiceValidator_WrongRestartMode(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Count:   1,
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process"},
+		Restart: &jobspec.RestartBlock{Mode: "on-failure"},
+		Update:  &jobspec.UpdateBlock{Strategy: "rolling"},
+		Ports:   []jobspec.PortSpec{{Name: "http", Port: 8080}},
+	}
+	err := ServiceValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for wrong restart mode, got nil")
+	}
+	if !strings.Contains(err.Error(), "restart mode must be") {
+		t.Errorf("error = %q, want 'restart mode must be'", err.Error())
+	}
+}
+
+func TestServiceValidator_MissingUpdate(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Count:   1,
+		Runtime: &jobspec.RuntimeBlock{OneOf: "process"},
+		Restart: &jobspec.RestartBlock{Mode: "service"},
+		Ports:   []jobspec.PortSpec{{Name: "http", Port: 8080}},
+	}
+	err := ServiceValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing update, got nil")
+	}
+	if !strings.Contains(err.Error(), "update block required") {
+		t.Errorf("error = %q, want 'update block required'", err.Error())
+	}
+}
+
+func TestServiceValidator_MissingRuntime(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "Service",
+		Name:    "web",
+		Count:   1,
+		Restart: &jobspec.RestartBlock{Mode: "service"},
+		Update:  &jobspec.UpdateBlock{Strategy: "rolling"},
+		Ports:   []jobspec.PortSpec{{Name: "http", Port: 8080}},
+	}
+	err := ServiceValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing runtime, got nil")
+	}
+	if !strings.Contains(err.Error(), "runtime block required") {
+		t.Errorf("error = %q, want 'runtime block required'", err.Error())
+	}
+}
+
+func TestServiceValidator_NilSpec(t *testing.T) {
+	v := ServiceValidator{}
+	if err := v.Validate(nil); err == nil {
+		t.Fatal("expected error for nil spec")
+	}
+}
+
+func TestDaemonSetValidator_Valid(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "log-shipper",
+		Schedule: &jobspec.ScheduleBlock{Mode: "every-node"},
+		Restart:  &jobspec.RestartBlock{Mode: "on-failure"},
+	}
+	v := DaemonSetValidator{}
+	if err := v.Validate(spec); err != nil {
+		t.Fatalf("expected nil, got %v", err)
+	}
+}
+
+func TestDaemonSetValidator_ValidMatchingMode(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "x",
+		Schedule: &jobspec.ScheduleBlock{Mode: "matching"},
+		Restart:  &jobspec.RestartBlock{Mode: "on-failure"},
+	}
+	v := DaemonSetValidator{}
+	if err := v.Validate(spec); err != nil {
+		t.Fatalf("matching mode should be accepted, got %v", err)
+	}
+}
+
+func TestDaemonSetValidator_ValidMandatoryMode(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "x",
+		Schedule: &jobspec.ScheduleBlock{Mode: "mandatory"},
+		Restart:  &jobspec.RestartBlock{Mode: "on-failure"},
+	}
+	v := DaemonSetValidator{}
+	if err := v.Validate(spec); err != nil {
+		t.Fatalf("mandatory mode should be accepted, got %v", err)
+	}
+}
+
+func TestDaemonSetValidator_MissingScheduleMode(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "x",
+		Schedule: &jobspec.ScheduleBlock{Mode: ""},
+		Restart:  &jobspec.RestartBlock{Mode: "on-failure"},
+	}
+	err := DaemonSetValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing schedule mode, got nil")
+	}
+	if !strings.Contains(err.Error(), "schedule mode required") {
+		t.Errorf("error = %q, want 'schedule mode required'", err.Error())
+	}
+}
+
+func TestDaemonSetValidator_MissingScheduleBlock(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:    "DaemonSet",
+		Name:    "x",
+		Restart: &jobspec.RestartBlock{Mode: "on-failure"},
+	}
+	err := DaemonSetValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing schedule block, got nil")
+	}
+	if !strings.Contains(err.Error(), "schedule block required") {
+		t.Errorf("error = %q, want 'schedule block required'", err.Error())
+	}
+}
+
+func TestDaemonSetValidator_InvalidScheduleMode(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "x",
+		Schedule: &jobspec.ScheduleBlock{Mode: "always"},
+		Restart:  &jobspec.RestartBlock{Mode: "on-failure"},
+	}
+	err := DaemonSetValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for invalid schedule mode, got nil")
+	}
+	if !strings.Contains(err.Error(), "schedule mode") {
+		t.Errorf("error = %q, want 'schedule mode'", err.Error())
+	}
+}
+
+func TestDaemonSetValidator_HasPorts(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "x",
+		Schedule: &jobspec.ScheduleBlock{Mode: "every-node"},
+		Restart:  &jobspec.RestartBlock{Mode: "on-failure"},
+		Ports:    []jobspec.PortSpec{{Name: "http", Port: 80}},
+	}
+	err := DaemonSetValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for ports on DaemonSet, got nil")
+	}
+	if !strings.Contains(err.Error(), "ports not allowed") {
+		t.Errorf("error = %q, want 'ports not allowed'", err.Error())
+	}
+}
+
+func TestDaemonSetValidator_HasCount(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "x",
+		Count:    3,
+		Schedule: &jobspec.ScheduleBlock{Mode: "every-node"},
+		Restart:  &jobspec.RestartBlock{Mode: "on-failure"},
+	}
+	err := DaemonSetValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for count on DaemonSet, got nil")
+	}
+	if !strings.Contains(err.Error(), "count not allowed") {
+		t.Errorf("error = %q, want 'count not allowed'", err.Error())
+	}
+}
+
+func TestDaemonSetValidator_MissingRestart(t *testing.T) {
+	spec := &jobspec.WorkloadSpec{
+		Kind:     "DaemonSet",
+		Name:     "x",
+		Schedule: &jobspec.ScheduleBlock{Mode: "every-node"},
+	}
+	err := DaemonSetValidator{}.Validate(spec)
+	if err == nil {
+		t.Fatal("expected error for missing restart on DaemonSet, got nil")
+	}
+	if !strings.Contains(err.Error(), "restart block required") {
+		t.Errorf("error = %q, want 'restart block required'", err.Error())
+	}
+}
+
+func TestDaemonSetValidator_NilSpec(t *testing.T) {
+	v := DaemonSetValidator{}
+	if err := v.Validate(nil); err == nil {
+		t.Fatal("expected error for nil spec")
+	}
+}
+
+func TestValidatorFor_EachKind(t *testing.T) {
+	cases := []struct {
+		kind string
+		want string
+	}{
+		{"Job", "schema.JobValidator"},
+		{"Service", "schema.ServiceValidator"},
+		{"DaemonSet", "schema.DaemonSetValidator"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.kind, func(t *testing.T) {
+			v, err := ValidatorFor(tc.kind)
+			if err != nil {
+				t.Fatalf("ValidatorFor(%q): %v", tc.kind, err)
+			}
+			got := fmtType(v)
+			if got != tc.want {
+				t.Errorf("ValidatorFor(%q) type = %q, want %q", tc.kind, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestValidatorFor_UnknownKind(t *testing.T) {
+	_, err := ValidatorFor("CronJob")
+	if err == nil {
+		t.Fatal("expected error for unknown kind, got nil")
+	}
+	if !strings.Contains(err.Error(), "unknown kind") {
+		t.Errorf("error = %q, want 'unknown kind'", err.Error())
+	}
+}
+
+// fmtType returns a readable type name for a validator. Uses fmt.Sprintf
+// with %T rather than reflection to keep the test surface minimal.
+func fmtType(v Validator) string {
+	switch v.(type) {
+	case JobValidator:
+		return "schema.JobValidator"
+	case ServiceValidator:
+		return "schema.ServiceValidator"
+	case DaemonSetValidator:
+		return "schema.DaemonSetValidator"
+	default:
+		return "unknown"
+	}
+}
+
+// Ensure composeErrors returns nil for empty input (covers the
+// short-circuit branch that the validators rely on).
+func TestComposeErrors_Empty(t *testing.T) {
+	if err := composeErrors("schema/X", nil); err != nil {
+		t.Errorf("composeErrors(nil) = %v, want nil", err)
+	}
+	if err := composeErrors("schema/X", []string{}); err != nil {
+		t.Errorf("composeErrors([]) = %v, want nil", err)
+	}
+}
+
+// Ensure the error type returned by composeErrors is a non-nil error
+// when violations are present (guards against accidental nil-return).
+func TestComposeErrors_NonEmpty(t *testing.T) {
+	err := composeErrors("schema/X", []string{"a", "b"})
+	if err == nil {
+		t.Fatal("expected non-nil error")
+	}
+	if !strings.Contains(err.Error(), "a") || !strings.Contains(err.Error(), "b") {
+		t.Errorf("error = %q, want both 'a' and 'b'", err.Error())
+	}
+}
+
+// Compile-time assertion that the validators implement the interface.
+var (
+	_ Validator = JobValidator{}
+	_ Validator = ServiceValidator{}
+	_ Validator = DaemonSetValidator{}
+)