feat(delivery): openframe-machine-delivery — engine, dispatcher and the first two specs (PR 1 of the plan)

[semen-flamingo]

Draft. This PR adds a module and two specs and wires nothing into existing flows: no service changes behaviour after merge, with or without the flag. The wiring (ack routing, success hooks, call sites) is the stacked follow-up PR on top of this branch.

Context

Four server→agent flows (tool install, tool update, client update, client uninstall) go through JetStream with a durable consumer per machine; the server keeps no state and has no ack, watchdog or metric for them. Scheduled scripts already use an application-level model (ScriptDeliveryRetryService, ScriptExecutionAcknowledgeListener, watchdog, openframe-rmm.yaml alerts). This is the shared engine that the four flows (and later the scripts) move onto; each delivery type is a spec, the same pattern as NotificationTypeSpec + NotificationTypeRegistry + NotificationEmitter.

Module layout

openframe-machine-delivery   com.openframe.delivery — the engine, no NATS dependency
  └─ openframe-data-mongo-sync        (MachineDelivery + repository, MachineRepository)
openframe-data-nats          NATS specs: ToolInstallationDeliverySpec, ClientUninstallDeliverySpec
  └─ openframe-machine-delivery
api-lib, api-service-core, management-service-core, client-core, tool-agent-nats-installation → both (PR 2+)

Publishers of deliveries live in five modules of three services (api, management, client); only client will run the sweep. Hence a module every publisher can include, with the engine's only contact with the machine document isolated in MachineOnlineStatus. "Machine" in the name is deliberate: the engine is transport-agnostic (specs own publish) but assumes a recipient with online/offline presence — it is not a generic outbox for push or Slack.

The pattern, mirrored from notifications

Notifications	Delivery
NotificationType	DeliveryType
NotificationSeed (per type, nested Spec.Seed)	DeliverySeed (per type, nested Spec.Seed)
NotificationTypeSpec<S>	DeliverySpec<S, P> — getType, getSeedClass, getPayloadClass, request(seed), publish(machineId, payload), onFailed(row, failure)
NotificationTypeRegistry	DeliverySpecRegistry
NotificationEmitter.notify(request)	DeliveryDispatcher.dispatch(seed)

A caller hands a seed to the dispatcher and knows nothing else:

deliveryDispatcher.dispatch(new ClientUninstallDeliverySpec.Seed(machineId));
deliveryDispatcher.dispatch(new ToolInstallationDeliverySpec.Seed(machineId, toolAgent, tool, reinstall));

The dispatcher resolves the spec by seed.type(), the spec builds the DeliveryRequest (payload, targetId, machineId), DeliveryRecorder writes the PENDING row, the spec publishes. The targetId a spec publishes with is the one its completion hook will receive from the agent (toolAgent.getKey() ↔ agentType in installed-agent; machine id ↔ X-Machine-Id on /api/agents/uninstall), so that pairing lives in one class per type.

How SUCCESS is detected and when re-delivery stops (wired in PR 2)

A row in machine_delivery is PENDING → ACKED → DONE | FAILED. The sweep only ever selects PENDING, so re-delivery stops the moment the row leaves PENDING. No new subject carries success — the signals the agent already sends will close the row:

Type	Agent signal (exists today)	Server hook (PR 2)	Transition
any	machine.{id}.execution.acknowledge + type, targetId	ScriptExecutionAcknowledgeListener → DeliveryTracker.acknowledge	PENDING → ACKED (re-sends stop)
TOOL_INSTALLATION	machine.{id}.installed-agent {agentType, version} after install()	InstalledAgentService → complete(TOOL_INSTALLATION, agentType, machineId)	→ DONE
CLIENT_UNINSTALL	HTTP POST /api/agents/uninstall	AgentUninstallService → complete(CLIENT_UNINSTALL, machineId, machineId)	→ DONE
SCRIPT_SCHEDULE (PR 7)	script-execution.result	RmmResultService → complete(SCRIPT_SCHEDULE, executionId, machineId)	→ DONE
*_UPDATE (PR 8)	same installed-agent, matched on version		→ DONE only when version == target

What the agent does not send today is a failure: after ack, a failed install is visible only as FAILED/TIMEOUT from the watchdog; an explicit failure result is a Rust-side follow-up (PR 3).

Engine

DeliveryDispatcher.dispatch(seed)         spec = registry.require(seed.type()); request = spec.request(seed)
                                          recorder.record(request)  → row PENDING (MongoDeliveryRecorder / NoopDeliveryRecorder by flag)
                                          spec.publish(machineId, payload)
DeliveryTracker.acknowledge / complete    PENDING → ACKED → DONE   (MongoDeliveryTracker / NoopDeliveryTracker by flag)
DeliverySweepScheduler.tick()  every openframe.delivery.sweep.interval under ShedLock, only where sweep.enabled=true:
   DeliverySweepService.retryPending()    PENDING older than ack-threshold →
       MachineOnlineStatus says OFFLINE → wait (RETRY_ON_RECONNECT) until reconnect-window, or FAILED now (SKIP)
       attempts < max                    → spec.publish(machineId, payload from payloadJson), attempts+1
       else                              → FAILED exhausted
   DeliveryWatchdogService.reapAcked()    ACKED older than result-timeout → FAILED timeout
DeliveryFailureRecorder.fail(row, reason) FAILED + expiresAt (TTL) + metric + spec.onFailed(row, reason)

