Observe the definition that started each serving process

dag/gunbc/spark/serving_observation_transaction.dag

@@ -69,6 +69,8 @@ type SparkServingProbeRequest
   | ModelBlobDigestsRequest { digests: List<NonEmptyStr> }
   | UserUnitReadRequest
   | UserUnitIsEnabledRequest
+  | RunningDefinitionRequest
+  | InvocationIdentityRequest
   | RuntimeRootRequest
   | RuntimeExecutableRequest
   | RuntimeExecutableBitRequest
@@ -95,6 +97,8 @@ fn spark_serving_probe_request_leg(request: SparkServingProbeRequest) -> SparkSe
     ModelBlobDigestsRequest { digests: _ } => ProbeModelBlobDigests
     UserUnitReadRequest => ProbeUserUnitRead
     UserUnitIsEnabledRequest => ProbeUserUnitIsEnabled
+    RunningDefinitionRequest => ProbeRunningDefinition
+    InvocationIdentityRequest => ProbeInvocationIdentity
     RuntimeRootRequest => ProbeRuntimeRoot
     RuntimeExecutableRequest => ProbeRuntimeExecutable
     RuntimeExecutableBitRequest => ProbeRuntimeExecutableBit
@@ -122,6 +126,8 @@ type SparkServingProbeLeg
   | ProbeModelBlobDigests
   | ProbeUserUnitRead
   | ProbeUserUnitIsEnabled
+  | ProbeRunningDefinition
+  | ProbeInvocationIdentity
   | ProbeRuntimeRoot
   | ProbeRuntimeExecutable
   | ProbeRuntimeExecutableBit
@@ -148,6 +154,8 @@ fn spark_serving_probe_leg_key(leg: SparkServingProbeLeg) -> NonEmptyStr {
     ProbeModelBlobDigests => "model_blob_digests" as NonEmptyStr
     ProbeUserUnitRead => "user_unit_read" as NonEmptyStr
     ProbeUserUnitIsEnabled => "user_unit_is_enabled" as NonEmptyStr
+    ProbeRunningDefinition => "running_definition" as NonEmptyStr
+    ProbeInvocationIdentity => "invocation_identity" as NonEmptyStr
     ProbeRuntimeRoot => "runtime_root" as NonEmptyStr
     ProbeRuntimeExecutable => "runtime_executable" as NonEmptyStr
     ProbeRuntimeExecutableBit => "runtime_executable_bit" as NonEmptyStr
@@ -590,13 +598,20 @@ fn spark_serving_transaction_duplicated_legs(
 // only. An absent capture must not be left for whichever downstream reader happens to want it
 // first, because that reader will read Absent and honestly report unobserved -- and the whole host
 // will look measured-and-empty rather than never-acquired.
+//
+// FOLLOW-UP TRIGGER: the receipt witnesses still spell the evaluated executed-count snapshots as
+// numeric literals. Before another probe leg joins this roster, expose the roster cardinality to
+// those witnesses and assert the relation `executed == required - refused` instead. The present
+// literals are honest for this closed population, but a third added leg can otherwise make the
+// oracle stale until a person recounts it; deriving the relation from this authority removes that
+// hand-maintained second count.
 fn spark_serving_required_probe_legs() -> List<SparkServingProbeLeg> {
   [
     ProbeLoadState, ProbeIsActive, ProbeVersion, ProbeTags, ProbeNvidiaDriver,
     ProbeRuntimeReceipt, ProbeActiveEnter, ProbeDefaultTargetEnter, ProbeUptime,
     ProbeGrants, ProbePasswd, ProbeAuthorizedKeys, ProbeLinger,
     ProbeModelManifest, ProbeModelBlobsListing, ProbeModelManifestDigest, ProbeModelBlobDigests,
-    ProbeUserUnitRead, ProbeUserUnitIsEnabled,
+    ProbeUserUnitRead, ProbeUserUnitIsEnabled, ProbeRunningDefinition, ProbeInvocationIdentity,
     ProbeRuntimeRoot, ProbeRuntimeExecutable, ProbeRuntimeExecutableBit, ProbeRuntimeRootListing,
   ]
 }

dag/gunbc/spark/serving_unit_render.dag

@@ -1,6 +1,8 @@
 module gunbc.spark.serving_unit_render
 
 import std.types { String, NonEmptyStr, FilePath }
+import std.measure { token_count_value, positive_slot_count_value }
+import std.dissolution { DissolutionCondition, unbound_dissolution }
 import std.content_hash { serialize_content_hash, content_hash_of_value }
 import std.bytes { bytes_octets, utf8_encode_bytes }
 import std.encoding { base64_encode, Standard }
@@ -65,6 +67,79 @@ fn spark_serving_exec_start_command(spec: SparkServingUserUnitSpec) -> NonEmptyS
   ], "") as NonEmptyStr
 }
 
