verify(P0c): 4-layer verification PASS — REQ-074

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

.ciagent/CHECKPOINT.json

@@ -1 +1 @@
-{ "phase": "P0a2", "stage": "verify", "milestone": "v0.9", "phase_role": "execution", "updated_at": "2026-08-05T03:15:00Z", "milestone_complete": false, "gates_cleared_this_phase": [], "verify": { "build": "pass", "go_test": "18/18", "bats": "20/20", "gofmt": "clean", "verify_reqs": "90 consistent" } }
+{ "phase": "P0c", "stage": "verify", "milestone": "v0.9", "phase_role": "execution", "updated_at": "2026-08-05T03:35:00Z", "milestone_complete": false, "verify": { "build": "pass", "go_test": "20/20", "bats": "20/20", "gofmt": "clean", "verify_reqs": "90 consistent" } }

internal/cli/job.go

@@ -74,7 +74,7 @@ var jobRunCmd = &cobra.Command{
 			peers := engine.NewPeerRegistry()
 			dispatcher := engine.NewDispatcher(newLogger(), store.NewCapacityRepo(db), peers, exec)
 			specBytes, _ := json.Marshal(map[string]any{
-				"name":    spec.Job.Name,
+				"name":    spec.Name,
 				"command": "/bin/true", // placeholder; full HCL dispatch lands in a later phase
 			})
 			jobID, nodeID, err := dispatcher.Submit(ctx, runTarget, specBytes, runIDKey)
@@ -93,11 +93,11 @@ var jobRunCmd = &cobra.Command{
 
 		job := &model.Job{
 			ID:     uuid.NewString(),
-			Name:   spec.Job.Name,
+			Name:   spec.Name,
 			Spec:   args[0],
 			Status: model.JobStatusPending,
 		}
-		if err := exec.Run(ctx, job, toTaskSpecs(spec.Tasks)); err != nil {
+		if err := exec.Run(ctx, job, workloadToTaskSpecs(spec)); err != nil {
 			if jsonOutput {
 				_ = printJSON(map[string]any{"id": job.ID, "status": "failed", "error": err.Error()})
 				return err
@@ -330,3 +330,22 @@ func toTaskSpecs(in []jobspec.TaskSpec) []engine.TaskSpec {
 	}
 	return out
 }
+
+// workloadToTaskSpecs converts a *WorkloadSpec into the engine.TaskSpec
+// slice consumed by the executor. For the HCL adapter path the runtime
+// block carries the legacy task[0].Command; for the Markdown path the
+// runtime block is the canonical runtime abstraction (P07 will expand
+// this). When Runtime is nil we emit a single no-op task to preserve
+// the legacy "at least one task" invariant.
+func workloadToTaskSpecs(spec *jobspec.WorkloadSpec) []engine.TaskSpec {
+	if spec == nil {
+		return nil
+	}
+	if spec.Runtime == nil {
+		return []engine.TaskSpec{{Name: spec.Name, Command: "/bin/true"}}
+	}
+	return []engine.TaskSpec{{
+		Name:    spec.Name,
+		Command: spec.Runtime.Command,
+	}}
+}

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/dispatch.go

@@ -0,0 +1,147 @@
+package jobspec
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+// ParseFile reads a jobspec file from disk and dispatches on file
+// extension (R-013, REQ-064):
+//
+//   - .md  → ParseMarkdown (canonical Markdown+frontmatter, R-014/R-015)
+//   - .yaml/.yml → ParseMarkdown with the whole file treated as
+//     frontmatter and Body = "" (pure YAML, no Markdown body)
+//   - .hcl → ParseHCL (legacy adapter; wraps the existing HCL parser
+//     and converts Spec{Job, Tasks} into *WorkloadSpec with Kind="Job",
+//     REQ-090 migration window)
+//
+// Unknown extensions return an error. The dispatcher preserves
+// `orca job run old-spec.hcl` during the v0.9→v0.10 migration window
+// (REQ-090).
+func ParseFile(path string) (*WorkloadSpec, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return nil, fmt.Errorf("read spec file: %w", err)
+	}
+	return Dispatch(data, filepath.Base(path))
+}
+
+// ParseHCLFile reads an HCL file and parses it via the legacy HCL parser,
+// returning the legacy *Spec. It is a convenience wrapper retained for
+// tests and direct HCL consumers that need the raw Spec{Job, Tasks}
+// shape during the v0.9→v0.10 migration window (REQ-090).
+//
+// Deprecated: use ParseFile (dispatcher) for new code. HCL is legacy per
+// R-013.
+func ParseHCLFile(path string) (*Spec, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return nil, fmt.Errorf("read spec file: %w", err)
+	}
+	return ParseHCLLegacy(data, filepath.Base(path))
+}
+
+// Dispatch routes raw jobspec bytes on file extension to the
+// appropriate parser. filename is used only for HCL (the HCL decoder
+// needs a filename for error messages and syntax sniffing).
+func Dispatch(data []byte, filename string) (*WorkloadSpec, error) {
+	ext := strings.ToLower(filepath.Ext(filename))
+	switch ext {
+	case ".md":
+		return ParseMarkdown(data)
+	case ".yaml", ".yml":
+		// Pure YAML file: no Markdown body. Treat the whole file as
+		// the frontmatter block. Body is empty (R-015: no body to
+		// preserve).
+		spec, err := parseYAMLFile(data)
+		if err != nil {
+			return nil, err
+		}
+		return spec, nil
+	case ".hcl":
+		return ParseHCL(data, filename)
+	default:
+		return nil, fmt.Errorf("parse jobspec: unknown extension %q (want .md, .yaml, .yml, or .hcl)", ext)
+	}
+}
+
+// parseYAMLFile treats the whole file as a frontmatter block (no
+// surrounding `---` delimiters, no Markdown body). This routes .yaml
+// and .yml files through the same hand-rolled parser as .md.
+func parseYAMLFile(data []byte) (*WorkloadSpec, error) {
+	block := string(data)
+	if strings.TrimSpace(block) == "" {
+		return nil, fmt.Errorf("parse yaml: empty file")
+	}
+	spec, err := parseFrontmatterBlock(block)
+	if err != nil {
+		return nil, err
+	}
+	spec.Body = ""
+	if err := validateWorkload(spec); err != nil {
+		return nil, err
+	}
+	return spec, nil
+}
+
+// ParseHCL parses a legacy HCL jobspec and adapts it into a *WorkloadSpec
+// (REQ-064 adapter, REQ-090 migration window). The existing HCL
+// Spec{Job, Tasks} shape is converted to:
+//
+//	Kind:     "Job"
+//	Name:     spec.Job.Name
+//	Runtime:  {one_of: "process", command: tasks[0].Command}
+//
+// Body is empty (HCL has no Markdown body). The legacy Spec struct and
+// ParseHCLLegacy are retained for direct HCL consumers that have not yet
+// migrated.
+//
+// Deprecated: use the dispatcher (ParseFile/Dispatch). HCL is legacy
+// per R-013; the HCL path is retained only for the v0.9→v0.10 migration
+// window (REQ-090) and will be removed in v1.0.
+func ParseHCL(data []byte, filename string) (*WorkloadSpec, error) {
+	spec, err := ParseHCLLegacy(data, filename)
+	if err != nil {
+		return nil, err
+	}
+	ws := &WorkloadSpec{
+		SpecVersion: "",
+		Kind:        "Job",
+		Name:        spec.Job.Name,
+		Count:       1,
+		Body:        "",
+	}
+	if len(spec.Tasks) > 0 {
+		ws.Runtime = &RuntimeBlock{
+			OneOf:   "process",
+			Command: spec.Tasks[0].Command,
+		}
+	}
+	return ws, nil
+}
+
+// ParseHCLLegacy is the original HCL-only parser retained for direct
+// HCL consumers (e.g. the cli/job.go toTaskSpecs path during the
+// migration window). New code should call ParseHCL (which returns a
+// *WorkloadSpec) or the dispatcher. Deprecated: HCL is legacy per
+// R-013; see ParseHCL.
+func ParseHCLLegacy(data []byte, filename string) (*Spec, error) {
+	var spec Spec
+	if err := hclDecode(filename, data, &spec); err != nil {
+		return nil, fmt.Errorf("decode hcl: %w", err)
+	}
+	if spec.Job.Name == "" {
+		return nil, fmt.Errorf("spec missing job name")
+	}
+	if len(spec.Tasks) == 0 {
+		return nil, fmt.Errorf("spec must have at least one task")
+	}
+	for i, t := range spec.Tasks {
+		if t.Command == "" {
+			return nil, fmt.Errorf("task[%d] (%s) missing command", i, t.Name)
+		}
+	}
+	return &spec, nil
+}

internal/jobspec/dispatch_test.go

@@ -0,0 +1,222 @@
+package jobspec
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func TestDispatch_Markdown(t *testing.T) {
+	input := "---\nkind: Job\nname: md-job\n---\nbody content\n"
+	ws, err := Dispatch([]byte(input), "spec.md")
+	if err != nil {
+		t.Fatalf("Dispatch .md: %v", err)
+	}
+	if ws.Kind != "Job" {
+		t.Errorf("Kind = %q, want Job", ws.Kind)
+	}
+	if ws.Name != "md-job" {
+		t.Errorf("Name = %q, want md-job", ws.Name)
+	}
+	if ws.Body != "body content\n" {
+		t.Errorf("Body = %q, want %q (R-015)", ws.Body, "body content\n")
+	}
+}
+
+func TestDispatch_YAML(t *testing.T) {
+	input := "kind: Service\nname: yaml-svc\nports:\n  - name: http\n    port: 80\n"
+	ws, err := Dispatch([]byte(input), "spec.yaml")
+	if err != nil {
+		t.Fatalf("Dispatch .yaml: %v", err)
+	}
+	if ws.Kind != "Service" {
+		t.Errorf("Kind = %q, want Service", ws.Kind)
+	}
+	if ws.Name != "yaml-svc" {
+		t.Errorf("Name = %q, want yaml-svc", ws.Name)
+	}
+	if ws.Body != "" {
+		t.Errorf("Body = %q, want empty (YAML has no body)", ws.Body)
+	}
+	if len(ws.Ports) != 1 || ws.Ports[0].Name != "http" || ws.Ports[0].Port != 80 {
+		t.Errorf("Ports = %+v, want one http:80", ws.Ports)
+	}
+}
+
+func TestDispatch_YML(t *testing.T) {
+	input := "kind: DaemonSet\nname: yml-ds\n"
+	ws, err := Dispatch([]byte(input), "spec.yml")
+	if err != nil {
+		t.Fatalf("Dispatch .yml: %v", err)
+	}
+	if ws.Kind != "DaemonSet" {
+		t.Errorf("Kind = %q, want DaemonSet", ws.Kind)
+	}
+	if ws.Body != "" {
+		t.Errorf("Body = %q, want empty", ws.Body)
+	}
+}
+
+func TestDispatch_HCLAdapter(t *testing.T) {
+	hcl := `job "demo" {}
+task "build" {
+  command = "/bin/echo"
+  args    = ["hello"]
+}
+`
+	ws, err := Dispatch([]byte(hcl), "spec.hcl")
+	if err != nil {
+		t.Fatalf("Dispatch .hcl: %v", err)
+	}
+	if ws.Kind != "Job" {
+		t.Errorf("Kind = %q, want Job (adapter always sets Job)", ws.Kind)
+	}
+	if ws.Name != "demo" {
+		t.Errorf("Name = %q, want demo (from spec.Job.Name)", ws.Name)
+	}
+	if ws.Runtime == nil {
+		t.Fatal("Runtime is nil; adapter should populate from tasks[0]")
+	}
+	if ws.Runtime.OneOf != "process" {
+		t.Errorf("Runtime.OneOf = %q, want process", ws.Runtime.OneOf)
+	}
+	if ws.Runtime.Command != "/bin/echo" {
+		t.Errorf("Runtime.Command = %q, want /bin/echo (from tasks[0].Command)", ws.Runtime.Command)
+	}
+	if ws.Body != "" {
+		t.Errorf("Body = %q, want empty (HCL has no body)", ws.Body)
+	}
+}
+
+func TestDispatch_HCLAdapterNoTasks(t *testing.T) {
+	hcl := `job "x" {}`
+	_, err := Dispatch([]byte(hcl), "spec.hcl")
+	if err == nil {
+		t.Fatal("expected error for HCL with no tasks")
+	}
+	if !strings.Contains(err.Error(), "at least one task") {
+		t.Errorf("error = %q, want it to contain 'at least one task'", err.Error())
+	}
+}
+
+func TestDispatch_UnknownExtension(t *testing.T) {
+	_, err := Dispatch([]byte("kind: Job\nname: x\n"), "spec.json")
+	if err == nil {
+		t.Fatal("expected error for unknown extension, got nil")
+	}
+	if !strings.Contains(err.Error(), "unknown extension") {
+		t.Errorf("error = %q, want it to contain 'unknown extension'", err.Error())
+	}
+}
+
+func TestDispatch_NoExtension(t *testing.T) {
+	_, err := Dispatch([]byte("kind: Job\nname: x\n"), "spec")
+	if err == nil {
+		t.Fatal("expected error for no extension, got nil")
+	}
+}
+
+func TestDispatch_EmptyYAML(t *testing.T) {
+	_, err := Dispatch([]byte(""), "spec.yaml")
+	if err == nil {
+		t.Fatal("expected error for empty YAML, got nil")
+	}
+}
+
+func TestParseFile_Markdown(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "spec.md")
+	content := "---\nkind: Job\nname: file-md\n---\nbody\n"
+	if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+	ws, err := ParseFile(path)
+	if err != nil {
+		t.Fatalf("ParseFile .md: %v", err)
+	}
+	if ws.Kind != "Job" || ws.Name != "file-md" {
+		t.Errorf("got Kind=%q Name=%q", ws.Kind, ws.Name)
+	}
+	if ws.Body != "body\n" {
+		t.Errorf("Body = %q, want %q", ws.Body, "body\n")
+	}
+}
+
+func TestParseFile_YAML(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "spec.yaml")
+	content := "kind: Service\nname: file-yaml\n"
+	if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+	ws, err := ParseFile(path)
+	if err != nil {
+		t.Fatalf("ParseFile .yaml: %v", err)
+	}
+	if ws.Kind != "Service" || ws.Name != "file-yaml" {
+		t.Errorf("got Kind=%q Name=%q", ws.Kind, ws.Name)
+	}
+	if ws.Body != "" {
+		t.Errorf("Body = %q, want empty", ws.Body)
+	}
+}
+
+func TestParseFile_HCL(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "spec.hcl")
+	content := `job "file-hcl" {}
+task "t" { command = "/bin/true" }
+`
+	if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+	ws, err := ParseFile(path)
+	if err != nil {
+		t.Fatalf("ParseFile .hcl: %v", err)
+	}
+	if ws.Kind != "Job" || ws.Name != "file-hcl" {
+		t.Errorf("got Kind=%q Name=%q", ws.Kind, ws.Name)
+	}
+	if ws.Runtime == nil || ws.Runtime.Command != "/bin/true" {
+		t.Errorf("Runtime.Command = %v, want /bin/true", ws.Runtime)
+	}
+}
+
+func TestParseFile_MissingFile(t *testing.T) {
+	_, err := ParseFile(filepath.Join(t.TempDir(), "nope.md"))
+	if err == nil {
+		t.Fatal("expected error for missing file, got nil")
+	}
+	if !strings.Contains(err.Error(), "read spec file") {
+		t.Errorf("error = %q, want it to contain 'read spec file'", err.Error())
+	}
+}
+
+func TestParseFile_UnknownExtension(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "spec.txt")
+	if err := os.WriteFile(path, []byte("kind: Job\nname: x\n"), 0o644); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+	_, err := ParseFile(path)
+	if err == nil {
+		t.Fatal("expected error for unknown extension, got nil")
+	}
+	if !strings.Contains(err.Error(), "unknown extension") {
+		t.Errorf("error = %q, want 'unknown extension'", err.Error())
+	}
+}
+
+func TestParseHCL_LegacySpec(t *testing.T) {
+	hcl := `job "legacy" {}
+task "t" { command = "/bin/echo" }
+`
+	ws, err := ParseHCL([]byte(hcl), "spec.hcl")
+	if err != nil {
+		t.Fatalf("ParseHCL: %v", err)
+	}
+	if ws.Kind != "Job" || ws.Name != "legacy" {
+		t.Errorf("adapter got Kind=%q Name=%q", ws.Kind, ws.Name)
+	}
+}