Override chain

Numbers never live in a spec. Resolution order, all in DeliveryProperties: openframe.delivery.defaults.* (required, validated at startup) → openframe.delivery.types.<TYPE>.* → row-level offlineBehavior / reconnectWindowSeconds (carried from ScheduleScript). A spec overrides behaviour only: request, publish, onFailed — compare ToolInstallationDeliverySpec.onFailed (no-op) with ClientUninstallDeliverySpec.onFailed (restores a machine parked in PENDING_DELETION).

openframe:
  delivery:
    enabled: true
    sweep:
      enabled: true          # client only
      interval: 30000
      lock-at-most-for: 2m
      lock-at-least-for: 10s
    defaults:
      ack-threshold-seconds: 30
      max-attempts: 3
      offline-behavior: RETRY_ON_RECONNECT
      reconnect-window-seconds: 86400
      result-timeout-seconds: 600
      ttl-seconds: 604800
    types:
      SCRIPT_SCHEDULE:
        offline-behavior: SKIP
      CLIENT_UNINSTALL:
        max-attempts: 5

What is in this PR

• openframe-machine-delivery (new): DeliverySeed, DeliverySpec, DeliverySpecRegistry, DeliveryRequest, DeliveryDispatcher, DeliveryRecorder (+ mongo / noop), DeliveryTracker (+ mongo / noop), DeliveryProperties, DeliverySweepService, DeliveryWatchdogService, DeliveryFailureRecorder, DeliveryMetrics (openframe.delivery.retried{type}, openframe.delivery.failed{type,reason}), DeliverySweepScheduler, MachineOnlineStatus. 27 unit tests.
• data-mongo-common: DeliveryType, DeliveryStatus, DeliveryFailure, MachineDelivery in document.delivery; data-mongo-sync: MachineDeliveryRepository.
• data-nats: ToolInstallationDeliverySpec, ClientUninstallDeliverySpec with nested Seed; publish(machineId, message) overloads and public buildMessage on the two publishers (existing callers unchanged). 5 tests.
• Root pom: module + dependencyManagement entry; shedlock-spring pinned like client-core (not managed by the parent).

Nothing outside the module and data-nats changes behaviour. Every flag-dependent bean has a no-op counterpart selected by matchIfMissing, so services that include the module without the YAML boot as before.

Plan by PRs

1. this PR — module, engine, dispatcher, two specs. Prod: no change.
2. oss-lib, stacked on this branch — usage: type/targetId on ScriptExecutionAcknowledgeMessage + listener routing; complete() in InstalledAgentService / AgentUninstallService; ToolInstallationService and ForceClientUninstallService through DeliveryDispatcher. Flag still off everywhere.
3. oss-lib, Rust — tool_installation_message_listener.rs, client_uninstall_message_listener.rs → core subscribe + ack with type/targetId (pattern: execution_listener.rs); delete dead tool_uninstall_message_listener.rs; explicit failure result. Owner needed.
4. oss-lib — publishers send core + JetStream under legacy-jetstream-publish; row only for machines whose agent ≥ the version from 3 (installed_agents, openframe-client), so old agents never produce false FAILED.
5. saas-tenant — openframe.delivery YAML block, enabled=true (client, api, management), sweep.enabled=true (client only), pin oss.libs.version; dev → stage → prod.
6. saas-shared — Grafana rules on openframe_delivery_failed_total{type,reason} in openframe-rmm.yaml.
7. oss-lib — scheduled scripts onto the engine (ScheduleFireDispatcher → dispatcher, ScriptScheduleDeliverySpec in client-core because its onFailed needs ScheduleJobExecutionWatchdogService, delete ScriptDeliveryRetry* and ScheduleDeliveryRepublisher); ad-hoc scripts from api-lib gain the retry they lack today.
8. oss-lib + Rust — client/tool updates: machine.all.* → per-machine fan-out with version on the row; delete PublishState + AgentVersionUpdatePublishFallbackScheduler; management dispatches through the same specs; CLIENT_UPDATE last (it is the agent rollout channel).
9. saas-tenant cleanup once the fleet is on the new agent — legacy-jetstream-publish=false, drop $JS.API.* from the NATS configmap, delete the streams explicitly (their consumers go with them), Mongock drop script_delivery_retry, remove openframe.rmm.execution.retry.*.

In parallel, outside the plan: NATS service/admin passwords into ExternalSecret per tenant.

Open decisions

1. Ack format: extend ScriptExecutionAcknowledgeMessage with type/targetId (PR 2) vs. a new subject.
2. Update broadcasts: fan-out rows per machine (proposed) vs. pull model.
3. PENDING_DELETION on exhausted uninstall: revert + alert (done here in ClientUninstallDeliverySpec) vs. leave.
4. Agent-side durable consumers only (proposed) vs. server-side too.

Verification

mvn -pl openframe-machine-delivery,openframe-data-nats,openframe-client-core,openframe-api-service-core -am test -Dtest='Delivery*Test,*DeliverySpecTest,*DeliveryRecorderTest,MongoDeliveryTrackerTest,MachineOnlineStatusTest' — build green on JDK 21; existing client-core and api-service-core sources are byte-identical to main.

🤖 Generated with Claude Code

https://claude.ai/code/session_01BnYUpYgKvuMw6hB5VZSimY