+// THE DEFINITION IDENTITY CARRIED BY THE INVOCATION IT STARTS.
+//
+// systemd's loaded unit and the file on disk can both advance while an already-running process
+// keeps the environment and command line from its earlier start.  The process therefore carries
+// the identity of the typed definition that created it.  This is derived before rendering (and so
+// is not a self-hash of a file containing its own hash), while still covering every typed unit-spec
+// field.  A later observer reads this value from /proc/<MainPID>/environ; it never substitutes the
+// file's current identity for the invocation's.
+// THE CURRENT RUNG IS A COMPILE-TIME TRIPWIRE, NOT STRUCTURAL IMPOSSIBILITY.
+//
+// content_hash_of_value currently accepts only NonEmptyStr, so the typed spec must be projected to
+// one canonical string wire. The total SparkServingUserUnitSpec reconstruction below makes adding a
+// field fail compilation here and prompts an identity-domain decision; every serialized projection
+// then reads from that reconstructed value. It remains mechanically possible for an author to add
+// the field to the reconstruction but omit it from the join, so this converts a silent omission into
+// a prompted one rather than making omission unwritable. String fields are base64-framed and field
+// names fixed, so delimiters inside a field cannot make distinct covered specs collide.
+//
+// DISSOLVES WHEN the content-hash primitive accepts structured values and SparkServingUserUnitSpec
+// can be hashed directly. That capability, not a record-fold artifact, makes every present and future
+// field enter the domain by construction. This wire is intentionally NOT rendered unit-file content:
+// those bytes contain the resulting identity and therefore cannot be their own hash domain.
+fn spark_serving_user_unit_spec_identity_wire(spec: SparkServingUserUnitSpec) -> NonEmptyStr {
+  let identity_spec = SparkServingUserUnitSpec {
+    description: spec.description,
+    after: spec.after,
+    service_type: spec.service_type,
+    bind_host_port: spec.bind_host_port,
+    models_root_rel: spec.models_root_rel,
+    runtime_root_rel: spec.runtime_root_rel,
+    exec_subcommand: spec.exec_subcommand,
+    restart: spec.restart,
+    wanted_by: spec.wanted_by,
+    default_context: spec.default_context,
+    serving_slots: spec.serving_slots,
+  }
+  join([
+    "description=", base64_encode(octets: bytes_octets(b: utf8_encode_bytes(s: identity_spec.description as String)), variant: Standard), "\n",
+    "after=", base64_encode(octets: bytes_octets(b: utf8_encode_bytes(s: identity_spec.after as String)), variant: Standard), "\n",
+    "service_type=", base64_encode(octets: bytes_octets(b: utf8_encode_bytes(s: identity_spec.service_type as String)), variant: Standard), "\n",
+    "bind_host_port=", base64_encode(octets: bytes_octets(b: utf8_encode_bytes(s: identity_spec.bind_host_port as String)), variant: Standard), "\n",
+    "models_root_rel=", base64_encode(octets: bytes_octets(b: utf8_encode_bytes(s: identity_spec.models_root_rel)), variant: Standard), "\n",
+    "runtime_root_rel=", base64_encode(octets: bytes_octets(b: utf8_encode_bytes(s: identity_spec.runtime_root_rel)), variant: Standard), "\n",
+    "exec_subcommand=", base64_encode(octets: bytes_octets(b: utf8_encode_bytes(s: identity_spec.exec_subcommand as String)), variant: Standard), "\n",
+    "restart=", base64_encode(octets: bytes_octets(b: utf8_encode_bytes(s: identity_spec.restart as String)), variant: Standard), "\n",
+    "wanted_by=", base64_encode(octets: bytes_octets(b: utf8_encode_bytes(s: identity_spec.wanted_by as String)), variant: Standard), "\n",
+    "default_context=", to_string(token_count_value(t: identity_spec.default_context)), "\n",
+    "serving_slots=", to_string(positive_slot_count_value(slots: identity_spec.serving_slots)), "\n",
+  ], "") as NonEmptyStr
+}
+
+data spark_serving_user_unit_spec_identity_wire_dissolution_trigger: DissolutionCondition = unbound_dissolution(
+  description: "dissolve-on: spark_serving_user_unit_spec_identity_wire -- canonical string projection and total-record tripwire remain because content_hash_of_value accepts only NonEmptyStr. DISSOLVES WHEN the content-hash primitive accepts structured values and hashes SparkServingUserUnitSpec directly, placing every present and future field in the definition-identity domain by construction." as NonEmptyStr,
+)
+
+fn spark_serving_user_unit_definition_identity(spec: SparkServingUserUnitSpec) -> NonEmptyStr {
+  serialize_content_hash(
+    hash: content_hash_of_value(value: spark_serving_user_unit_spec_identity_wire(spec: spec)),
+  ) as NonEmptyStr
+}
+
+data spark_serving_definition_identity_environment_name: NonEmptyStr = "GUNBC_SPARK_SERVING_DEFINITION_IDENTITY" as NonEmptyStr
+
+fn spark_serving_definition_identity_environment_assignment(
+  spec: SparkServingUserUnitSpec,
+) -> NonEmptyStr {
+  join([
+    spark_serving_definition_identity_environment_name as String,
+    "=",
+    spark_serving_user_unit_definition_identity(spec: spec) as String,
+  ], "") as NonEmptyStr
+}
+
 // The runtime and model roots come from the install-paths authority rather than from the spec's
 // relative strings, so the unit and the installer cannot disagree about where the runtime lives.
 fn spark_serving_user_unit_file(spec: SparkServingUserUnitSpec) -> SystemdUnitFile {
@@ -85,6 +160,7 @@ fn spark_serving_user_unit_file(spec: SparkServingUserUnitSpec) -> SystemdUnitFi
       Environment {
         assignment: ollama_num_parallel_env_assignment(slots: spec.serving_slots),
       },
+      Environment { assignment: spark_serving_definition_identity_environment_assignment(spec: spec) },
       WorkingDirectory { path: spark_serving_required_runtime_root() as NonEmptyStr },
       ExecStart { command: spark_serving_exec_start_command(spec: spec) },
       Restart { policy: spec.restart },
@@ -110,6 +186,16 @@ fn spark_serving_desired_user_unit_text() -> String {
   )
 }
 