internal/jobspec/markdown.go

@@ -0,0 +1,648 @@
+package jobspec
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+// WorkloadSpec is the unified canonical jobspec populated by both the
+// Markdown frontmatter parser (canonical path, R-013/R-014) and the HCL
+// legacy adapter (REQ-064, REQ-090). It is the single shape consumed by
+// downstream phases (P0c schemas, P01 transport). The Markdown body
+// after the closing `---` is preserved verbatim in Body (R-015
+// byte-exact preservation is a load-bearing invariant enforced by the
+// fuzz harness in markdown_fuzz_test.go).
+type WorkloadSpec struct {
+	SpecVersion string
+	Kind        string
+	Name        string
+	Runtime     *RuntimeBlock
+	Count       int
+	Ports       []PortSpec
+	Env         map[string]string
+	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
+// Markdown parser. The full runtime abstraction lands in P07; for now
+// only the one_of/image/command fields are parsed and stored (REQ-064).
+type RuntimeBlock struct {
+	OneOf   string
+	Image   string
+	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
+	HostPort int
+	Port     int
+	Protocol string
+	HostIP   string
+}
+
+// VolumeSpec is a minimal volume mount entry. Fields are stored raw
+// pending the P0c schema work (REQ-074).
+type VolumeSpec struct {
+	Name     string
+	Type     string
+	Source   string
+	Target   string
+	ReadOnly bool
+}
+
+// validKinds is the set of workload kinds accepted by the parser per
+// R-012. Unknown kinds are rejected.
+var validKinds = map[string]bool{
+	"Job":       true,
+	"Service":   true,
+	"DaemonSet": true,
+}
+
+// ParseMarkdown parses a Markdown jobspec with YAML frontmatter into a
+// *WorkloadSpec (R-013 canonical format, R-014 frontmatter). The body
+// after the closing `---` is preserved verbatim in result.Body
+// (R-015 byte-exact, including trailing newlines, CRLF, and BOM in the
+// body). The frontmatter parser is a minimal hand-rolled YAML-ish
+// key:value reader — gopkg.in/yaml.v3 is intentionally not added (same
+// approach as internal/config/markdown.go and internal/ns/parse.go).
+//
+// For .yaml/.yml files (no Markdown body), the dispatcher calls this
+// with the whole file treated as frontmatter and Body left empty (see
+// dispatch.go).
+func ParseMarkdown(data []byte) (*WorkloadSpec, error) {
+	content := string(data)
+	block, body, ok := splitFrontmatter(content)
+	if !ok {
+		return nil, fmt.Errorf("parse markdown: missing frontmatter delimiters")
+	}
+	if strings.TrimSpace(block) == "" {
+		return nil, fmt.Errorf("parse markdown: empty frontmatter")
+	}
+	spec, err := parseFrontmatterBlock(block)
+	if err != nil {
+		return nil, err
+	}
+	spec.Body = body
+	if err := validateWorkload(spec); err != nil {
+		return nil, err
+	}
+	return spec, nil
+}
+
+// splitFrontmatter splits the file content into the YAML frontmatter
+// block and the verbatim body that follows the closing `---`. A leading
+// UTF-8 BOM is stripped from the frontmatter scan (R-015: BOM is not
+// preserved in the frontmatter, but a BOM inside the body would be
+// preserved because the body is verbatim). Returns (block, body, ok).
+// ok is false when no opening `---` delimiter is present, or no closing
+// `---` delimiter is found, or the block is empty after the opening
+// delimiter (handled by caller).
+func splitFrontmatter(content string) (block, body string, ok bool) {
+	// Strip a leading UTF-8 BOM if present (EF BB BF). Only the
+	// frontmatter scan is BOM-stripped; the body is byte-exact, so a BOM
+	// appearing inside the body is preserved verbatim.
+	stripped := content
+	if strings.HasPrefix(stripped, "\uFEFF") {
+		stripped = stripped[len("\uFEFF"):]
+	}
+	// Trim leading horizontal whitespace and newlines before the
+	// opening delimiter. We do NOT trim trailing — body must be exact.
+	trimmed := strings.TrimLeft(stripped, "\r\n\t ")
+	if !strings.HasPrefix(trimmed, "---") {
+		return "", "", false
+	}
+	// The opening delimiter must be on its own line: `---` optionally
+	// followed by a line terminator.
+	rest := trimmed[3:]
+	// The opening `---` must be followed by a newline or end-of-file
+	// (a `---foo` prefix is not a valid delimiter).
+	if len(rest) > 0 && rest[0] != '\n' && rest[0] != '\r' {
+		return "", "", false
+	}
+	rest = strings.TrimLeft(rest, "\r\n")
+
+	// Find the closing delimiter line. The closing `---` must be on its
+	// own line: preceded by a newline (or at the start of `rest`) and
+	// followed by a newline or end-of-file.
+	idx := findClosingDelimiter(rest)
+	if idx < 0 {
+		return "", "", false
+	}
+	block = rest[:idx]
+	// Body is everything after the closing `---` line. The closing
+	// delimiter line itself (including its trailing newline) is NOT
+	// part of the body. We compute the byte offset in the original
+	// `content` so the body is byte-exact (R-015).
+	afterClose := rest[idx:]
+	// afterClose starts with `---`. Strip the delimiter line.
+	delimLen := 3
+	// Account for an optional trailing `...` or spaces on the delimiter
+	// line — the delimiter is `---` followed by anything up to and
+	// including the line terminator. Body starts after the newline.
+	// Find the end of the delimiter line.
+	newlineIdx := strings.IndexAny(afterClose, "\r\n")
+	var bodyStart int
+	if newlineIdx < 0 {
+		// Closing `---` is the last line: body is empty.
+		bodyStart = len(afterClose)
+	} else {
+		// Consume the delimiter line and its line terminator(s).
+		bodyStart = newlineIdx
+		// Strip a single CRLF or LF.
+		if strings.HasPrefix(afterClose[bodyStart:], "\r\n") {
+			bodyStart += 2
+		} else {
+			bodyStart += 1
+		}
+	}
+	body = afterClose[bodyStart:]
+	_ = delimLen
+	return block, body, true
+}
+
+// findClosingDelimiter returns the byte index in `rest` where the
+// closing `---` delimiter line begins, or -1 if none is found. The
+// delimiter must be on its own line: either at the start of `rest` or
+// preceded by a newline, and followed by a newline or end-of-file.
+func findClosingDelimiter(rest string) int {
+	// Special case: closing delimiter at the very start (frontmatter
+	// block is empty). The opening `---` is immediately followed by the
+	// closing `---`. We require the opening to be its own line, so the
+	// closing at index 0 means the opening had no body — invalid (empty
+	// frontmatter handled by caller). We still report it; caller
+	// rejects empty block.
+	for i := 0; i < len(rest); i++ {
+		if rest[i] != '\n' {
+			continue
+		}
+		// Candidate: line after this newline starts with `---`.
+		j := i + 1
+		if j+3 <= len(rest) && rest[j] == '-' && rest[j+1] == '-' && rest[j+2] == '-' {
+			// Must be followed by newline, CRLF, or end-of-file.
+			end := j + 3
+			if end == len(rest) {
+				return j
+			}
+			if rest[end] == '\n' || rest[end] == '\r' {
+				return j
+			}
+		}
+	}
+	// Final candidate: closing delimiter at the very start of rest
+	// (immediately after the opening delimiter + its newline). This
+	// happens when frontmatter is empty: `---\n---\n`. We already trim
+	// leading newlines off `rest`, so if rest itself starts with `---`
+	// AND it's a closing delimiter (followed by newline/EOF), it is the
+	// empty-frontmatter case.
+	if strings.HasPrefix(rest, "---") {
+		end := 3
+		if end == len(rest) {
+			return 0
+		}
+		if rest[end] == '\n' || rest[end] == '\r' {
+			return 0
+		}
+	}
+	return -1
+}
+
+// parseFrontmatterBlock parses a minimal YAML-ish frontmatter block into
+// a *WorkloadSpec (without Body, which is filled by the caller).
+//
+// Supported shapes:
+//
+//	kind: Job
+//	name: my-job
+//	count: 3
+//	runtime:
+//	  one_of: process
+//	  image: docker.io/nginx:latest
+//	  command: /bin/sh -c
+//	ports:
+//	  - name: http
+//	    port: 8080
+//	    host_port: 80
+//	    protocol: tcp
+//	env:
+//	  FOO: bar
+//	  BAR:
+//	    from: secret:my-secret
+//	secrets:
+//	  - db-password
+//	volumes:
+//	  - name: data
+//	    type: host
+//	    source: /data
+//	    target: /data
+//	    read_only: true
+//
+// Comments (# ...) and blank lines are ignored. Quoted scalar values
+// ("..." or '...') are unwrapped. No flow collections except the
+// inline-array form for `secrets`. Multi-line block scalars (|, >) are
+// not supported — by design, to avoid adding a YAML dependency for this
+// small surface.
+func parseFrontmatterBlock(block string) (*WorkloadSpec, error) {
+	spec := &WorkloadSpec{Count: 1}
+	lines := strings.Split(block, "\n")
+
+	type section int
+	const (
+		secNone section = iota
+		secRuntime
+		secPorts
+		secEnv
+		secSecrets
+		secVolumes
+	)
+	cur := secNone
+	var curPort *PortSpec
+	var curVol *VolumeSpec
+
+	flushPort := func() {
+		if curPort != nil {
+			spec.Ports = append(spec.Ports, *curPort)
+			curPort = nil
+		}
+	}
+	flushVol := func() {
+		if curVol != nil {
+			spec.Volumes = append(spec.Volumes, *curVol)
+			curVol = nil
+		}
+	}
+
+	for lineNo, raw := range lines {
+		line := stripComment(raw)
+		if strings.TrimSpace(line) == "" {
+			continue
+		}
+		indent := countIndent(line)
+		trimmed := strings.TrimSpace(line)
+
+		if indent == 0 {
+			// Flush any pending nested entry before switching sections.
+			flushPort()
+			flushVol()
+			cur = secNone
+
+			key, val, ok := splitKV(trimmed)
+			if !ok {
+				return nil, fmt.Errorf("parse markdown: line %d: malformed key:value", lineNo+1)
+			}
+			switch key {
+			case "orca-spec-version":
+				spec.SpecVersion = unquote(val)
+			case "kind":
+				spec.Kind = unquote(val)
+			case "name":
+				spec.Name = unquote(val)
+			case "count":
+				if n, err := strconv.Atoi(strings.TrimSpace(unquote(val))); err == nil {
+					spec.Count = n
+				} else {
+					return nil, fmt.Errorf("parse markdown: line %d: count: %v", lineNo+1, err)
+				}
+			case "runtime":
+				spec.Runtime = &RuntimeBlock{}
+				if strings.TrimSpace(val) != "" {
+					// Inline value (unusual); ignore — runtime is a block.
+				}
+				cur = secRuntime
+			case "ports":
+				cur = secPorts
+			case "env":
+				spec.Env = map[string]string{}
+				cur = secEnv
+			case "secrets":
+				if strings.TrimSpace(val) != "" {
+					arr, err := parseStringArray(val)
+					if err != nil {
+						return nil, fmt.Errorf("parse markdown: line %d: secrets: %w", lineNo+1, err)
+					}
+					spec.Secrets = append(spec.Secrets, arr...)
+					cur = secNone
+				} else {
+					cur = secSecrets
+				}
+			case "volumes":
+				cur = secVolumes
+			default:
+				// Unknown top-level keys are ignored (forward-compat).
+				cur = secNone
+			}
+			continue
+		}
+
+		// Indented line: a nested entry under the current section.
+		switch cur {
+		case secRuntime:
+			if spec.Runtime == nil {
+				spec.Runtime = &RuntimeBlock{}
+			}
+			key, val, ok := splitKV(trimmed)
+			if !ok {
+				continue
+			}
+			switch key {
+			case "one_of":
+				spec.Runtime.OneOf = unquote(val)
+			case "image":
+				spec.Runtime.Image = unquote(val)
+			case "command":
+				spec.Runtime.Command = unquote(val)
+			}
+		case secPorts:
+			if strings.HasPrefix(trimmed, "- ") || trimmed == "-" {
+				flushPort()
+				p := PortSpec{}
+				curPort = &p
+				rest := strings.TrimSpace(strings.TrimPrefix(trimmed, "-"))
+				if rest != "" {
+					applyPortKV(curPort, rest)
+				}
+			} else if curPort != nil {
+				applyPortKV(curPort, trimmed)
+			}
+		case secEnv:
+			key, val, ok := splitKV(trimmed)
+			if !ok {
+				continue
+			}
+			if val == "" {
+				// Nested mapping under env (e.g. `BAR:\n  from: ...`).
+				// Store the raw string for now (REQ-064: store raw).
+				spec.Env[key] = ""
+			} else if strings.HasPrefix(val, "{") && strings.HasSuffix(val, "}") {
+				// Inline object form: `BAR: {from: "secret:..."}`.
+				// Store the raw object string for now.
+				spec.Env[key] = val
+			} else {
+				spec.Env[key] = unquote(val)
+			}
+		case secSecrets:
+			if strings.HasPrefix(trimmed, "- ") || trimmed == "-" {
+				item := strings.TrimSpace(strings.TrimPrefix(trimmed, "-"))
+				if item != "" {
+					spec.Secrets = append(spec.Secrets, unquote(item))
+				}
+			}
+		case secVolumes:
+			if strings.HasPrefix(trimmed, "- ") || trimmed == "-" {
+				flushVol()
+				v := VolumeSpec{}
+				curVol = &v
+				rest := strings.TrimSpace(strings.TrimPrefix(trimmed, "-"))
+				if rest != "" {
+					applyVolumeKV(curVol, rest)
+				}
+			} else if curVol != nil {
+				applyVolumeKV(curVol, trimmed)
+			}
+		}
+	}
+	flushPort()
+	flushVol()
+	return spec, nil
+}
+
+// applyPortKV applies a `key: value` pair to a PortSpec entry.
+func applyPortKV(p *PortSpec, s string) {
+	key, val, ok := splitKV(s)
+	if !ok {
+		return
+	}
+	switch key {
+	case "name":
+		p.Name = unquote(val)
+	case "host_port":
+		if n, err := strconv.Atoi(strings.TrimSpace(unquote(val))); err == nil {
+			p.HostPort = n
+		}
+	case "port":
+		if n, err := strconv.Atoi(strings.TrimSpace(unquote(val))); err == nil {
+			p.Port = n
+		}
+	case "protocol":
+		p.Protocol = unquote(val)
+	case "host_ip":
+		p.HostIP = unquote(val)
+	}
+}
+
+// applyVolumeKV applies a `key: value` pair to a VolumeSpec entry.
+func applyVolumeKV(v *VolumeSpec, s string) {
+	key, val, ok := splitKV(s)
+	if !ok {
+		return
+	}
+	switch key {
+	case "name":
+		v.Name = unquote(val)
+	case "type":
+		v.Type = unquote(val)
+	case "source":
+		v.Source = unquote(val)
+	case "target":
+		v.Target = unquote(val)
+	case "read_only":
+		switch strings.ToLower(strings.TrimSpace(unquote(val))) {
+		case "true", "yes", "on", "1":
+			v.ReadOnly = true
+		}
+	}
+}
+
+// validateWorkload enforces required fields and kind validity (R-012).
+func validateWorkload(spec *WorkloadSpec) error {
+	if spec.Kind == "" {
+		return fmt.Errorf("parse markdown: missing kind")
+	}
+	if !validKinds[spec.Kind] {
+		return fmt.Errorf("parse markdown: kind %q is not one of Job, Service, DaemonSet", spec.Kind)
+	}
+	if strings.TrimSpace(spec.Name) == "" {
+		return fmt.Errorf("parse markdown: missing name")
+	}
+	return nil
+}
+
+// parseStringArray parses an inline YAML flow-array of scalars, e.g.
+// `["a", "b"]` or `['a', 'b']` or `[a, b]`. Empty array `[]` returns nil.
+func parseStringArray(val string) ([]string, error) {
+	val = strings.TrimSpace(val)
+	if val == "" {
+		return nil, nil
+	}
+	if !strings.HasPrefix(val, "[") || !strings.HasSuffix(val, "]") {
+		return nil, fmt.Errorf("expected [..] array, got %q", val)
+	}
+	inner := strings.TrimSpace(val[1 : len(val)-1])
+	if inner == "" {
+		return nil, nil
+	}
+	parts := splitFlowItems(inner)
+	out := make([]string, 0, len(parts))
+	for _, p := range parts {
+		p = strings.TrimSpace(p)
+		if p == "" {
+			continue
+		}
+		out = append(out, unquote(p))
+	}
+	return out, nil
+}
+
+// splitFlowItems splits a comma-separated flow-array body, respecting
+// single and double quotes.
+func splitFlowItems(s string) []string {
+	var out []string
+	inSingle := false
+	inDouble := false
+	start := 0
+	for i := 0; i < len(s); i++ {
+		c := s[i]
+		switch c {
+		case '\'':
+			if !inDouble {
+				inSingle = !inSingle
+			}
+		case '"':
+			if !inSingle {
+				inDouble = !inDouble
+			}
+		case ',':
+			if !inSingle && !inDouble {
+				out = append(out, s[start:i])
+				start = i + 1
+			}
+		}
+	}
+	out = append(out, s[start:])
+	return out
+}
+
+func countIndent(s string) int {
+	n := 0
+	for _, r := range s {
+		if r == ' ' || r == '\t' {
+			n++
+			continue
+		}
+		break
+	}
+	return n
+}
+
+func splitKV(s string) (key, val string, ok bool) {
+	idx := strings.Index(s, ":")
+	if idx < 0 {
+		return "", "", false
+	}
+	key = strings.TrimSpace(s[:idx])
+	val = strings.TrimSpace(s[idx+1:])
+	if key == "" {
+		return "", "", false
+	}
+	return key, val, true
+}
+
+func stripComment(s string) string {
+	inSingle := false
+	inDouble := false
+	for i := 0; i < len(s); i++ {
+		c := s[i]
+		switch c {
+		case '\'':
+			if !inDouble {
+				inSingle = !inSingle
+			}
+		case '"':
+			if !inSingle {
+				inDouble = !inDouble
+			}
+		case '#':
+			if !inSingle && !inDouble {
+				if i == 0 || s[i-1] == ' ' || s[i-1] == '\t' {
+					return s[:i]
+				}
+			}
+		}
+	}
+	return s
+}
+
+func unquote(s string) string {
+	s = strings.TrimSpace(s)
+	if len(s) >= 2 {
+		if (s[0] == '"' && s[len(s)-1] == '"') || (s[0] == '\'' && s[len(s)-1] == '\'') {
+			return s[1 : len(s)-1]
+		}
+	}
+	return s
+}

internal/jobspec/markdown_fuzz_test.go

@@ -0,0 +1,109 @@
+package jobspec
+
+import (
+	"strings"
+	"testing"
+)
+
+// FuzzParseMarkdownRoundTrip is the REQ-067 fuzz harness for R-015
+// byte-exact body preservation. It generates random frontmatter + body
+// combinations, runs ParseMarkdown, and asserts that the parsed Body
+// equals the original body byte-for-byte whenever parsing succeeds.
+// When parsing fails (bad frontmatter), the iteration passes — the
+// parser is allowed to reject malformed input.
+//
+// The seed corpus (added via f.Add) covers adversarial fixtures: CRLF
+// body, BOM prefix, no frontmatter, only-closing-separator, body with
+// `---` inside a code fence, trailing whitespace, empty body. The seed
+// corpus runs as regular tests under `go test` (CI); random input runs
+// only under `go test -fuzz=FuzzParseMarkdownRoundTrip` in a dedicated
+// process.
+func FuzzParseMarkdownRoundTrip(f *testing.F) {
+	// Seed 1: valid frontmatter + simple body.
+	f.Add([]byte("---\nkind: Job\nname: seed1\n---\n# body\n"))
+
+	// Seed 2: CRLF body.
+	f.Add([]byte("---\r\nkind: Job\r\nname: seed2\r\n---\r\n# body\r\nCRLF\r\n"))
+
+	// Seed 3: BOM prefix.
+	f.Add([]byte("\uFEFF---\nkind: Job\nname: seed3\n---\nbody\n"))
+
+	// Seed 4: no frontmatter (just body) — should fail to parse.
+	f.Add([]byte("# just a body\nno frontmatter\n"))
+
+	// Seed 5: frontmatter with only the closing `---` (no opening).
+	f.Add([]byte("body\n---\nmore body\n"))
+
+	// Seed 6: body containing `---` in a code fence.
+	f.Add([]byte("---\nkind: Job\nname: seed6\n---\n```bash\necho '---'\n```\n"))
+
+	// Seed 7: body with trailing whitespace.
+	f.Add([]byte("---\nkind: Job\nname: seed7\n---\nbody with trailing spaces   \n"))
+
+	// Seed 8: empty body.
+	f.Add([]byte("---\nkind: Job\nname: seed8\n---\n"))
+
+	// Seed 9: empty frontmatter (should fail).
+	f.Add([]byte("---\n---\nbody\n"))
+
+	// Seed 10: body with no trailing newline.
+	f.Add([]byte("---\nkind: Job\nname: seed10\n---\nno trailing newline"))
+
+	f.Fuzz(func(t *testing.T, data []byte) {
+		// Reconstruct the body from the input so we can assert
+		// byte-exact round-trip. We do this by re-splitting the
+		// frontmatter using the same logic the parser uses, but only
+		// to extract the expected body. If the input has no valid
+		// frontmatter delimiter pair, ParseMarkdown will return an
+		// error and we pass the iteration.
+		expectedBody := extractExpectedBody(string(data))
+
+		spec, err := ParseMarkdown(data)
+		if err != nil {
+			// Parser rejected the input — acceptable for a fuzz
+			// iteration (the input may be malformed). Pass.
+			return
+		}
+		// R-015: body must be byte-exact.
+		if spec.Body != expectedBody {
+			t.Errorf("R-015 body round-trip mismatch:\n got = %q\nwant = %q", spec.Body, expectedBody)
+		}
+	})
+}
+
+// extractExpectedBody returns the body portion of a Markdown jobspec
+// input using the same delimiter-splitting logic as splitFrontmatter,
+// so the fuzz harness can assert byte-exact preservation independently
+// of the parser's internal extraction. If the input has no valid
+// frontmatter, the result is "" (and ParseMarkdown will error).
+func extractExpectedBody(content string) string {
+	stripped := content
+	if strings.HasPrefix(stripped, "\uFEFF") {
+		stripped = stripped[len("\uFEFF"):]
+	}
+	trimmed := strings.TrimLeft(stripped, "\r\n\t ")
+	if !strings.HasPrefix(trimmed, "---") {
+		return ""
+	}
+	rest := trimmed[3:]
+	if len(rest) > 0 && rest[0] != '\n' && rest[0] != '\r' {
+		return ""
+	}
+	rest = strings.TrimLeft(rest, "\r\n")
+	idx := findClosingDelimiter(rest)
+	if idx < 0 {
+		return ""
+	}
+	afterClose := rest[idx:]
+	newlineIdx := strings.IndexAny(afterClose, "\r\n")
+	if newlineIdx < 0 {
+		return ""
+	}
+	bodyStart := newlineIdx
+	if strings.HasPrefix(afterClose[bodyStart:], "\r\n") {
+		bodyStart += 2
+	} else {
+		bodyStart += 1
+	}
+	return afterClose[bodyStart:]
+}

internal/jobspec/markdown_test.go

@@ -0,0 +1,357 @@
+package jobspec
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestParseMarkdown_FullFrontmatter(t *testing.T) {
+	body := "# Hello\n\nThis is the body.\n\nTrailing newline preserved.\n"
+	input := "---\n" +
+		"orca-spec-version: \"1\"\n" +
+		"kind: Job\n" +
+		"name: my-job\n" +
+		"count: 3\n" +
+		"---\n" +
+		body
+	spec, err := ParseMarkdown([]byte(input))
+	if err != nil {
+		t.Fatalf("ParseMarkdown: %v", err)
+	}
+	if spec.SpecVersion != "1" {
+		t.Errorf("SpecVersion = %q, want %q", spec.SpecVersion, "1")
+	}
+	if spec.Kind != "Job" {
+		t.Errorf("Kind = %q, want %q", spec.Kind, "Job")
+	}
+	if spec.Name != "my-job" {
+		t.Errorf("Name = %q, want %q", spec.Name, "my-job")
+	}
+	if spec.Count != 3 {
+		t.Errorf("Count = %d, want 3", spec.Count)
+	}
+	if spec.Body != body {
+		t.Errorf("Body = %q, want %q (byte-exact, R-015)", spec.Body, body)
+	}
+}
+
+func TestParseMarkdown_BodyByteExactTrailingNewline(t *testing.T) {
+	cases := []struct {
+		name string
+		body string
+	}{
+		{"with_trailing_newline", "# Title\n\nbody\n"},
+		{"with_double_trailing_newline", "# Title\n\nbody\n\n"},
+		{"no_trailing_newline", "# Title\n\nbody"},
+		{"empty_body_with_newline", "\n"},
+		{"only_newlines", "\n\n\n"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			input := "---\nkind: Job\nname: x\n---\n" + tc.body
+			spec, err := ParseMarkdown([]byte(input))
+			if err != nil {
+				t.Fatalf("ParseMarkdown: %v", err)
+			}
+			if spec.Body != tc.body {
+				t.Errorf("Body byte-exact mismatch (R-015):\n got = %q\nwant = %q", spec.Body, tc.body)
+			}
+		})
+	}
+}
+
+func TestParseMarkdown_NoFrontmatter(t *testing.T) {
+	input := "# Just a body\n\nNo frontmatter here."
+	_, err := ParseMarkdown([]byte(input))
+	if err == nil {
+		t.Fatal("expected error for missing frontmatter, got nil")
+	}
+	if !strings.Contains(err.Error(), "frontmatter") {
+		t.Errorf("error = %q, want it to contain 'frontmatter'", err.Error())
+	}
+}
+
+func TestParseMarkdown_EmptyFrontmatter(t *testing.T) {
+	input := "---\n---\n\nbody"
+	_, err := ParseMarkdown([]byte(input))
+	if err == nil {
+		t.Fatal("expected error for empty frontmatter, got nil")
+	}
+	if !strings.Contains(err.Error(), "empty frontmatter") {
+		t.Errorf("error = %q, want it to contain 'empty frontmatter'", err.Error())
+	}
+}
+
+func TestParseMarkdown_UnknownKind(t *testing.T) {
+	input := "---\nkind: CronJob\nname: x\n---\nbody\n"
+	_, err := ParseMarkdown([]byte(input))
+	if err == nil {
+		t.Fatal("expected error for unknown kind, got nil")
+	}
+	if !strings.Contains(err.Error(), "not one of") {
+		t.Errorf("error = %q, want it to contain 'not one of'", err.Error())
+	}
+}
+
+func TestParseMarkdown_MissingName(t *testing.T) {
+	input := "---\nkind: Job\n---\nbody\n"
+	_, err := ParseMarkdown([]byte(input))
+	if err == nil {
+		t.Fatal("expected error for missing name, got nil")
+	}
+	if !strings.Contains(err.Error(), "missing name") {
+		t.Errorf("error = %q, want it to contain 'missing name'", err.Error())
+	}
+}
+
+func TestParseMarkdown_MissingKind(t *testing.T) {
+	input := "---\nname: x\n---\nbody\n"
+	_, err := ParseMarkdown([]byte(input))
+	if err == nil {
+		t.Fatal("expected error for missing kind, got nil")
+	}
+	if !strings.Contains(err.Error(), "missing kind") {
+		t.Errorf("error = %q, want it to contain 'missing kind'", err.Error())
+	}
+}
+
+func TestParseMarkdown_EachValidKind(t *testing.T) {
+	cases := []string{"Job", "Service", "DaemonSet"}
+	for _, kind := range cases {
+		t.Run(kind, func(t *testing.T) {
+			input := "---\nkind: " + kind + "\nname: x\n---\nbody\n"
+			spec, err := ParseMarkdown([]byte(input))
+			if err != nil {
+				t.Fatalf("ParseMarkdown: %v", err)
+			}
+			if spec.Kind != kind {
+				t.Errorf("Kind = %q, want %q", spec.Kind, kind)
+			}
+		})
+	}
+}
+
+func TestParseMarkdown_EnvScalarAndObject(t *testing.T) {
+	input := "---\n" +
+		"kind: Job\n" +
+		"name: x\n" +
+		"env:\n" +
+		"  FOO: bar\n" +
+		"  BAZ: \"qux\"\n" +
+		"  SECRET_REF:\n" +
+		"    from: \"secret:db-password\"\n" +
+		"  INLINE: {from: \"secret:token\"}\n" +
+		"---\nbody\n"
+	spec, err := ParseMarkdown([]byte(input))
+	if err != nil {
+		t.Fatalf("ParseMarkdown: %v", err)
+	}
+	if got := spec.Env["FOO"]; got != "bar" {
+		t.Errorf("env[FOO] = %q, want %q", got, "bar")
+	}
+	if got := spec.Env["BAZ"]; got != "qux" {
+		t.Errorf("env[BAZ] = %q, want %q", got, "qux")
+	}
+	if got := spec.Env["INLINE"]; got != `{from: "secret:token"}` {
+		t.Errorf("env[INLINE] = %q, want the raw object string", got)
+	}
+	if _, ok := spec.Env["SECRET_REF"]; !ok {
+		t.Errorf("env[SECRET_REF] missing; nested from: stored as empty string")
+	}
+}
+
+func TestParseMarkdown_PortsArray(t *testing.T) {
+	input := "---\n" +
+		"kind: Service\n" +
+		"name: web\n" +
+		"ports:\n" +
+		"  - name: http\n" +
+		"    port: 8080\n" +
+		"    host_port: 80\n" +
+		"    protocol: tcp\n" +
+		"  - name: https\n" +
+		"    port: 8443\n" +
+		"    host_port: 443\n" +
+		"---\nbody\n"
+	spec, err := ParseMarkdown([]byte(input))
+	if err != nil {
+		t.Fatalf("ParseMarkdown: %v", err)
+	}
+	if len(spec.Ports) != 2 {
+		t.Fatalf("Ports = %d, want 2", len(spec.Ports))
+	}
+	if spec.Ports[0].Name != "http" || spec.Ports[0].Port != 8080 || spec.Ports[0].HostPort != 80 || spec.Ports[0].Protocol != "tcp" {
+		t.Errorf("Ports[0] = %+v", spec.Ports[0])
+	}
+	if spec.Ports[1].Name != "https" || spec.Ports[1].Port != 8443 || spec.Ports[1].HostPort != 443 {
+		t.Errorf("Ports[1] = %+v", spec.Ports[1])
+	}
+}
+
+func TestParseMarkdown_VolumesArray(t *testing.T) {
+	input := "---\n" +
+		"kind: Job\n" +
+		"name: x\n" +
+		"volumes:\n" +
+		"  - name: data\n" +
+		"    type: host\n" +
+		"    source: /data\n" +
+		"    target: /data\n" +
+		"    read_only: true\n" +
+		"---\nbody\n"
+	spec, err := ParseMarkdown([]byte(input))
+	if err != nil {
+		t.Fatalf("ParseMarkdown: %v", err)
+	}
+	if len(spec.Volumes) != 1 {
+		t.Fatalf("Volumes = %d, want 1", len(spec.Volumes))
+	}
+	v := spec.Volumes[0]
+	if v.Name != "data" || v.Type != "host" || v.Source != "/data" || v.Target != "/data" || !v.ReadOnly {
+		t.Errorf("Volumes[0] = %+v", v)
+	}
+}
+
+func TestParseMarkdown_RuntimeBlock(t *testing.T) {
+	input := "---\n" +
+		"kind: Job\n" +
+		"name: x\n" +
+		"runtime:\n" +
+		"  one_of: process\n" +
+		"  image: docker.io/nginx:latest\n" +
+		"  command: /bin/sh -c 'echo hi'\n" +
+		"---\nbody\n"
+	spec, err := ParseMarkdown([]byte(input))
+	if err != nil {
+		t.Fatalf("ParseMarkdown: %v", err)
+	}
+	if spec.Runtime == nil {
+		t.Fatal("Runtime is nil")
+	}
+	if spec.Runtime.OneOf != "process" {
+		t.Errorf("Runtime.OneOf = %q, want %q", spec.Runtime.OneOf, "process")
+	}
+	if spec.Runtime.Image != "docker.io/nginx:latest" {
+		t.Errorf("Runtime.Image = %q, want %q", spec.Runtime.Image, "docker.io/nginx:latest")
+	}
+	if spec.Runtime.Command != "/bin/sh -c 'echo hi'" {
+		t.Errorf("Runtime.Command = %q, want %q", spec.Runtime.Command, "/bin/sh -c 'echo hi'")
+	}
+}
+
+func TestParseMarkdown_SecretsInlineArray(t *testing.T) {
+	input := "---\nkind: Job\nname: x\nsecrets: [\"db-password\", \"api-token\"]\n---\nbody\n"
+	spec, err := ParseMarkdown([]byte(input))
+	if err != nil {
+		t.Fatalf("ParseMarkdown: %v", err)
+	}
+	if len(spec.Secrets) != 2 {
+		t.Fatalf("Secrets = %d, want 2", len(spec.Secrets))
+	}
+	if spec.Secrets[0] != "db-password" || spec.Secrets[1] != "api-token" {
+		t.Errorf("Secrets = %v, want [db-password api-token]", spec.Secrets)
+	}
+}
+
+func TestParseMarkdown_SecretsBlockArray(t *testing.T) {
+	input := "---\n" +
+		"kind: Job\n" +
+		"name: x\n" +
+		"secrets:\n" +
+		"  - db-password\n" +
+		"  - api-token\n" +
+		"---\nbody\n"
+	spec, err := ParseMarkdown([]byte(input))
+	if err != nil {
+		t.Fatalf("ParseMarkdown: %v", err)
+	}
+	if len(spec.Secrets) != 2 {
+		t.Fatalf("Secrets = %d, want 2", len(spec.Secrets))
+	}
+	if spec.Secrets[0] != "db-password" || spec.Secrets[1] != "api-token" {
+		t.Errorf("Secrets = %v, want [db-password api-token]", spec.Secrets)
+	}
+}
+
+func TestParseMarkdown_CRLFBodyPreserved(t *testing.T) {
+	body := "# Title\r\n\r\nCRLF body.\r\n"
+	input := "---\r\nkind: Job\r\nname: x\r\n---\r\n" + body
+	spec, err := ParseMarkdown([]byte(input))
+	if err != nil {
+		t.Fatalf("ParseMarkdown: %v", err)
+	}
+	if spec.Body != body {
+		t.Errorf("CRLF body not preserved (R-015):\n got = %q\nwant = %q", spec.Body, body)
+	}
+}
+
+func TestParseMarkdown_BOMStrippedFromFrontmatter(t *testing.T) {
+	body := "# body\n"
+	input := "\uFEFF" + "---\nkind: Job\nname: x\n---\n" + body
+	spec, err := ParseMarkdown([]byte(input))
+	if err != nil {
+		t.Fatalf("ParseMarkdown: %v", err)
+	}
+	if spec.Kind != "Job" {
+		t.Errorf("Kind = %q, want Job (BOM should be stripped from frontmatter scan)", spec.Kind)
+	}
+	if spec.Body != body {
+		t.Errorf("Body = %q, want %q", spec.Body, body)
+	}
+}
+
+func TestParseMarkdown_BodyWithCodeFenceContainingDashes(t *testing.T) {
+	body := "```bash\n" +
+		"echo '---'\n" +
+		"echo '--- end ---'\n" +
+		"```\n"
+	input := "---\nkind: Job\nname: x\n---\n" + body
+	spec, err := ParseMarkdown([]byte(input))
+	if err != nil {
+		t.Fatalf("ParseMarkdown: %v", err)
+	}
+	if spec.Body != body {
+		t.Errorf("Body with code-fence --- not preserved (R-015):\n got = %q\nwant = %q", spec.Body, body)
+	}
+}
+
+func TestParseMarkdown_OnlyClosingSeparator(t *testing.T) {
+	input := "no opening\n---\nbody\n"
+	_, err := ParseMarkdown([]byte(input))
+	if err == nil {
+		t.Fatal("expected error for input with only closing separator, got nil")
+	}
+}
+
+func TestParseMarkdown_QuotedValues(t *testing.T) {
+	input := "---\nkind: \"Job\"\nname: 'my-job'\n---\nbody\n"
+	spec, err := ParseMarkdown([]byte(input))
+	if err != nil {
+		t.Fatalf("ParseMarkdown: %v", err)
+	}
+	if spec.Kind != "Job" {
+		t.Errorf("Kind = %q, want Job (double-quoted)", spec.Kind)
+	}
+	if spec.Name != "my-job" {
+		t.Errorf("Name = %q, want my-job (single-quoted)", spec.Name)
+	}
+}
+
+func TestParseMarkdown_CountDefault(t *testing.T) {
+	input := "---\nkind: Job\nname: x\n---\nbody\n"
+	spec, err := ParseMarkdown([]byte(input))
+	if err != nil {
+		t.Fatalf("ParseMarkdown: %v", err)
+	}
+	if spec.Count != 1 {
+		t.Errorf("Count default = %d, want 1", spec.Count)
+	}
+}
+
+func TestParseMarkdown_UnknownKeyIgnored(t *testing.T) {
+	input := "---\nkind: Job\nname: x\nfuture_field: value\n---\nbody\n"
+	_, err := ParseMarkdown([]byte(input))
+	if err != nil {
+		t.Fatalf("ParseMarkdown should ignore unknown keys: %v", err)
+	}
+}

