feat(syva-cp): nodes, assignments, and streaming subscriptions

AGENT.md

@@ -5,6 +5,22 @@ Read CLAUDE.md first. This file covers *how to work*, not *what exists*.
 
 ---
 
+## Heartbeat Audit Exception
+
+Node heartbeats are the single control-plane operation exempt from ADR 0003
+Rule 8's "audit every mutation" discipline.
+
+- Heartbeats happen many times per minute per node.
+- `last_seen_at` is telemetry, not policy.
+- Auditing every heartbeat would drown the audit log in low-value noise.
+
+Heartbeats still write a `control_plane_events` row with
+`event_type = 'node.heartbeat'`, preserving the causal spine. They do **not**
+write to `audit_log`. This exception is specific to heartbeats and must not be
+copied to any other operation.
+
+---
+
 ## Mental Model: How to Think About This Codebase
 
 Syvä is a kernel enforcement boundary. Every line of eBPF code runs in a

syva-cp/Cargo.toml

@@ -53,6 +53,7 @@ thiserror = { workspace = true }
 clap = { workspace = true }
 sha2 = "0.10"
 hex = "0.4"
+ulid = "1.1"
 
 [dev-dependencies]
 # sqlx::test macro runs each test in an isolated database.

syva-cp/README.md

@@ -139,3 +139,51 @@ brew install grpcurl
 curl -L https://github.com/fullstorydev/grpcurl/releases/download/v1.9.1/grpcurl_1.9.1_linux_x86_64.tar.gz \
   | tar xz -C /tmp && sudo mv /tmp/grpcurl /usr/local/bin/
 ```
+
+## E2E Smoke Test — Nodes and Assignments (Session 3)
+
+```bash
+# Assume syva-cp is already running locally from the setup above.
+
+# 1. Create a team
+TEAM_ID=$(grpcurl -plaintext -d '{"name":"platform"}' \
+    localhost:50051 syva.control.v1.TeamService/CreateTeam \
+    | jq -r '.team.id')
+echo "team: $TEAM_ID"
+
+# 2. Create a zone with a label-based selector
+ZONE_ID=$(grpcurl -plaintext -d "{
+    \"team_id\":\"$TEAM_ID\",
+    \"name\":\"agents\",
+    \"policy_json\":\"{\\\"allowed_zones\\\":[]}\",
+    \"selector_json\":\"{\\\"match_labels\\\":{\\\"tier\\\":\\\"prod\\\"}}\"
+}" localhost:50051 syva.control.v1.ZoneService/CreateZone \
+    | jq -r '.zone.id')
+echo "zone: $ZONE_ID"
+
+# 3. Generate a node_id on the node side
+NODE_ID=$(uuidgen)
+echo "node: $NODE_ID"
+
+# 4. Register the node with a matching label
+grpcurl -plaintext -d "{
+    \"node_name\":\"n01\",
+    \"proposed_id\":\"$NODE_ID\",
+    \"labels\":{\"tier\":\"prod\"}
+}" localhost:50051 syva.control.v1.NodeService/RegisterNode
+
+# 5. In another shell, subscribe to assignment updates
+grpcurl -plaintext -d "{\"node_id\":\"$NODE_ID\"}" \
+    localhost:50051 syva.control.v1.AssignmentService/SubscribeAssignments
+
+# Expected: one FULL_SNAPSHOT with one assignment
+
+# 6. Flip the labels so the node no longer matches
+grpcurl -plaintext -d "{
+    \"node_id\":\"$NODE_ID\",
+    \"if_version\":1,
+    \"labels\":{\"tier\":\"dev\"}
+}" localhost:50051 syva.control.v1.NodeService/SetNodeLabels
+
+# Expected on the stream: a REMOVE for the zone
+```

syva-cp/migrations/20260420140000_nodes_and_assignments.sql

@@ -0,0 +1,79 @@
+-- Nodes: registered node agents
+CREATE TABLE nodes (
+    id UUID PRIMARY KEY,
+    node_name TEXT UNIQUE NOT NULL,
+    cluster_id TEXT NULL,
+    status TEXT NOT NULL CHECK (status IN ('online', 'offline', 'decommissioning', 'decommissioned')),
+    fingerprint TEXT NULL,
+    last_seen_at TIMESTAMPTZ NULL,
+    last_heartbeat_event_id UUID NULL REFERENCES control_plane_events(id),
+    current_token_expires_at TIMESTAMPTZ NULL,
+    capabilities_json JSONB NOT NULL DEFAULT '{}',
+    metadata_json JSONB NOT NULL DEFAULT '{}',
+    created_at TIMESTAMPTZ NOT NULL,
+    updated_at TIMESTAMPTZ NOT NULL,
+    version BIGINT NOT NULL,
+    caused_by_event_id UUID NULL REFERENCES control_plane_events(id)
+);
+
+CREATE INDEX idx_nodes_status ON nodes(status) WHERE status IN ('online', 'offline');
+CREATE INDEX idx_nodes_last_seen ON nodes(last_seen_at DESC);
+CREATE UNIQUE INDEX idx_nodes_fingerprint ON nodes(fingerprint) WHERE fingerprint IS NOT NULL;
+
+-- Node labels: key/value pairs used by selector matching.
+CREATE TABLE node_labels (
+    node_id UUID NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
+    key TEXT NOT NULL,
+    value TEXT NOT NULL,
+    PRIMARY KEY (node_id, key)
+);
+
+CREATE INDEX idx_node_labels_key_value ON node_labels(key, value);
+
+-- Assignments: desired state, which zones should be present on which nodes.
+CREATE TABLE assignments (
+    id UUID PRIMARY KEY,
+    zone_id UUID NOT NULL REFERENCES zones(id),
+    node_id UUID NOT NULL REFERENCES nodes(id),
+    status TEXT NOT NULL CHECK (status IN (
+        'desired', 'applying', 'applied', 'drifted', 'removing', 'removed', 'failed'
+    )),
+    desired_policy_id UUID NOT NULL REFERENCES policies(id),
+    desired_zone_version BIGINT NOT NULL,
+    actual_policy_id UUID NULL REFERENCES policies(id),
+    actual_zone_version BIGINT NULL,
+    last_reported_at TIMESTAMPTZ NULL,
+    error_json JSONB NULL,
+    created_at TIMESTAMPTZ NOT NULL,
+    updated_at TIMESTAMPTZ NOT NULL,
+    version BIGINT NOT NULL,
+    caused_by_event_id UUID NOT NULL REFERENCES control_plane_events(id),
+    UNIQUE (zone_id, node_id)
+);
+
+CREATE INDEX idx_assignments_node_status ON assignments(node_id, status);
+CREATE INDEX idx_assignments_zone ON assignments(zone_id);
+CREATE INDEX idx_assignments_status ON assignments(status)
+    WHERE status IN ('desired', 'applying', 'drifted', 'failed');
+
+-- assignment_versions: append-only history snapshots.
+CREATE TABLE assignment_versions (
+    id UUID PRIMARY KEY,
+    assignment_id UUID NOT NULL REFERENCES assignments(id),
+    version BIGINT NOT NULL,
+    snapshot_json JSONB NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL,
+    caused_by_event_id UUID NOT NULL REFERENCES control_plane_events(id),
+    UNIQUE (assignment_id, version)
+);
+
+CREATE INDEX idx_assignment_versions_assignment_version
+    ON assignment_versions(assignment_id, version DESC);
+
+CREATE TRIGGER assignment_versions_no_update
+    BEFORE UPDATE ON assignment_versions
+    FOR EACH ROW EXECUTE FUNCTION reject_mutation();
+
+CREATE TRIGGER assignment_versions_no_delete
+    BEFORE DELETE ON assignment_versions
+    FOR EACH ROW EXECUTE FUNCTION reject_mutation();