[TASK][MVP] Verify Provider Trigger Publication Against the Installed Agent

.github/actions/agent/assert-trigger-publication/action.yml

@@ -0,0 +1,66 @@
+name: Assert durable Agent trigger acceptance
+description: Validate Provider publications in the installed Agent journal.
+
+inputs:
+  container-name:
+    description: Fedora container running the installed Agent.
+    required: true
+  provider-instance-id:
+    description: Provider identity whose durable publications are expected.
+    required: true
+  expected-triggers:
+    description: Space-separated trigger identifiers expected in the journal.
+    required: true
+  evidence-name:
+    description: Safe basename for the resulting JSON evidence.
+    required: true
+  evidence-directory:
+    description: Host directory receiving validated journal evidence.
+    required: true
+  timeout-seconds:
+    description: Maximum time to wait for the exact publication set.
+    required: false
+    default: "60"
+
+runs:
+  using: composite
+  steps:
+    - name: Assert durable trigger publications
+      shell: bash
+      env:
+        ACTION_PATH: ${{ github.action_path }}
+        CONTAINER_NAME: ${{ inputs.container-name }}
+        EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}
+        EVIDENCE_NAME: ${{ inputs.evidence-name }}
+        EXPECTED_TRIGGERS: ${{ inputs.expected-triggers }}
+        PROVIDER_INSTANCE_ID: ${{ inputs.provider-instance-id }}
+        TIMEOUT_SECONDS: ${{ inputs.timeout-seconds }}
+      run: |
+        set -euo pipefail
+        if [[ ! "${EVIDENCE_NAME}" =~ ^[a-z][a-z0-9-]{0,40}$ \
+          || ! "${TIMEOUT_SECONDS}" =~ ^[1-9][0-9]*$ ]]; then
+          echo "Trigger assertion inputs are invalid" >&2
+          exit 2
+        fi
+        read -r -a expected_triggers <<< "${EXPECTED_TRIGGERS}"
+        if (( ${#expected_triggers[@]} == 0 )); then
+          echo "At least one expected trigger is required" >&2
+          exit 2
+        fi
+
+        mkdir -p "${EVIDENCE_DIRECTORY}"
+        verifier_directory=/opt/stateforge-provider-tests/trigger-publication
+        verifier="${verifier_directory}/assert.py"
+        docker exec "${CONTAINER_NAME}" mkdir -p "${verifier_directory}"
+        docker cp "${ACTION_PATH}/assert.py" "${CONTAINER_NAME}:${verifier}"
+
+        arguments=()
+        for identifier in "${expected_triggers[@]}"; do
+          arguments+=(--expected-trigger "${identifier}")
+        done
+        docker exec "${CONTAINER_NAME}" /usr/bin/python3 "${verifier}" \
+          --database /var/lib/stateforge-agent/journal/agent.db \
+          --provider-instance-id "${PROVIDER_INSTANCE_ID}" \
+          --timeout-seconds "${TIMEOUT_SECONDS}" \
+          "${arguments[@]}" \
+          > "${EVIDENCE_DIRECTORY}/${EVIDENCE_NAME}.json"

.github/actions/agent/assert-trigger-publication/assert.py

@@ -0,0 +1,130 @@
+"""Validate durable Trigger Publication evidence in an installed Agent."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import sqlite3
+import sys
+import time
+from pathlib import Path
+from uuid import UUID
+
+from stateforge_protocol.sfp.trigger_publication.v1 import (
+    trigger_publication_pb2,
+    validation,
+)
+
+
+def _arguments() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--database", type=Path, required=True)
+    parser.add_argument("--provider-instance-id", type=UUID, required=True)
+    parser.add_argument("--expected-trigger", action="append", required=True)
+    parser.add_argument("--timeout-seconds", type=int, required=True)
+    return parser.parse_args()
+
+
+def _rows(
+    database: Path,
+    provider_instance_id: UUID,
+) -> list[sqlite3.Row]:
+    connection = sqlite3.connect(f"file:{database}?mode=ro", uri=True)
+    connection.row_factory = sqlite3.Row
+    try:
+        return list(
+            connection.execute(
+                """
+                SELECT publication_id, content_fingerprint, publication,
+                       trigger_identifier, observed_at, endpoint_generation,
+                       session_id, manifest_revision, accepted_at,
+                       forwarding_state
+                FROM trigger_publications
+                WHERE provider_instance_id = ?
+                ORDER BY accepted_at, publication_id
+                """,
+                (str(provider_instance_id),),
+            )
+        )
+    finally:
+        connection.close()
+
+
+def _validated_evidence(
+    rows: list[sqlite3.Row],
+    expected_triggers: list[str],
+) -> dict[str, object]:
+    actual = [str(row["trigger_identifier"]) for row in rows]
+    if sorted(actual) != sorted(expected_triggers):
+        raise ValueError(
+            "Durable publications do not match the expected trigger set"
+        )
+
+    publications: list[dict[str, object]] = []
+    for row in rows:
+        publication = trigger_publication_pb2.TriggerPublication.FromString(
+            row["publication"]
+        )
+        validation.validate_trigger_publication(publication)
+        fingerprint = validation.compute_publication_content_fingerprint(
+            publication
+        )
+        if publication.publication_id != row["publication_id"]:
+            raise ValueError("Stored publication identity is inconsistent")
+        if publication.trigger_identifier != row["trigger_identifier"]:
+            raise ValueError("Stored trigger identifier is inconsistent")
+        if fingerprint != row["content_fingerprint"]:
+            raise ValueError("Stored publication fingerprint is inconsistent")
+        if row["forwarding_state"] != "pending":
+            raise ValueError(
+                "Accepted publication lost pending responsibility"
+            )
+        publications.append(
+            {
+                "accepted_at": row["accepted_at"],
+                "content_fingerprint": fingerprint,
+                "endpoint_generation": row["endpoint_generation"],
+                "forwarding_state": row["forwarding_state"],
+                "manifest_revision": row["manifest_revision"],
+                "observed_at": row["observed_at"],
+                "publication_id": publication.publication_id,
+                "session_id_sha256": hashlib.sha256(
+                    row["session_id"].encode()
+                ).hexdigest(),
+                "trigger_identifier": publication.trigger_identifier,
+            }
+        )
+    return {
+        "publication_count": len(publications),
+        "publications": publications,
+    }
+
+
+def main() -> None:
+    """Wait for, validate, and emit the expected durable evidence."""
+    arguments = _arguments()
+    if arguments.timeout_seconds <= 0:
+        raise ValueError("timeout-seconds must be positive")
+    deadline = time.monotonic() + arguments.timeout_seconds
+    last_error: Exception | None = None
+    while time.monotonic() < deadline:
+        try:
+            evidence = _validated_evidence(
+                _rows(arguments.database, arguments.provider_instance_id),
+                arguments.expected_trigger,
+            )
+        except (OSError, sqlite3.Error, ValueError) as error:
+            last_error = error
+            time.sleep(0.25)
+            continue
+        sys.stdout.write(json.dumps(evidence, indent=2, sort_keys=True))
+        sys.stdout.write("\n")
+        return
+    raise RuntimeError(
+        "Installed Agent did not retain the expected trigger publications"
+    ) from last_error
+
+
+if __name__ == "__main__":
+    main()

.github/actions/agent/start/action.yml

@@ -10,10 +10,19 @@ inputs:
     required: false
     default: "60"
 
+outputs:
+  service-state:
+    description: Final systemd active state.
+    value: ${{ steps.start.outputs.service-state }}
+  service-invocation-id:
+    description: systemd invocation that reached readiness.
+    value: ${{ steps.start.outputs.service-invocation-id }}
+
 runs:
   using: composite
   steps:
     - name: Start Agent service
+      id: start
       shell: bash
       env:
         CONTAINER_NAME: ${{ inputs.container-name }}
@@ -27,9 +36,18 @@ runs:
         while (( SECONDS < deadline )); do
           state="$(docker exec "${CONTAINER_NAME}" \
             systemctl is-active stateforge-agent.service 2>/dev/null || true)"
-          if [[ "${state}" == "active" ]] && docker exec \
+          invocation_id="$(docker exec "${CONTAINER_NAME}" \
+            systemctl show stateforge-agent.service \
+              --property InvocationID --value 2>/dev/null || true)"
+          if [[ "${state}" == "active" \
+            && "${invocation_id}" =~ ^[0-9a-f]{32}$ ]] && docker exec \
             "${CONTAINER_NAME}" journalctl -u stateforge-agent.service \
+              "_SYSTEMD_INVOCATION_ID=${invocation_id}" \
               --no-pager -o cat | grep -Fq 'StateForge Agent started'; then
+            {
+              echo "service-state=${state}"
+              echo "service-invocation-id=${invocation_id}"
+            } >> "${GITHUB_OUTPUT}"
             exit 0
           fi
           if [[ "${state}" == "failed" || "${state}" == "inactive" ]]; then

.github/actions/provider/start/action.yml

@@ -32,6 +32,22 @@ inputs:
     description: Enable Capability cancellation and message-limit fixtures.
     required: false
     default: "false"
+  trigger-publication:
+    description: Enable the dedicated Trigger Publication fixture controls.
+    required: false
+    default: "false"
+  drop-trigger-acks-until-marker:
+    description: Discard terminal trigger ACKs until the state marker exists.
+    required: false
+    default: "false"
+  first-trigger-publication-id:
+    description: Deterministic UUIDv4 for the first fixture publication.
+    required: false
+    default: ""
+  trigger-startup-sequence:
+    description: Payload sequence emitted by the decorated startup source.
+    required: false
+    default: "1"
 
 outputs:
   unit-name:
@@ -52,6 +68,9 @@ outputs:
   event-log:
     description: Lifecycle event log inside Fedora.
     value: ${{ steps.start.outputs.event-log }}
+  service-invocation-id:
+    description: systemd invocation that published this endpoint.
+    value: ${{ steps.start.outputs.service-invocation-id }}
   state-directory:
     description: Stable Provider installation state inside Fedora.
     value: ${{ steps.start.outputs.state-directory }}
@@ -68,13 +87,17 @@ runs:
       env:
         CONTAINER_NAME: ${{ inputs.container-name }}
         CAPABILITY_BOUNDARIES: ${{ inputs.capability-boundaries }}
+        DROP_TRIGGER_ACKS: ${{ inputs.drop-trigger-acks-until-marker }}
         DISPLAY_NAME: ${{ inputs.display-name }}
         EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}
         FIXTURE_PATH: ${{ inputs.fixture-path }}
         IDENTITY: ${{ inputs.identity }}
         INSTANCE_NAME: ${{ inputs.instance-name }}
+        FIRST_TRIGGER_PUBLICATION_ID: ${{ inputs.first-trigger-publication-id }}
         STATE_NAME: ${{ inputs.state-name }}
         TIMEOUT_SECONDS: ${{ inputs.timeout-seconds }}
+        TRIGGER_PUBLICATION: ${{ inputs.trigger-publication }}
+        TRIGGER_STARTUP_SEQUENCE: ${{ inputs.trigger-startup-sequence }}
       run: |
         set -euo pipefail
         if [[ ! "${INSTANCE_NAME}" =~ ^[a-z][a-z0-9-]{0,20}$ || \
@@ -137,6 +160,47 @@ runs:
             ;;
         esac
 
+        trigger_arguments=()
+        case "${TRIGGER_PUBLICATION}" in
+          true)
+            if [[ "${DROP_TRIGGER_ACKS}" != "true" \
+              && "${DROP_TRIGGER_ACKS}" != "false" ]]; then
+              echo "drop-trigger-acks-until-marker must be true or false" >&2
+              exit 2
+            fi
+            if [[ -n "${FIRST_TRIGGER_PUBLICATION_ID}" \
+              && ! "${FIRST_TRIGGER_PUBLICATION_ID}" =~ ^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$ ]]; then
+              echo "first-trigger-publication-id must be a lowercase UUIDv4" >&2
+              exit 2
+            fi
+            if [[ ! "${TRIGGER_STARTUP_SEQUENCE}" =~ ^[1-9][0-9]*$ ]]; then
+              echo "trigger-startup-sequence must be positive" >&2
+              exit 2
+            fi
+            trigger_arguments+=(
+              --trigger-publication
+              --trigger-startup-sequence "${TRIGGER_STARTUP_SEQUENCE}"
+            )
+            if [[ "${DROP_TRIGGER_ACKS}" == "true" ]]; then
+              trigger_arguments+=(--drop-trigger-acks-until-marker)
+              docker exec "${CONTAINER_NAME}" rm -f \
+                "${state_directory}/accept-trigger-acks"
+            fi
+            if [[ -n "${FIRST_TRIGGER_PUBLICATION_ID}" ]]; then
+              trigger_arguments+=(
+                --first-trigger-publication-id \
+                "${FIRST_TRIGGER_PUBLICATION_ID}"
+              )
+            fi
+            ;;
+          false)
+            ;;
+          *)
+            echo "trigger-publication must be true or false" >&2
+            exit 2
+            ;;
+        esac
+
         docker exec "${CONTAINER_NAME}" systemd-run \
           --unit "${unit_name}" \
           --property "User=${service_user}" \
@@ -150,7 +214,8 @@ runs:
           --state-directory "${state_directory}" \
           --event-log "${event_log}" \
           --target-path "${target_path}" \
-          "${capability_arguments[@]}"
+          "${capability_arguments[@]}" \
+          "${trigger_arguments[@]}"
 
         deadline=$((SECONDS + TIMEOUT_SECONDS))
         while (( SECONDS < deadline )); do
@@ -184,16 +249,35 @@ runs:
 
         socket_path="$(docker exec "${CONTAINER_NAME}" \
           jq -r .transport.socket_path "${descriptor}")"
+        invocation_id="$(docker exec "${CONTAINER_NAME}" \
+          systemctl show "${unit_name}" \
+            --property InvocationID --value)"
+        main_pid="$(docker exec "${CONTAINER_NAME}" \
+          systemctl show "${unit_name}" --property MainPID --value)"
+        if [[ ! "${invocation_id}" =~ ^[0-9a-f]{32}$ \
+          || ! "${main_pid}" =~ ^[1-9][0-9]*$ ]]; then
+          echo "Provider service identity is incomplete" >&2
+          exit 1
+        fi
         docker exec "${CONTAINER_NAME}" test -S "${socket_path}"
         docker exec "${CONTAINER_NAME}" cat "${descriptor}" \
           > "${EVIDENCE_DIRECTORY}/descriptor.json"
+        printf '%s\n' \
+          "unit-name=${unit_name}" \
+          "service-invocation-id=${invocation_id}" \
+          "main-pid=${main_pid}" \
+          "provider-instance-id=${provider_id}" \
+          "endpoint-generation=${generation}" \
+          "socket-path=${socket_path}" \
+          > "${EVIDENCE_DIRECTORY}/process-identity.txt"
         {
           echo "unit-name=${unit_name}"
           echo "descriptor-path=${descriptor}"
           echo "provider-instance-id=${provider_id}"
           echo "endpoint-generation=${generation}"
           echo "socket-path=${socket_path}"
           echo "event-log=${event_log}"
+          echo "service-invocation-id=${invocation_id}"
           echo "state-directory=${state_directory}"
           echo "target-path=${target_path}"
         } >> "${GITHUB_OUTPUT}"