internal/jobspec/spec.go

@@ -2,7 +2,6 @@ package jobspec
 
 import (
 	"fmt"
-	"os"
 	"strings"
 
 	"github.com/hashicorp/hcl/v2"
@@ -10,16 +9,24 @@ import (
 	"github.com/hashicorp/hcl/v2/hclsimple"
 )
 
+// Spec is the legacy HCL-only jobspec shape. It is retained for the
+// v0.9→v0.10 migration window (REQ-090) and is populated by ParseHCLLegacy.
+//
+// Deprecated: HCL is legacy per R-013; new code should consume the
+// unified *WorkloadSpec returned by ParseFile/Dispatch (see
+// dispatch.go and markdown.go).
 type Spec struct {
 	Job   JobSpec    `hcl:"job,block"`
 	Tasks []TaskSpec `hcl:"task,block"`
 }
 
+// JobSpec is the legacy HCL job block.
 type JobSpec struct {
 	Name string `hcl:"name,label"`
 	Type string `hcl:"type,optional"`
 }
 
+// TaskSpec is the legacy HCL task block.
 type TaskSpec struct {
 	Name    string   `hcl:"name,label"`
 	Command string   `hcl:"command"`
@@ -27,34 +34,14 @@ type TaskSpec struct {
 	Env     []string `hcl:"env,optional"`
 }
 
-func ParseFile(path string) (*Spec, error) {
-	data, err := os.ReadFile(path)
-	if err != nil {
-		return nil, fmt.Errorf("read spec file: %w", err)
-	}
-	return Parse(data, path)
-}
-
-func Parse(data []byte, filename string) (*Spec, error) {
-	var spec Spec
-	err := hclsimple.Decode(filename, data, nil, &spec)
-	if err != nil {
-		return nil, fmt.Errorf("decode hcl: %w", err)
-	}
-	if spec.Job.Name == "" {
-		return nil, fmt.Errorf("spec missing job name")
-	}
-	if len(spec.Tasks) == 0 {
-		return nil, fmt.Errorf("spec must have at least one task")
-	}
-	for i, t := range spec.Tasks {
-		if t.Command == "" {
-			return nil, fmt.Errorf("task[%d] (%s) missing command", i, t.Name)
-		}
-	}
-	return &spec, nil
+// hclDecode wraps hclsimple.Decode for testability.
+func hclDecode(filename string, data []byte, spec *Spec) error {
+	return hclsimple.Decode(filename, data, nil, spec)
 }
 
+// Validate is the legacy HCL Spec validator retained for the migration
+// window (REQ-090). New code should use validateWorkload on a
+// *WorkloadSpec.
 func (s *Spec) Validate() error {
 	if strings.TrimSpace(s.Job.Name) == "" {
 		return fmt.Errorf("job name is required")
@@ -65,5 +52,17 @@ func (s *Spec) Validate() error {
 	return nil
 }
 
+// Parse is the original HCL-only entry point retained for backward
+// compatibility with direct HCL callers during the v0.9→v0.10 migration
+// window (REQ-090). New code should call the dispatcher ParseFile (which
+// returns *WorkloadSpec) or ParseHCL (which adapts HCL into
+// *WorkloadSpec).
+//
+// Deprecated: use ParseFile (dispatcher) or ParseHCL (adapter). HCL is
+// legacy per R-013.
+func Parse(data []byte, filename string) (*Spec, error) {
+	return ParseHCLLegacy(data, filename)
+}
+
 var _ = hcl.Diagnostics{}
 var _ = gohcl.DecodeBody

internal/jobspec/spec_test.go

@@ -130,9 +130,9 @@ func TestParse_GoldenFiles(t *testing.T) {
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
 			path := filepath.Join("testdata", tc.file)
-			spec, err := ParseFile(path)
+			spec, err := ParseHCLFile(path)
 			if err != nil {
-				t.Fatalf("ParseFile(%s): %v", tc.file, err)
+				t.Fatalf("ParseHCLFile(%s): %v", tc.file, err)
 			}
 			if spec.Job.Name != tc.wantJob {
 				t.Errorf("job name = %q, want %q", spec.Job.Name, tc.wantJob)
@@ -254,9 +254,9 @@ func TestSpec_Validate(t *testing.T) {
 
 func TestSpec_Validate_RoundTripFromParse(t *testing.T) {
 	path := filepath.Join("testdata", "valid_single_task.hcl")
-	spec, err := ParseFile(path)
+	spec, err := ParseHCLFile(path)
 	if err != nil {
-		t.Fatalf("ParseFile: %v", err)
+		t.Fatalf("ParseHCLFile: %v", err)
 	}
 	if err := spec.Validate(); err != nil {
 		t.Errorf("Validate on parsed spec: %v", err)

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{}
+)