+fn spark_serving_desired_user_unit_definition_identity() -> NonEmptyStr {
+  spark_serving_user_unit_definition_identity(
+    spec: spark_serving_user_unit_spec(
+      bind_host_port: spark_serving_desired_bind_host_port() as NonEmptyStr,
+      default_context: spark_serving_desired_default_context(),
+      serving_slots: spark_serving_desired_serving_slots(),
+    ),
+  )
+}
+
 // ONE IDENTITY FOR UNIT CONTENT, used by the desired side and the observed side alike.
 //
 // The member's state IS the file's content, so its identity must be a function of bytes and nothing

dag/test/claim/spark/spark_serving_hermetic_acquisition_fixture.dag

@@ -74,6 +74,8 @@ import gunbc.spark.serving_observation_transaction {
   ModelBlobDigestsRequest,
   UserUnitReadRequest,
   UserUnitIsEnabledRequest,
+  RunningDefinitionRequest,
+  InvocationIdentityRequest,
   RuntimeRootRequest,
   RuntimeExecutableRequest,
   RuntimeExecutableBitRequest,
@@ -194,7 +196,7 @@ fn hermetic_capture(request: SparkServingProbeRequest, out: String) -> SparkServ
 // THE COMPLETE POPULATION, BECAUSE A TRANSACTION MISSING ONE OF THEM IS NOT A TRANSACTION.
 //
 // Coverage is checked at identity grain against the required roster rather than by counting, so this
-// enumerates the requests: a count would be satisfied by twenty-three copies of the wrong one.
+// enumerates the requests: a count would be satisfied by twenty-five copies of the wrong one.
 fn hermetic_every_request() -> List<SparkServingProbeRequest>
   admit_callers: [
     decl_ref(module_path: "test.claim.spark.spark_serving_hermetic_acquisition_fixture", decl_name: "hermetic_complete_captures"),
@@ -206,7 +208,7 @@ fn hermetic_every_request() -> List<SparkServingProbeRequest>
     GrantsRequest, PasswdRequest, AuthorizedKeysRequest, LingerRequest,
     ModelManifestRequest, ModelBlobsListingRequest, ModelManifestDigestRequest,
     ModelBlobDigestsRequest { digests: ["deadbeef" as NonEmptyStr] },
-    UserUnitReadRequest, UserUnitIsEnabledRequest,
+    UserUnitReadRequest, UserUnitIsEnabledRequest, RunningDefinitionRequest, InvocationIdentityRequest,
     RuntimeRootRequest, RuntimeExecutableRequest, RuntimeExecutableBitRequest,
     RuntimeRootListingRequest,
   ]

dag/test/claim/spark/spark_serving_observation_transaction_witness_test.dag

@@ -475,7 +475,7 @@ test fn the_receipt_line_says_which_provenance_arm_it_is() -> Bool {
     && !string_contains(s: hermetic, pattern: "observation-attempt=")
 }
 
-// The transaction receipt reports coverage and the per-outcome shape, not a single total: 23
+// The transaction receipt reports coverage and the per-outcome shape, not a single total: 25
 // captures is equally true of a transaction whose every leg the transport refused.
 test fn the_transaction_receipt_reports_coverage_and_outcome_shape() -> Bool {
   match txn_of(captures: complete_captures()) {
@@ -486,7 +486,7 @@ test fn the_transaction_receipt_reports_coverage_and_outcome_shape() -> Bool {
       )
       string_contains(s: line, pattern: "missing=0")
         && string_contains(s: line, pattern: "duplicated=0")
-        && string_contains(s: line, pattern: "executed=23")
+        && string_contains(s: line, pattern: "executed=25")
         && string_contains(s: line, pattern: "transport-refused=0")
         && string_contains(s: line, pattern: "prerequisite-refused=0")
         && string_contains(s: line, pattern: concat("observation-attempt=", concat(hermetic_fleet_attempt_raw, "/srv5/plan")))
@@ -536,7 +536,7 @@ test fn a_transport_refused_leg_moves_the_receipt_counts() -> Bool {
       let line = spark_serving_provenance_receipt_line(
         p: spark_serving_transaction_provenance(txn: t),
       )
-      string_contains(s: line, pattern: "executed=22")
+      string_contains(s: line, pattern: "executed=24")
         && string_contains(s: line, pattern: "transport-refused=1")
         && string_contains(s: line, pattern: "missing=0")
     }
@@ -581,7 +581,7 @@ test fn the_report_body_carries_the_acquisition_receipt_for_an_acquired_host() -
     Absent => false
     Present { value: body } =>
       string_contains(s: body, pattern: "spark-serving-provenance host=srv5 acquired")
-        && string_contains(s: body, pattern: "executed=23")
+        && string_contains(s: body, pattern: "executed=25")
         && string_contains(s: body, pattern: "missing=0")
         && string_contains(s: body, pattern: "duplicated=0")
   }
@@ -613,9 +613,9 @@ test fn changed_capture_outcomes_move_the_line_the_report_emits() -> Bool {
   match acquired_observation_body(captures: swap_is_active(with: refused)) {
     Absent => false
     Present { value: body } =>
-      string_contains(s: body, pattern: "executed=22")
+      string_contains(s: body, pattern: "executed=24")
         && string_contains(s: body, pattern: "transport-refused=1")
-        && !string_contains(s: body, pattern: "executed=23")
+        && !string_contains(s: body, pattern: "executed=25")
   }
 }