From 800dcd553ab78ee9f8d0a45f45758eeb15bf8cb6 Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Sun, 1 Nov 2026 11:53:08 +0200 Subject: [PATCH 1/2] test: verify trigger publication with agent --- .../assert-trigger-publication/action.yml | 66 +++ .../assert-trigger-publication/assert.py | 130 ++++ .github/actions/agent/start/action.yml | 20 +- .github/actions/provider/start/action.yml | 86 ++- .../trigger/installed-agent/action.yml | 553 ++++++++++++++++++ .../trigger/installed-agent/fixture.py | 237 ++++++++ .../scenarios/trigger/publication/verify.py | 315 ++++++++-- .../workflows/provider-runtime-scenarios.yml | 36 ++ docs/ci-automation.md | 55 +- tests/test_ci_automation.py | 69 ++- 10 files changed, 1495 insertions(+), 72 deletions(-) create mode 100644 .github/actions/agent/assert-trigger-publication/action.yml create mode 100644 .github/actions/agent/assert-trigger-publication/assert.py create mode 100644 .github/actions/scenarios/trigger/installed-agent/action.yml create mode 100644 .github/actions/scenarios/trigger/installed-agent/fixture.py diff --git a/.github/actions/agent/assert-trigger-publication/action.yml b/.github/actions/agent/assert-trigger-publication/action.yml new file mode 100644 index 0000000..c33ee39 --- /dev/null +++ b/.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" diff --git a/.github/actions/agent/assert-trigger-publication/assert.py b/.github/actions/agent/assert-trigger-publication/assert.py new file mode 100644 index 0000000..de6acac --- /dev/null +++ b/.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() diff --git a/.github/actions/agent/start/action.yml b/.github/actions/agent/start/action.yml index 63f64b2..b1f4f65 100644 --- a/.github/actions/agent/start/action.yml +++ b/.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 diff --git a/.github/actions/provider/start/action.yml b/.github/actions/provider/start/action.yml index 1f427fa..814cf04 100644 --- a/.github/actions/provider/start/action.yml +++ b/.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,9 +249,27 @@ 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}" @@ -194,6 +277,7 @@ runs: 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}" diff --git a/.github/actions/scenarios/trigger/installed-agent/action.yml b/.github/actions/scenarios/trigger/installed-agent/action.yml new file mode 100644 index 0000000..f780b01 --- /dev/null +++ b/.github/actions/scenarios/trigger/installed-agent/action.yml @@ -0,0 +1,553 @@ +name: Verify Trigger Publication with the installed Agent +description: Exercise the exact Provider wheel against the exact Agent RPM. + +inputs: + agent-artifact-directory: + description: Directory containing the exact Agent RPM build. + required: true + agent-test-image-directory: + description: Bundled exact Agent Fedora test-image definition. + required: true + provider-artifact-directory: + description: Directory containing the exact Provider build. + required: true + protocol-artifact-directory: + description: Directory containing the exact Protocol build. + required: true + runtime-directory: + description: Scenario-owned directory under the runner temp directory. + required: true + evidence-directory: + description: Directory receiving scenario evidence. + required: true + +runs: + using: composite + steps: + - name: Install exact runtime artifacts + id: system + uses: ./.github/actions/agent/install + with: + agent-artifact-directory: ${{ inputs.agent-artifact-directory }} + agent-test-image-directory: ${{ inputs.agent-test-image-directory }} + provider-artifact-directory: ${{ inputs.provider-artifact-directory }} + protocol-artifact-directory: ${{ inputs.protocol-artifact-directory }} + evidence-directory: ${{ inputs.evidence-directory }}/installation + + - name: Start installed Agent + id: agent + uses: ./.github/actions/agent/start + with: + container-name: ${{ steps.system.outputs.container-name }} + + - name: Stage Trigger Publication fixture + id: fixture + shell: bash + env: + ACTION_PATH: ${{ github.action_path }} + RUNTIME_DIRECTORY: ${{ inputs.runtime-directory }} + run: | + set -euo pipefail + root="${RUNTIME_DIRECTORY}/trigger-provider" + mkdir -p "${root}" + install -m 0644 "${ACTION_PATH}/fixture.py" "${root}/provider.py" + { + echo "fixture-path=${root}/provider.py" + echo "instance-name=trigger-provider" + echo "state-name=trigger-provider" + } >> "${GITHUB_OUTPUT}" + + - name: Start real Provider and lose its first acknowledgement + id: provider + uses: ./.github/actions/provider/start + with: + container-name: ${{ steps.system.outputs.container-name }} + fixture-path: ${{ steps.fixture.outputs.fixture-path }} + instance-name: ${{ steps.fixture.outputs.instance-name }} + state-name: ${{ steps.fixture.outputs.state-name }} + identity: stateforge.test.trigger + trigger-publication: "true" + drop-trigger-acks-until-marker: "true" + first-trigger-publication-id: 22222222-2222-4222-8222-222222222222 + evidence-directory: ${{ inputs.evidence-directory }}/provider + + - name: Assert Agent-accepted session + uses: ./.github/actions/agent/assert-session + with: + container-name: ${{ steps.system.outputs.container-name }} + provider-instance-id: ${{ steps.provider.outputs.provider-instance-id }} + endpoint-generation: ${{ steps.provider.outputs.endpoint-generation }} + evidence-directory: ${{ inputs.evidence-directory }}/agent + + - name: Assert durable accepted-new publication + uses: ./.github/actions/agent/assert-trigger-publication + with: + container-name: ${{ steps.system.outputs.container-name }} + provider-instance-id: ${{ steps.provider.outputs.provider-instance-id }} + expected-triggers: stateforge.test/runtime.started@1 + evidence-name: before-agent-restart + evidence-directory: ${{ inputs.evidence-directory }}/trigger + + - name: Confirm durable commit preceded the lost acknowledgement + shell: bash + env: + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/trigger + EVENT_LOG: ${{ steps.provider.outputs.event-log }} + run: | + set -euo pipefail + publication_id=22222222-2222-4222-8222-222222222222 + expected="trigger-ack-dropped:${publication_id}:accepted_new" + deadline=$((SECONDS + 60)) + while (( SECONDS < deadline )); do + if docker exec "${CONTAINER_NAME}" grep -Fxq "${expected}" "${EVENT_LOG}"; then + break + fi + sleep 1 + done + docker exec "${CONTAINER_NAME}" grep -Fx "${expected}" "${EVENT_LOG}" \ + > "${EVIDENCE_DIRECTORY}/lost-acknowledgement.txt" + jq -e --arg publication_id "${publication_id}" \ + '.publication_count == 1 + and .publications[0].publication_id == $publication_id' \ + "${EVIDENCE_DIRECTORY}/before-agent-restart.json" >/dev/null + + - name: Crash the installed Agent after durable acceptance + shell: bash + env: + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + run: | + set -euo pipefail + docker exec "${CONTAINER_NAME}" \ + systemctl kill --kill-whom=main --signal=SIGKILL \ + stateforge-agent.service + docker exec "${CONTAINER_NAME}" systemctl stop stateforge-agent.service + + - name: Permit the replay acknowledgement + shell: bash + env: + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + STATE_DIRECTORY: ${{ steps.provider.outputs.state-directory }} + run: | + set -euo pipefail + docker exec "${CONTAINER_NAME}" touch \ + "${STATE_DIRECTORY}/accept-trigger-acks" + + - name: Restart the exact installed Agent + id: agent-restart + uses: ./.github/actions/agent/start + with: + container-name: ${{ steps.system.outputs.container-name }} + + - name: Confirm immutable replay received accepted-duplicate + shell: bash + env: + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/trigger + EVENT_LOG: ${{ steps.provider.outputs.event-log }} + run: | + set -euo pipefail + publication_id=22222222-2222-4222-8222-222222222222 + expected="trigger-ack-observed:${publication_id}:accepted_duplicate" + deadline=$((SECONDS + 60)) + while (( SECONDS < deadline )); do + if docker exec "${CONTAINER_NAME}" grep -Fxq "${expected}" "${EVENT_LOG}"; then + break + fi + sleep 1 + done + docker exec "${CONTAINER_NAME}" grep -Fx "${expected}" "${EVENT_LOG}" \ + > "${EVIDENCE_DIRECTORY}/replay-outcome.txt" + + - name: Assert restart retained one immutable publication + uses: ./.github/actions/agent/assert-trigger-publication + with: + container-name: ${{ steps.system.outputs.container-name }} + provider-instance-id: ${{ steps.provider.outputs.provider-instance-id }} + expected-triggers: stateforge.test/runtime.started@1 + evidence-name: after-agent-restart + evidence-directory: ${{ inputs.evidence-directory }}/trigger + + - name: Compare durable evidence across replay + shell: bash + env: + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/trigger + run: | + set -euo pipefail + jq -e --slurp \ + '.[0].publication_count == 1 + and .[1].publication_count == 1 + and .[0].publications[0].publication_id + == .[1].publications[0].publication_id + and .[0].publications[0].content_fingerprint + == .[1].publications[0].content_fingerprint' \ + "${EVIDENCE_DIRECTORY}/before-agent-restart.json" \ + "${EVIDENCE_DIRECTORY}/after-agent-restart.json" >/dev/null + + - name: Publish through the callback API + shell: bash + env: + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + UNIT_NAME: ${{ steps.provider.outputs.unit-name }} + run: | + set -euo pipefail + docker exec "${CONTAINER_NAME}" \ + systemctl kill --kill-whom=main --signal=SIGALRM "${UNIT_NAME}" + + - name: Assert callback accepted-new + uses: ./.github/actions/agent/assert-trigger-publication + with: + container-name: ${{ steps.system.outputs.container-name }} + provider-instance-id: ${{ steps.provider.outputs.provider-instance-id }} + expected-triggers: >- + stateforge.test/runtime.started@1 + stateforge.test/runtime.refreshed@1 + evidence-name: after-callback + evidence-directory: ${{ inputs.evidence-directory }}/trigger + + - name: Deny Agent journal writes and publish again + shell: bash + env: + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/trigger + UNIT_NAME: ${{ steps.provider.outputs.unit-name }} + run: | + set -euo pipefail + journal_directory=/var/lib/stateforge-agent/journal + database="${journal_directory}/agent.db" + docker exec "${CONTAINER_NAME}" stat --format='%A %a %U:%G %n' \ + "${journal_directory}" "${database}" \ + > "${EVIDENCE_DIRECTORY}/journal-permissions-before.txt" + docker exec "${CONTAINER_NAME}" bash -lc \ + "chmod u-w '${journal_directory}' '${database}'*" + docker exec "${CONTAINER_NAME}" \ + systemctl kill --kill-whom=main --signal=SIGALRM "${UNIT_NAME}" + sleep 3 + + - name: Assert retryable storage rejection assumed no responsibility + uses: ./.github/actions/agent/assert-trigger-publication + with: + container-name: ${{ steps.system.outputs.container-name }} + provider-instance-id: ${{ steps.provider.outputs.provider-instance-id }} + expected-triggers: >- + stateforge.test/runtime.started@1 + stateforge.test/runtime.refreshed@1 + evidence-name: during-storage-failure + timeout-seconds: "5" + evidence-directory: ${{ inputs.evidence-directory }}/trigger + + - name: Correlate the retryable storage rejection + id: storage-rejection + shell: bash + env: + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/trigger + EVENT_LOG: ${{ steps.provider.outputs.event-log }} + run: | + set -euo pipefail + pattern='^trigger-ack-observed:[0-9a-f-]+:rejected:delivery_responsibility_unavailable:true$' + deadline=$((SECONDS + 60)) + line="" + while (( SECONDS < deadline )); do + line="$(docker exec "${CONTAINER_NAME}" \ + grep -E "${pattern}" "${EVENT_LOG}" | tail -n 1 || true)" + [[ -n "${line}" ]] && break + sleep 1 + done + if [[ -z "${line}" ]]; then + echo "Provider did not observe the retryable storage rejection" >&2 + exit 1 + fi + publication_id="$(cut -d: -f2 <<< "${line}")" + if [[ ! "${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 "Retryable rejection has an invalid publication identity" >&2 + exit 1 + fi + printf '%s\n' "${line}" \ + > "${EVIDENCE_DIRECTORY}/retryable-storage-rejection.txt" + echo "publication-id=${publication_id}" >> "${GITHUB_OUTPUT}" + + - name: Restore Agent journal writes + if: ${{ always() }} + shell: bash + env: + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/trigger + run: | + set -euo pipefail + journal_directory=/var/lib/stateforge-agent/journal + database="${journal_directory}/agent.db" + if docker inspect "${CONTAINER_NAME}" >/dev/null 2>&1; then + docker exec "${CONTAINER_NAME}" bash -lc \ + "chmod u+w '${journal_directory}' '${database}'*" + docker exec "${CONTAINER_NAME}" stat --format='%A %a %U:%G %n' \ + "${journal_directory}" "${database}" \ + > "${EVIDENCE_DIRECTORY}/journal-permissions-after.txt" + fi + + - name: Assert queued publication succeeds after storage recovery + uses: ./.github/actions/agent/assert-trigger-publication + with: + container-name: ${{ steps.system.outputs.container-name }} + provider-instance-id: ${{ steps.provider.outputs.provider-instance-id }} + expected-triggers: >- + stateforge.test/runtime.started@1 + stateforge.test/runtime.refreshed@1 + stateforge.test/runtime.refreshed@1 + evidence-name: after-storage-recovery + evidence-directory: ${{ inputs.evidence-directory }}/trigger + + - name: Confirm the same rejected publication was accepted + shell: bash + env: + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/trigger + EVENT_LOG: ${{ steps.provider.outputs.event-log }} + PUBLICATION_ID: ${{ steps.storage-rejection.outputs.publication-id }} + run: | + set -euo pipefail + acknowledgement="trigger-ack-observed:${PUBLICATION_ID}:accepted_new" + outcome="trigger-outcome:${PUBLICATION_ID}:accepted_new" + deadline=$((SECONDS + 60)) + while (( SECONDS < deadline )); do + if docker exec "${CONTAINER_NAME}" \ + grep -Fxq "${acknowledgement}" "${EVENT_LOG}" \ + && docker exec "${CONTAINER_NAME}" \ + grep -Fxq "${outcome}" "${EVENT_LOG}"; then + break + fi + sleep 1 + done + docker exec "${CONTAINER_NAME}" \ + grep -Fx "${acknowledgement}" "${EVENT_LOG}" \ + > "${EVIDENCE_DIRECTORY}/storage-recovery-acknowledgement.txt" + docker exec "${CONTAINER_NAME}" grep -Fx "${outcome}" "${EVENT_LOG}" \ + > "${EVIDENCE_DIRECTORY}/storage-recovery-outcome.txt" + jq -e --arg publication_id "${PUBLICATION_ID}" \ + 'any(.publications[]; .publication_id == $publication_id + and .trigger_identifier == "stateforge.test/runtime.refreshed@1")' \ + "${EVIDENCE_DIRECTORY}/after-storage-recovery.json" >/dev/null + + - name: Supersede the live Provider with a newer generation + id: conflict + uses: ./.github/actions/provider/start + with: + container-name: ${{ steps.system.outputs.container-name }} + fixture-path: ${{ steps.fixture.outputs.fixture-path }} + instance-name: trigger-conflict + state-name: ${{ steps.fixture.outputs.state-name }} + identity: stateforge.test.trigger + trigger-publication: "true" + first-trigger-publication-id: 22222222-2222-4222-8222-222222222222 + trigger-startup-sequence: "99" + evidence-directory: ${{ inputs.evidence-directory }}/provider-conflict + + - name: Assert replacement session uses a newer endpoint fence + shell: bash + env: + INITIAL_GENERATION: ${{ steps.provider.outputs.endpoint-generation }} + REPLACEMENT_GENERATION: ${{ steps.conflict.outputs.endpoint-generation }} + run: | + set -euo pipefail + if (( REPLACEMENT_GENERATION <= INITIAL_GENERATION )); then + echo "Replacement Provider did not advance its endpoint fence" >&2 + exit 1 + fi + + - name: Assert Agent-accepted replacement session + uses: ./.github/actions/agent/assert-session + with: + container-name: ${{ steps.system.outputs.container-name }} + provider-instance-id: ${{ steps.conflict.outputs.provider-instance-id }} + endpoint-generation: ${{ steps.conflict.outputs.endpoint-generation }} + evidence-directory: ${{ inputs.evidence-directory }}/agent-conflict + + - name: Preserve invocation identities and endpoint fences + shell: bash + env: + AGENT_INITIAL_INVOCATION: ${{ steps.agent.outputs.service-invocation-id }} + AGENT_RESTART_INVOCATION: ${{ steps.agent-restart.outputs.service-invocation-id }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/trigger + PROVIDER_INITIAL_GENERATION: ${{ steps.provider.outputs.endpoint-generation }} + PROVIDER_INITIAL_INVOCATION: ${{ steps.provider.outputs.service-invocation-id }} + PROVIDER_REPLACEMENT_GENERATION: ${{ steps.conflict.outputs.endpoint-generation }} + PROVIDER_REPLACEMENT_INVOCATION: ${{ steps.conflict.outputs.service-invocation-id }} + run: | + set -euo pipefail + if [[ "${AGENT_INITIAL_INVOCATION}" == "${AGENT_RESTART_INVOCATION}" \ + || "${PROVIDER_INITIAL_INVOCATION}" == "${PROVIDER_REPLACEMENT_INVOCATION}" ]]; then + echo "Restarted processes retained stale systemd invocations" >&2 + exit 1 + fi + printf '%s\n' \ + "agent-initial-invocation=${AGENT_INITIAL_INVOCATION}" \ + "agent-restart-invocation=${AGENT_RESTART_INVOCATION}" \ + "provider-initial-invocation=${PROVIDER_INITIAL_INVOCATION}" \ + "provider-initial-generation=${PROVIDER_INITIAL_GENERATION}" \ + "provider-replacement-invocation=${PROVIDER_REPLACEMENT_INVOCATION}" \ + "provider-replacement-generation=${PROVIDER_REPLACEMENT_GENERATION}" \ + > "${EVIDENCE_DIRECTORY}/invocation-fences.txt" + + - name: Confirm conflicting content receives terminal rejection + shell: bash + env: + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/trigger + EVENT_LOG: ${{ steps.conflict.outputs.event-log }} + run: | + set -euo pipefail + publication_id=22222222-2222-4222-8222-222222222222 + expected="trigger-ack-observed:${publication_id}:rejected:idempotency_conflict:false" + deadline=$((SECONDS + 60)) + while (( SECONDS < deadline )); do + if docker exec "${CONTAINER_NAME}" grep -Fxq "${expected}" "${EVENT_LOG}"; then + break + fi + sleep 1 + done + docker exec "${CONTAINER_NAME}" grep -Fx "${expected}" "${EVENT_LOG}" \ + > "${EVIDENCE_DIRECTORY}/conflicting-content.txt" + + - name: Assert conflict created no extra durable publication + uses: ./.github/actions/agent/assert-trigger-publication + with: + container-name: ${{ steps.system.outputs.container-name }} + provider-instance-id: ${{ steps.conflict.outputs.provider-instance-id }} + expected-triggers: >- + stateforge.test/runtime.started@1 + stateforge.test/runtime.refreshed@1 + stateforge.test/runtime.refreshed@1 + evidence-name: after-conflict + evidence-directory: ${{ inputs.evidence-directory }}/trigger + + - name: Attempt publication from the superseded Provider session + id: stale-publication + shell: bash + env: + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/trigger + EVENT_LOG: ${{ steps.provider.outputs.event-log }} + UNIT_NAME: ${{ steps.provider.outputs.unit-name }} + run: | + set -euo pipefail + before="$(docker exec "${CONTAINER_NAME}" \ + grep -Ec '^trigger-queued:[0-9a-f-]+$' "${EVENT_LOG}" || true)" + docker exec "${CONTAINER_NAME}" \ + systemctl kill --kill-whom=main --signal=SIGALRM "${UNIT_NAME}" + expected=$((before + 1)) + deadline=$((SECONDS + 60)) + after=0 + while (( SECONDS < deadline )); do + after="$(docker exec "${CONTAINER_NAME}" \ + grep -Ec '^trigger-queued:[0-9a-f-]+$' "${EVENT_LOG}" || true)" + [[ "${after}" == "${expected}" ]] && break + sleep 1 + done + if [[ "${after}" != "${expected}" ]]; then + echo "Superseded Provider did not queue the stale publication" >&2 + exit 1 + fi + line="$(docker exec "${CONTAINER_NAME}" \ + grep -E '^trigger-queued:[0-9a-f-]+$' "${EVENT_LOG}" | tail -n 1)" + publication_id="${line#trigger-queued:}" + printf '%s\n' "${line}" \ + > "${EVIDENCE_DIRECTORY}/superseded-publication.txt" + echo "publication-id=${publication_id}" >> "${GITHUB_OUTPUT}" + sleep 3 + if docker exec "${CONTAINER_NAME}" \ + grep -Eq "^trigger-(ack-(observed|dropped)|outcome):${publication_id}:" \ + "${EVENT_LOG}"; then + echo "Superseded Provider publication received an acknowledgement" >&2 + exit 1 + fi + + - name: Assert superseded publication was not accepted + uses: ./.github/actions/agent/assert-trigger-publication + with: + container-name: ${{ steps.system.outputs.container-name }} + provider-instance-id: ${{ steps.provider.outputs.provider-instance-id }} + expected-triggers: >- + stateforge.test/runtime.started@1 + stateforge.test/runtime.refreshed@1 + stateforge.test/runtime.refreshed@1 + evidence-name: after-stale-publication + timeout-seconds: "5" + evidence-directory: ${{ inputs.evidence-directory }}/trigger + + - name: Compare durable state across stale publication + shell: bash + env: + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/trigger + PUBLICATION_ID: ${{ steps.stale-publication.outputs.publication-id }} + run: | + set -euo pipefail + jq -e --arg publication_id "${PUBLICATION_ID}" --slurp \ + '.[0].publication_count == .[1].publication_count + and .[0].publications == .[1].publications + and all(.[1].publications[]; .publication_id != $publication_id)' \ + "${EVIDENCE_DIRECTORY}/after-conflict.json" \ + "${EVIDENCE_DIRECTORY}/after-stale-publication.json" >/dev/null + + - name: Stop superseded Provider without withdrawing current descriptor + shell: bash + env: + CONTAINER_NAME: ${{ steps.system.outputs.container-name }} + CURRENT_DESCRIPTOR: ${{ steps.conflict.outputs.descriptor-path }} + CURRENT_GENERATION: ${{ steps.conflict.outputs.endpoint-generation }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }}/provider + EVENT_LOG: ${{ steps.provider.outputs.event-log }} + SOCKET_PATH: ${{ steps.provider.outputs.socket-path }} + STALE_PUBLICATION_ID: ${{ steps.stale-publication.outputs.publication-id }} + UNIT_NAME: ${{ steps.provider.outputs.unit-name }} + run: | + set -euo pipefail + docker exec "${CONTAINER_NAME}" systemctl stop "${UNIT_NAME}" + deadline=$((SECONDS + 30)) + state="" + while (( SECONDS < deadline )); do + state="$(docker exec "${CONTAINER_NAME}" \ + systemctl is-active "${UNIT_NAME}" 2>/dev/null || true)" + [[ "${state}" == "inactive" || "${state}" == "failed" ]] && break + sleep 1 + done + if [[ "${state}" != "inactive" && "${state}" != "failed" ]]; then + echo "Superseded Provider did not stop" >&2 + exit 1 + fi + docker exec "${CONTAINER_NAME}" test ! -e "${SOCKET_PATH}" + docker exec "${CONTAINER_NAME}" jq -e \ + --argjson generation "${CURRENT_GENERATION}" \ + '.endpoint_generation == $generation' "${CURRENT_DESCRIPTOR}" \ + > "${EVIDENCE_DIRECTORY}/current-descriptor-after-old-stop.json" + docker exec "${CONTAINER_NAME}" grep -Fxq startup "${EVENT_LOG}" + docker exec "${CONTAINER_NAME}" grep -Fxq shutdown "${EVENT_LOG}" + if docker exec "${CONTAINER_NAME}" \ + grep -Eq "^trigger-(ack-(observed|dropped)|outcome):${STALE_PUBLICATION_ID}:" \ + "${EVENT_LOG}"; then + echo "Superseded publication resolved during Provider shutdown" >&2 + exit 1 + fi + docker exec "${CONTAINER_NAME}" cat "${EVENT_LOG}" \ + > "${EVIDENCE_DIRECTORY}/lifecycle-events.txt" + docker exec "${CONTAINER_NAME}" \ + journalctl -u "${UNIT_NAME}" --no-pager \ + > "${EVIDENCE_DIRECTORY}/provider-journal.txt" + + - name: Stop replacement Provider cleanly + uses: ./.github/actions/provider/stop + with: + container-name: ${{ steps.system.outputs.container-name }} + unit-name: ${{ steps.conflict.outputs.unit-name }} + descriptor-path: ${{ steps.conflict.outputs.descriptor-path }} + socket-path: ${{ steps.conflict.outputs.socket-path }} + event-log: ${{ steps.conflict.outputs.event-log }} + evidence-directory: ${{ inputs.evidence-directory }}/provider-conflict + + - name: Preserve bounded evidence and clean up + if: ${{ always() }} + uses: ./.github/actions/evidence/collect + with: + stage: trigger-publication-agent + container-name: ${{ steps.system.outputs.container-name }} + runtime-directory: ${{ inputs.runtime-directory }} + evidence-directory: ${{ inputs.evidence-directory }} diff --git a/.github/actions/scenarios/trigger/installed-agent/fixture.py b/.github/actions/scenarios/trigger/installed-agent/fixture.py new file mode 100644 index 0000000..0e5f0e0 --- /dev/null +++ b/.github/actions/scenarios/trigger/installed-agent/fixture.py @@ -0,0 +1,237 @@ +"""Run the public Trigger API against the installed StateForge Agent.""" + +from __future__ import annotations + +import argparse +import asyncio +import contextlib +import signal +from collections.abc import AsyncIterator, Callable +from pathlib import Path +from uuid import UUID + +from stateforge_protocol.sfp.trigger_publication.v1 import ( + trigger_publication_pb2, +) + +from stateforge_provider import ( + Platform, + Provider, + ProviderModel, + Trigger, + trigger, +) +from stateforge_provider._services.trigger_publication import ( + runtime as trigger_runtime, +) +from stateforge_provider._services.trigger_publication.runtime import ( + TriggerPublicationRuntime, + TriggerStreamLease, +) + + +class RuntimeEvent(ProviderModel): + """Describe deterministic fixture events.""" + + sequence: int + source: str + + +RUNTIME_REFRESHED = Trigger( + "stateforge.test/runtime.refreshed@1", + RuntimeEvent, +) + + +def _record(path: Path, event: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as stream: + stream.write(f"{event}\n") + + +class TriggerFixtureProvider(Provider): + """Publish through both supported public author-facing APIs.""" + + identity = "stateforge.test.trigger" + display_name = "StateForge Trigger CI Provider" + version = "1.0.0" + platforms = (Platform("linux", "x86_64"),) + triggers = (RUNTIME_REFRESHED,) + + def __init__( + self, + *, + discovery_directory: Path, + state_directory: Path, + event_log: Path, + startup_sequence: int, + ) -> None: + """Bind deterministic trigger inputs to one Provider installation.""" + super().__init__( + discovery_directory=discovery_directory, + state_directory=state_directory, + ) + self._event_log = event_log + self._startup_sequence = startup_sequence + self._startup_events: asyncio.Queue[RuntimeEvent] = asyncio.Queue() + self._tasks: set[asyncio.Task[None]] = set() + + async def on_startup(self) -> None: + """Publish the decorated event and install the callback signal.""" + loop = asyncio.get_running_loop() + loop.add_signal_handler(signal.SIGALRM, self._schedule_callback) + self._startup_events.put_nowait( + RuntimeEvent( + sequence=self._startup_sequence, + source="decorated", + ) + ) + _record(self._event_log, "startup") + + async def on_shutdown(self) -> None: + """Remove signal ownership and join callback tasks.""" + with contextlib.suppress(NotImplementedError, ValueError): + asyncio.get_running_loop().remove_signal_handler(signal.SIGALRM) + if self._tasks: + await asyncio.gather(*self._tasks, return_exceptions=True) + _record(self._event_log, "shutdown") + + @trigger("stateforge.test/runtime.started@1") + async def runtime_started(self) -> AsyncIterator[RuntimeEvent]: + """Publish one decorated startup event.""" + yield await self._startup_events.get() + + def _schedule_callback(self) -> None: + _record( + self._event_log, + f"trigger-scheduled:{RUNTIME_REFRESHED.identifier}", + ) + task = asyncio.create_task(self._publish_callback()) + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) + + async def _publish_callback(self) -> None: + receipt = await self.publish_trigger( + RUNTIME_REFRESHED, + RuntimeEvent(sequence=2, source="callback"), + ) + _record(self._event_log, f"trigger-queued:{receipt.publication_id}") + outcome = await receipt.outcome() + suffix = "" + if outcome.rejection is not None: + suffix = ( + f":{outcome.rejection.reason.value}" + f":{str(outcome.rejection.retryable).lower()}" + ) + _record( + self._event_log, + "trigger-outcome:" + f"{outcome.publication_id}:{outcome.disposition.value}{suffix}", + ) + + +def _install_ack_observer(arguments: argparse.Namespace) -> None: + original = TriggerPublicationRuntime.acknowledge + marker = arguments.state_directory / "accept-trigger-acks" + + async def acknowledge( + runtime: TriggerPublicationRuntime, + lease: TriggerStreamLease, + acknowledgement: trigger_publication_pb2.TriggerAcknowledgement, + *, + is_session_current: Callable[[str], bool], + ) -> None: + disposition = ( + trigger_publication_pb2.TriggerAcknowledgementDisposition.Name( + acknowledgement.disposition + ) + .removeprefix("TRIGGER_ACKNOWLEDGEMENT_DISPOSITION_") + .lower() + ) + suffix = "" + if acknowledgement.HasField("rejection"): + reason = ( + trigger_publication_pb2.TriggerRejectionReason.Name( + acknowledgement.rejection.reason + ) + .removeprefix("TRIGGER_REJECTION_REASON_") + .lower() + ) + suffix = ( + f":{reason}:{str(acknowledgement.rejection.retryable).lower()}" + ) + if arguments.drop_trigger_acks_until_marker and not marker.exists(): + _record( + arguments.event_log, + "trigger-ack-dropped:" + f"{acknowledgement.publication_id}:{disposition}{suffix}", + ) + return + await original( + runtime, + lease, + acknowledgement, + is_session_current=is_session_current, + ) + _record( + arguments.event_log, + "trigger-ack-observed:" + f"{acknowledgement.publication_id}:{disposition}{suffix}", + ) + + TriggerPublicationRuntime.acknowledge = acknowledge + + +def _install_first_publication_id(configured: str) -> None: + if not configured: + return + first = UUID(configured) + original = trigger_runtime.uuid4 + pending = True + + def uuid4() -> UUID: + nonlocal pending + if pending: + pending = False + return first + return original() + + trigger_runtime.uuid4 = uuid4 + + +def _arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--identity", required=True) + parser.add_argument("--display-name", required=True) + parser.add_argument("--discovery-directory", type=Path, required=True) + parser.add_argument("--state-directory", type=Path, required=True) + parser.add_argument("--event-log", type=Path, required=True) + parser.add_argument("--target-path", type=Path, required=True) + parser.add_argument("--trigger-publication", action="store_true") + parser.add_argument( + "--drop-trigger-acks-until-marker", action="store_true" + ) + parser.add_argument("--first-trigger-publication-id", default="") + parser.add_argument("--trigger-startup-sequence", type=int, default=1) + return parser.parse_args() + + +def main() -> None: + """Install deterministic faults and run the standalone Provider.""" + arguments = _arguments() + if not arguments.trigger_publication: + raise ValueError("Trigger fixture requires --trigger-publication") + TriggerFixtureProvider.identity = arguments.identity + TriggerFixtureProvider.display_name = arguments.display_name + _install_ack_observer(arguments) + _install_first_publication_id(arguments.first_trigger_publication_id) + TriggerFixtureProvider( + discovery_directory=arguments.discovery_directory.resolve(), + state_directory=arguments.state_directory.resolve(), + event_log=arguments.event_log.resolve(), + startup_sequence=arguments.trigger_startup_sequence, + ).run() + + +if __name__ == "__main__": + main() diff --git a/.github/actions/scenarios/trigger/publication/verify.py b/.github/actions/scenarios/trigger/publication/verify.py index 19ad5c0..c7bdbdc 100644 --- a/.github/actions/scenarios/trigger/publication/verify.py +++ b/.github/actions/scenarios/trigger/publication/verify.py @@ -191,6 +191,79 @@ async def _ack( ) +async def _assert_stale_fence( + stub: trigger_publication_pb2_grpc.TriggerPublicationServiceStub, + handshake: bootstrap_pb2.HandshakeResponse, + snapshot: manifest_pb2.GetManifestResponse, +) -> str: + """Prove an endpoint-generation mismatch cannot open a stream.""" + call = stub.StreamTriggers(timeout=3) + try: + await call.write( + trigger_publication_pb2.StreamTriggersRequest( + open=trigger_publication_pb2.TriggerStreamOpen( + session_binding=( + trigger_publication_pb2.TriggerSessionBinding( + expected_provider_instance_id=( + handshake.provider_instance_id + ), + expected_endpoint_generation=( + handshake.endpoint_generation + 1 + ), + expected_session_id=handshake.session_id, + expected_manifest_revision=( + snapshot.manifest_revision + ), + ) + ), + requested_max_in_flight_publications=1, + ) + ) + ) + await call.read() + except grpc.aio.AioRpcError as error: + _require( + error.code() is grpc.StatusCode.FAILED_PRECONDITION, + "stale endpoint fence status", + ) + return error.code().name + finally: + call.cancel() + raise RuntimeError("Trigger Publication accepted a stale endpoint fence") + + +async def _open_verified_stream( + stub: trigger_publication_pb2_grpc.TriggerPublicationServiceStub, + handshake: bootstrap_pb2.HandshakeResponse, + snapshot: manifest_pb2.GetManifestResponse, +) -> tuple[ + grpc.aio.StreamStreamCall[ + trigger_publication_pb2.StreamTriggersRequest, + trigger_publication_pb2.StreamTriggersResponse, + ], + str, +]: + """Open the current stream after proving stale fences are rejected.""" + stale_fence_status = await _assert_stale_fence( + stub, + handshake, + snapshot, + ) + return ( + await _stream(stub, handshake, snapshot), + stale_fence_status, + ) + + +async def _prepare_runtime(runtime_directory: Path) -> None: + """Create only the scenario-owned runtime directories.""" + await asyncio.to_thread(runtime_directory.mkdir, parents=True) + await asyncio.to_thread( + (runtime_directory / "discovery").mkdir, + parents=True, + ) + + async def _prove_source_acknowledgement( provider: TriggerProvider, call: grpc.aio.StreamStreamCall[ @@ -229,13 +302,178 @@ async def _prove_source_acknowledgement( return sentinel.publication_id, outcome.disposition +async def _prove_terminal_rejection( + provider: TriggerProvider, + call: grpc.aio.StreamStreamCall[ + trigger_publication_pb2.StreamTriggersRequest, + trigger_publication_pb2.StreamTriggersResponse, + ], +) -> TriggerPublicationDisposition: + """Prove nonretryable rejection resolves the public receipt once.""" + receipt = await provider.publish_trigger( + DEPLOYMENT_COMPLETED, + DeploymentCompleted(environment="rejected", revision="terminal"), + ) + publication = (await call.read()).publication + await call.write( + trigger_publication_pb2.StreamTriggersRequest( + acknowledgement=trigger_publication_pb2.TriggerAcknowledgement( + publication_id=publication.publication_id, + disposition=( + trigger_publication_pb2.TRIGGER_ACKNOWLEDGEMENT_DISPOSITION_REJECTED + ), + rejection=trigger_publication_pb2.TriggerRejection( + reason=( + trigger_publication_pb2.TRIGGER_REJECTION_REASON_TRIGGER_NOT_DECLARED + ), + retryable=False, + explanation="Injected terminal rejection", + ), + ) + ) + ) + outcome = await receipt.outcome() + _require( + outcome.disposition is TriggerPublicationDisposition.REJECTED, + "terminal rejection outcome", + ) + _require( + outcome.rejection is not None and not outcome.rejection.retryable, + "terminal rejection details", + ) + return outcome.disposition + + +async def _prove_receive_window_pressure( + provider: TriggerProvider, + call: grpc.aio.StreamStreamCall[ + trigger_publication_pb2.StreamTriggersRequest, + trigger_publication_pb2.StreamTriggersResponse, + ], +) -> dict[str, object]: + """Prove a selected window of one blocks a second publication.""" + first_receipt = await provider.publish_trigger( + DEPLOYMENT_COMPLETED, + DeploymentCompleted(environment="pressure", revision="first"), + ) + second_receipt = await provider.publish_trigger( + DEPLOYMENT_COMPLETED, + DeploymentCompleted(environment="pressure", revision="second"), + ) + first = (await call.read()).publication + pending_second = asyncio.create_task(call.read()) + try: + await asyncio.wait_for(asyncio.shield(pending_second), timeout=0.05) + except TimeoutError: + second_blocked = True + else: + second_blocked = False + _require(second_blocked, "receive-window pressure") + + await _ack( + call, + first.publication_id, + trigger_publication_pb2.TRIGGER_ACKNOWLEDGEMENT_DISPOSITION_ACCEPTED_NEW, + ) + second = (await asyncio.wait_for(pending_second, timeout=2)).publication + _require( + first.publication_id != second.publication_id, + "pressure publication identity", + ) + _require( + first.payload != second.payload, + "pressure publication payload", + ) + await _ack( + call, + second.publication_id, + trigger_publication_pb2.TRIGGER_ACKNOWLEDGEMENT_DISPOSITION_ACCEPTED_NEW, + ) + first_outcome, second_outcome = await asyncio.gather( + first_receipt.outcome(), + second_receipt.outcome(), + ) + _require( + first_outcome.disposition + is TriggerPublicationDisposition.ACCEPTED_NEW, + "first pressure acknowledgement", + ) + _require( + second_outcome.disposition + is TriggerPublicationDisposition.ACCEPTED_NEW, + "second pressure acknowledgement", + ) + return { + "first_content_fingerprint": ( + compute_publication_content_fingerprint(first) + ), + "first_publication_id": first.publication_id, + "second_blocked_until_acknowledgement": second_blocked, + "second_content_fingerprint": ( + compute_publication_content_fingerprint(second) + ), + "second_publication_id": second.publication_id, + } + + +async def _prove_retryable_rejection( + provider: TriggerProvider, + call: grpc.aio.StreamStreamCall[ + trigger_publication_pb2.StreamTriggersRequest, + trigger_publication_pb2.StreamTriggersResponse, + ], +) -> dict[str, object]: + """Prove a retryable rejection preserves the complete publication.""" + receipt = await provider.publish_trigger( + DEPLOYMENT_COMPLETED, + DeploymentCompleted(environment="production", revision="abc123"), + observed_at=datetime(2026, 10, 28, 12, 0, tzinfo=UTC), + ) + first = (await call.read()).publication + retry_after = Duration() + retry_after.FromMilliseconds(1) + await call.write( + trigger_publication_pb2.StreamTriggersRequest( + acknowledgement=trigger_publication_pb2.TriggerAcknowledgement( + publication_id=first.publication_id, + disposition=( + trigger_publication_pb2.TRIGGER_ACKNOWLEDGEMENT_DISPOSITION_REJECTED + ), + rejection=trigger_publication_pb2.TriggerRejection( + reason=( + trigger_publication_pb2.TRIGGER_REJECTION_REASON_AGENT_CAPACITY_EXHAUSTED + ), + retryable=True, + retry_after=retry_after, + explanation="Temporary capacity exhaustion", + ), + ) + ) + ) + retried = (await call.read()).publication + _require(retried == first, "immutable retry publication") + await _ack( + call, + retried.publication_id, + trigger_publication_pb2.TRIGGER_ACKNOWLEDGEMENT_DISPOSITION_ACCEPTED_DUPLICATE, + ) + outcome = await receipt.outcome() + _require( + outcome.disposition + is TriggerPublicationDisposition.ACCEPTED_DUPLICATE, + "callback acknowledgement outcome", + ) + return { + "content_fingerprint": compute_publication_content_fingerprint(first), + "delivery_attempts": 2, + "disposition": outcome.disposition.value, + "publication_id": first.publication_id, + } + + async def _verify(runtime_directory: Path) -> dict[str, object]: """Exercise retry and reconnect replay over the installed endpoint.""" - await asyncio.to_thread(runtime_directory.mkdir, parents=True) - await asyncio.to_thread( - (runtime_directory / "discovery").mkdir, - parents=True, - ) + await _prepare_runtime(runtime_directory) provider = TriggerProvider(runtime_directory) provider_task = asyncio.create_task(provider.serve()) channel: grpc.aio.Channel | None = None @@ -266,50 +504,17 @@ async def _verify(runtime_directory: Path) -> dict[str, object]: }, "manifest trigger declarations", ) - call = await _stream(trigger_stub, handshake, snapshot) - - receipt = await provider.publish_trigger( - DEPLOYMENT_COMPLETED, - DeploymentCompleted(environment="production", revision="abc123"), - observed_at=datetime(2026, 10, 28, 12, 0, tzinfo=UTC), - ) - first = (await call.read()).publication - retry_after = Duration() - retry_after.FromMilliseconds(1) - await call.write( - trigger_publication_pb2.StreamTriggersRequest( - acknowledgement=( - trigger_publication_pb2.TriggerAcknowledgement( - publication_id=first.publication_id, - disposition=( - trigger_publication_pb2.TRIGGER_ACKNOWLEDGEMENT_DISPOSITION_REJECTED - ), - rejection=trigger_publication_pb2.TriggerRejection( - reason=( - trigger_publication_pb2.TRIGGER_REJECTION_REASON_AGENT_CAPACITY_EXHAUSTED - ), - retryable=True, - retry_after=retry_after, - explanation="Temporary capacity exhaustion", - ), - ) - ) - ) - ) - retried = (await call.read()).publication - _require(retried == first, "immutable retry publication") - await _ack( - call, - retried.publication_id, - trigger_publication_pb2.TRIGGER_ACKNOWLEDGEMENT_DISPOSITION_ACCEPTED_DUPLICATE, - ) - callback_outcome = await receipt.outcome() - _require( - callback_outcome.disposition - is TriggerPublicationDisposition.ACCEPTED_DUPLICATE, - "callback acknowledgement outcome", + call, stale_fence_status = await _open_verified_stream( + trigger_stub, + handshake, + snapshot, ) + pressure = await _prove_receive_window_pressure(provider, call) + retryable_rejection = await _prove_retryable_rejection(provider, call) + + terminal_rejection = await _prove_terminal_rejection(provider, call) + await provider.failures.put( UnitFailed(unit="nginx.service", result="exit-code") ) @@ -355,12 +560,16 @@ async def _verify(runtime_directory: Path) -> dict[str, object]: "replacement_session_id": replacement_handshake.session_id, "manifest_revision": snapshot.manifest_revision, "declared_triggers": sorted(snapshot.manifest.triggers), - "callback_publication_id": first.publication_id, - "callback_fingerprint": ( - compute_publication_content_fingerprint(first) - ), - "callback_outcome": callback_outcome.disposition.value, - "callback_delivery_attempts": 2, + "callback_publication_id": retryable_rejection["publication_id"], + "callback_fingerprint": retryable_rejection["content_fingerprint"], + "callback_outcome": retryable_rejection["disposition"], + "callback_delivery_attempts": retryable_rejection[ + "delivery_attempts" + ], + "retryable_rejection": retryable_rejection, + "receive_window_pressure": pressure, + "terminal_rejection": terminal_rejection.value, + "stale_fence_status": stale_fence_status, "source_publication_id": source_publication.publication_id, "source_replayed_after_reconnect": replayed == source_publication, "source_acknowledgement_observed": ( diff --git a/.github/workflows/provider-runtime-scenarios.yml b/.github/workflows/provider-runtime-scenarios.yml index 08e9812..55cac3e 100644 --- a/.github/workflows/provider-runtime-scenarios.yml +++ b/.github/workflows/provider-runtime-scenarios.yml @@ -131,6 +131,42 @@ jobs: if-no-files-found: warn retention-days: ${{ inputs.retention-days }} + trigger-publication-agent: + name: Trigger Publication / installed Agent RPM + if: ${{ inputs.scenario-tier == 'full' }} + runs-on: ubuntu-24.04 + timeout-minutes: 40 + steps: + - name: Check out scenario automation + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Download exact runtime bundle + uses: actions/download-artifact@9000827ccba6bdab643e8b6fd33ac0654aef8333 # v8.0.2 + with: + name: ${{ inputs.artifact-name }} + path: ${{ runner.temp }}/stateforge-provider-bundle + + - name: Run installed-Agent Trigger Publication scenario + uses: ./.github/actions/scenarios/trigger/installed-agent + with: + agent-artifact-directory: ${{ runner.temp }}/stateforge-provider-bundle/agent + agent-test-image-directory: ${{ runner.temp }}/stateforge-provider-bundle/agent/test-image + provider-artifact-directory: ${{ runner.temp }}/stateforge-provider-bundle/provider + protocol-artifact-directory: ${{ runner.temp }}/stateforge-provider-bundle/protocol + runtime-directory: ${{ runner.temp }}/sfp-trigger-agent + evidence-directory: evidence/runtime/trigger-publication-agent + + - name: Upload runtime evidence + if: ${{ always() }} + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: ${{ inputs.evidence-prefix }}-trigger-publication-agent + path: evidence/runtime/trigger-publication-agent + if-no-files-found: warn + retention-days: ${{ inputs.retention-days }} + capability-assessment: name: Capability Assessment / installed Agent RPM if: ${{ inputs.scenario-tier == 'full' }} diff --git a/docs/ci-automation.md b/docs/ci-automation.md index c29f64a..1f97dab 100644 --- a/docs/ci-automation.md +++ b/docs/ci-automation.md @@ -145,15 +145,27 @@ The runtime suite currently covers: - two distinct Provider installations running concurrently; - same-installation replacement with a stable identity and newer generation; - rejection of malformed Provider discovery data followed by valid recovery; -- abrupt process failure followed by a valid newer generation; and -- uninstall and reinstall of the exact Provider wheel; and +- abrupt process failure followed by a valid newer generation; +- uninstall and reinstall of the exact Provider wheel; - installed-wheel Trigger Publication through an independent Protocol peer, including callback and decorated sources, a bounded receive window, retryable rejection with immutable replay, accepted duplicate disposition, - stream loss, and replay under a fresh fenced session. - -The fixtures import only the public `stateforge_provider` API. Each runtime -scenario installs the exact Agent RPM into an isolated Fedora system, starts + terminal rejection, stale-fence rejection, stream loss, and replay under a + fresh fenced session; and +- end-to-end Trigger Publication through the exact installed Agent RPM and + exact Provider and Protocol wheels, including durable `accepted_new`, a lost + acknowledgement, Agent crash and restart, immutable replay with + `accepted_duplicate`, callback publication, retry across journal write +failure with an explicitly correlated retryable rejection, +conflicting-content rejection, a newer Provider endpoint generation, +rejection of publication from the superseded session, and clean shutdown. + +Runtime fixtures exercise Provider behavior through the public +`stateforge_provider` API. The R18 fixture uses narrowly scoped private hooks +only to discard one acknowledgement and assign one deterministic publication +ID for fault injection; its publications still originate from the public +decorator and callback APIs. Each runtime scenario installs the exact Agent +RPM into an isolated Fedora system, starts the packaged `stateforge-agent.service`, installs the exact Provider and Protocol wheels beside it, and observes acceptance or rejection through the real Agent journal. Runtime scenarios run on Linux because the current endpoint @@ -190,14 +202,29 @@ request, rejects an oversized request before author code, and rejects a valid assessment whose repeated findings would exceed the response limit. A normal assessment must still succeed after both limit failures. -R4 Trigger Publication verification also uses a real Unix socket, but its peer -is deliberately the canonical Protocol contract rather than an invented Agent -runtime. The current Agent does not implement the P3 consumer yet. A4 owns that -implementation. After A4 lands, the real Agent exchange can be added to the -Provider runtime matrix without replacing this direct Provider contract test. -Until then, the direct peer proves the installed Provider wheel's exact P3 -framing, flow control, retry, and replay semantics without pretending that -Agent support already exists. +R18 Trigger Publication verification keeps two intentionally different +boundaries. The focused Protocol peer injects exact wire decisions and checks +Provider-owned retry, flow control, terminal rejection, stale fencing, and +shutdown without replacing any Agent behavior. The installed-runtime scenario +runs the packaged Agent and Provider as separate systemd-managed processes over +the real generation-specific Unix socket. It validates the Agent's durable +journal with canonical Protocol parsing and content fingerprints. + +The installed scenario deliberately loses the first terminal acknowledgement +after the Agent has committed the publication, crashes and restarts the Agent, +and proves the Provider replays the same publication identity and content. The +Agent returns `accepted_duplicate`, after which callback publication continues. +The same scenario makes the Agent journal temporarily unwritable and proves the +Provider retains and retries the publication until responsibility can be +accepted. Finally, the Provider is restarted with the same installation +identity and publication ID but different content; the Agent returns the +terminal idempotency-conflict decision without creating another durable row. +The focused peer also fixes the receive window at one, queues two publications, +and proves the second remains blocked until the first terminal acknowledgement +releases capacity. The installed Agent scenario keeps the first Provider alive +while accepting a newer endpoint generation, then proves a publication queued +by the superseded Provider is never acknowledged or added to durable Agent +state. ## Evidence and security diff --git a/tests/test_ci_automation.py b/tests/test_ci_automation.py index 39176af..2baf7ae 100644 --- a/tests/test_ci_automation.py +++ b/tests/test_ci_automation.py @@ -8,7 +8,7 @@ SERVICES = ROOT / "src" / "stateforge_provider" / "_services" SHA_LENGTH = 40 SUPPORTED_PYTHON_COUNT = 3 -SCENARIO_COUNT = 9 +SCENARIO_COUNT = 10 DOWNSTREAM_BUNDLE_CONSUMER_COUNT = 3 DEPENDENCY_CHECKOUT_COUNT = 2 @@ -23,6 +23,7 @@ def test_action_library_has_focused_domain_boundaries() -> None: expected = { "agent/assert-invalid-descriptor", "agent/assert-capability-assessment", + "agent/assert-trigger-publication", "agent/assert-session", "agent/install", "agent/start", @@ -44,6 +45,7 @@ def test_action_library_has_focused_domain_boundaries() -> None: "scenarios/resilience/crash-restart", "scenarios/smoke/normal-lifecycle", "scenarios/trigger/publication", + "scenarios/trigger/installed-agent", } observed = { str(path.parent.relative_to(ACTIONS)) @@ -207,6 +209,8 @@ def test_provider_start_uses_installed_runtime_in_agent_system() -> None: assert "import stateforge_provider" not in action assert "previous_generation=0" in action assert "generation > previous_generation" in action + assert "service-invocation-id" in action + assert "process-identity.txt" in action assert ( 'log_directory="/var/log/stateforge-provider-tests/${INSTANCE_NAME}"' in action @@ -224,6 +228,16 @@ def test_each_runtime_scenario_is_an_independent_action() -> None: assert "./.github/actions/environment/install" in content assert "./.github/actions/agent/" not in content assert '"${ISOLATED_PYTHON}" -I' in content + elif "scenarios/trigger/installed-agent" in str(scenario): + assert "./.github/actions/agent/install" in content + assert "./.github/actions/agent/start" in content + assert ( + "./.github/actions/agent/assert-trigger-publication" in content + ) + assert "./.github/actions/provider/start" in content + assert "./.github/actions/provider/stop" in content + assert "./.github/actions/evidence/collect" in content + assert "if: ${{ always() }}" in content else: assert "./.github/actions/agent/install" in content assert "./.github/actions/agent/start" in content @@ -244,6 +258,7 @@ def test_runtime_catalog_covers_required_behavior() -> None: "uninstall-reinstall", "standard-health", "trigger-publication", + "trigger-publication-agent", "capability-assessment", } for scenario in scenarios: @@ -258,8 +273,8 @@ def test_runtime_catalog_covers_required_behavior() -> None: assert "/agent/test-image" in workflow -def test_trigger_publication_uses_installed_wheels_and_direct_peer() -> None: - """Verify R4 without inventing an Agent-side trigger implementation.""" +def test_trigger_publication_keeps_the_focused_protocol_peer() -> None: + """Keep retry, pressure, rejection, and fencing faults deterministic.""" scenario = read( ACTIONS / "scenarios" / "trigger" / "publication" / "action.yml" ) @@ -285,6 +300,9 @@ def test_trigger_publication_uses_installed_wheels_and_direct_peer() -> None: "initial_session_id", "replacement_session_id", "callback_delivery_attempts", + "receive_window_pressure", + "terminal_rejection", + "stale_fence_status", "source_acknowledgement_observed", "sentinel_publication_id", "sentinel_outcome", @@ -296,6 +314,51 @@ def test_trigger_publication_uses_installed_wheels_and_direct_peer() -> None: assert "Trigger Publication / direct Protocol peer" in workflow +def test_trigger_publication_uses_exact_installed_agent_runtime() -> None: + """Verify R18 across the real RPM, wheels, socket, and journal.""" + scenario = read( + ACTIONS / "scenarios" / "trigger" / "installed-agent" / "action.yml" + ) + fixture = read( + ACTIONS / "scenarios" / "trigger" / "installed-agent" / "fixture.py" + ) + assertion = read( + ACTIONS / "agent" / "assert-trigger-publication" / "assert.py" + ) + workflow = read(WORKFLOWS / "provider-runtime-scenarios.yml") + + for action in ( + "agent/install", + "agent/start", + "agent/assert-session", + "agent/assert-trigger-publication", + "provider/start", + "provider/stop", + "evidence/collect", + ): + assert f"./.github/actions/{action}" in scenario + assert "@trigger(" in fixture + assert "publish_trigger(" in fixture + assert "TriggerPublicationRuntime.acknowledge" in fixture + assert "trigger_runtime.uuid4" in fixture + assert "accepted_duplicate" in scenario + assert "delivery_responsibility_unavailable" in scenario + assert "idempotency_conflict" in scenario + assert "journal-permissions-before" in scenario + assert "after-storage-recovery" in scenario + assert "after-stale-publication" in scenario + assert "superseded-publication.txt" in scenario + assert "trigger-queued" in fixture + assert "endpoint-generation" in scenario + assert "invocation-fences.txt" in scenario + assert "service-invocation-id" in scenario + assert "trigger_publications" in assertion + assert "validate_trigger_publication" in assertion + assert "compute_publication_content_fingerprint" in assertion + assert "Trigger Publication / installed Agent RPM" in workflow + assert "${{ runner.temp }}/sfp-trigger-agent" in workflow + + def test_health_scenario_uses_standard_wire_service_and_public_hook() -> None: """Keep R2 coverage on the real endpoint and installed runtime bundle.""" scenario = read( From f144310747a41fde7ee954414a9bcd5990fecd23 Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Sun, 1 Nov 2026 12:00:24 +0200 Subject: [PATCH 2/2] test: declare asyncio loop scopes --- pyproject.toml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index cb8ea1d..73a3c06 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -164,6 +164,8 @@ addopts = [ ] testpaths = ["tests"] asyncio_mode = "strict" +asyncio_default_fixture_loop_scope = "function" +asyncio_default_test_loop_scope = "function" [tool.coverage.run]