Reliable Automation Execution Core v1 Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Build a durable, restart-safe automation runtime that converts telemetry into deduplicated endpoint events, creates idempotent command intents, dispatches correlated MQTT RPC commands, and only marks success after applied-state evidence is recorded.

Architecture: Keep the existing rule and endpoint authoring model, but replace inline telemetry dispatch with a staged runtime under backend/src/automation/. PostgreSQL is the source of truth for endpoint events, rule executions, command intents, and command receipts; Redis is only used for short-TTL dedupe, worker leases, inflight correlation, and per-endpoint dispatch serialization. Nest schedule-driven workers advance the state machine in small polling loops so the first production-grade version fits the current stack without introducing a queue platform.

Tech Stack: NestJS, TypeORM, PostgreSQL, Redis, MQTT, Jest, Bun, Biome

File Structure

New files

  • backend/src/automation/entities/automation-endpoint-event.entity.ts
    Durable normalized telemetry events for automation evaluation and replay.
  • backend/src/automation/entities/automation-command-intent.entity.ts
    Durable command ledger and retry state machine anchor.
  • backend/src/automation/entities/device-command-receipt.entity.ts
    Raw ack/result/state-report evidence table.
  • backend/src/automation/enums/automation-endpoint-event-status.enum.ts
    Event worker lifecycle (pending, processed, discarded, failed).
  • backend/src/automation/enums/automation-command-intent-status.enum.ts
    Command lifecycle (pending_dispatch, dispatched, acknowledged, applied, retrying, failed, expired, uncertain).
  • backend/src/automation/enums/device-command-receipt-type.enum.ts
    Receipt kinds (rpc_ack, rpc_result, state_report, timeout_mark, retry_mark).
  • backend/src/automation/runtime/automation-runtime.types.ts
    Shared runtime payload and policy types.
  • backend/src/automation/runtime/automation-runtime-policy.ts
    Pure functions for dedupe keys, idempotency keys, retry policy, and state transitions.
  • backend/src/automation/runtime/automation-endpoint-events.service.ts
    Persists deduplicated endpoint events and owns event status updates.
  • backend/src/automation/runtime/automation-command-payload.factory.ts
    Builds RPC payloads from endpoint transport binding plus runtime correlation fields.
  • backend/src/automation/runtime/automation-command-dispatcher.service.ts
    Dispatches command intents, acquires serialization lease, and updates durable state.
  • backend/src/automation/runtime/automation-reconciliation.service.ts
    Persists receipts, detects applied evidence, handles timeout and uncertain transitions.
  • backend/src/automation/runtime/automation-runtime-worker.service.ts
    Poll workers for event evaluation, dispatch, retry, and recovery.
  • backend/src/automation/runtime/automation-runtime-diagnostics.service.ts
    Timeline and backlog queries for production debugging.
  • backend/src/automation/runtime/automation-runtime-policy.spec.ts
  • backend/src/automation/runtime/automation-endpoint-events.service.spec.ts
  • backend/src/automation/runtime/automation-command-payload.factory.spec.ts
  • backend/src/automation/runtime/automation-command-dispatcher.service.spec.ts
  • backend/src/automation/runtime/automation-reconciliation.service.spec.ts
  • backend/src/automation/runtime/automation-runtime-worker.service.spec.ts
  • backend/src/automation/runtime/automation-runtime-diagnostics.service.spec.ts
  • backend/src/automation/automation-runtime.integration.spec.ts
  • backend/src/automation/automation-runtime.contract.spec.ts
  • backend/src/migrations/1778371200000-CreateReliableAutomationRuntimeTables.ts

Modified files

  • backend/src/automation/automation.module.ts
    Register new entities, worker services, and MQTT/Redis dependencies.
  • backend/src/automation/automation-engine.service.ts
    Stop publishing MQTT directly; create executions plus command intents only.
  • backend/src/automation/automation-engine.service.spec.ts
    Update expectations from direct dispatch to durable command-intent creation.
  • backend/src/automation/endpoint-event-normalizer.service.ts
    Preserve enough endpoint context for event dedupe and applied-state reconciliation.
  • backend/src/automation/entities/automation-rule-execution.entity.ts
    Expand execution envelope with triggering event, command intent, outcome, and evidence metadata.
  • backend/src/automation/enums/automation-rule-execution-status.enum.ts
    Align execution envelope statuses with runtime lifecycle.
  • backend/src/telemetry/telemetry.service.ts
    Persist telemetry and endpoint events only; remove inline automation dispatch.
  • backend/src/telemetry/telemetry.service.spec.ts
    Assert durable event creation instead of inline rule execution.
  • backend/src/mqtt/mqtt.service.ts
    Add publish helper for correlated automation RPC and route responses/status to reconciliation.
  • backend/src/mqtt/mqtt.service.spec.ts
    Assert correlation payload and receipt forwarding behavior.
  • backend/src/mqtt/dto/rpc-request.dto.ts
    Add correlation fields needed by firmware contract.
  • backend/src/devices/device-command-dispatch.service.ts
    Convert to transport-binding helper used by automation dispatcher and existing manual flows.
  • backend/src/devices/device-command-dispatch.service.spec.ts
    Assert payload-building behavior instead of immediate durable completion assumptions.
  • backend/src/redis/redis.service.ts
    Add lease and short-TTL helper methods needed by worker coordination.
  • backend/src/app.module.ts
    Add new automation runtime entities to TypeORM registration.

Task 1: Define runtime statuses and pure policy helpers

Files:

  • Create: backend/src/automation/enums/automation-endpoint-event-status.enum.ts

  • Create: backend/src/automation/enums/automation-command-intent-status.enum.ts

  • Create: backend/src/automation/enums/device-command-receipt-type.enum.ts

  • Create: backend/src/automation/runtime/automation-runtime.types.ts

  • Create: backend/src/automation/runtime/automation-runtime-policy.ts

  • Test: backend/src/automation/runtime/automation-runtime-policy.spec.ts

  • Modify: backend/src/automation/enums/automation-rule-execution-status.enum.ts

  • Step 1: Write the failing policy tests

import {
  buildCommandIdempotencyKey,
  buildEndpointEventDedupeKey,
  classifyDispatchFailure,
  computeNextAttemptAt,
  shouldRetryIntent,
  transitionIntentStatus,
} from '@/automation/runtime/automation-runtime-policy';
import { AutomationCommandIntentStatus } from '@/automation/enums/automation-command-intent-status.enum';

describe('automation-runtime-policy', () => {
  it('builds a stable endpoint dedupe key from device, endpoint, timestamp, value, and fingerprint', () => {
    expect(
      buildEndpointEventDedupeKey({
        deviceId: 'device-1',
        deviceInstanceEndpointId: 'endpoint-1',
        eventTimestamp: new Date('2026-05-10T10:00:00.000Z'),
        normalizedValue: 3.8,
        telemetryFingerprint: 'telemetry:abc',
      }),
    ).toBe('device-1:endpoint-1:1746871200000:3.8:telemetry:abc');
  });

  it('builds a stable command idempotency key from rule, target endpoint, target value, and triggering event', () => {
    expect(
      buildCommandIdempotencyKey({
        ruleId: 'rule-1',
        targetDeviceEndpointId: 'target-endpoint-1',
        targetValue: true,
        triggeringEventId: 'event-1',
      }),
    ).toBe('rule-1:target-endpoint-1:true:event-1');
  });

  it('uses exponential backoff with jitter-free reference schedule 5s, 15s, 45s', () => {
    expect(computeNextAttemptAt(new Date('2026-05-10T10:00:00.000Z'), 1)?.toISOString()).toBe(
      '2026-05-10T10:00:05.000Z',
    );
    expect(computeNextAttemptAt(new Date('2026-05-10T10:00:00.000Z'), 2)?.toISOString()).toBe(
      '2026-05-10T10:00:15.000Z',
    );
    expect(computeNextAttemptAt(new Date('2026-05-10T10:00:00.000Z'), 3)?.toISOString()).toBe(
      '2026-05-10T10:00:45.000Z',
    );
  });

  it('only retries transient dispatch failures', () => {
    expect(shouldRetryIntent({ attemptCount: 1, maxAttempts: 3, failureClass: 'timeout' })).toBe(
      true,
    );
    expect(
      shouldRetryIntent({ attemptCount: 1, maxAttempts: 3, failureClass: 'binding_invalid' }),
    ).toBe(false);
  });

  it('guards invalid lifecycle transitions', () => {
    expect(
      transitionIntentStatus(
        AutomationCommandIntentStatus.PENDING_DISPATCH,
        'dispatch_succeeded',
      ),
    ).toBe(AutomationCommandIntentStatus.DISPATCHED);

    expect(() =>
      transitionIntentStatus(AutomationCommandIntentStatus.APPLIED, 'dispatch_succeeded'),
    ).toThrow('Invalid automation command transition');
  });

  it('classifies permanent versus transient errors', () => {
    expect(classifyDispatchFailure('Device is not online')).toBe('device_offline');
    expect(classifyDispatchFailure('Target endpoint binding is invalid')).toBe('binding_invalid');
  });
});
  • Step 2: Run the policy test and verify it fails

Run from backend/: bun test -- src/automation/runtime/automation-runtime-policy.spec.ts

Expected: FAIL with missing module errors for automation-runtime-policy and the new enums.

  • Step 3: Implement the minimal runtime policy layer
export enum AutomationCommandIntentStatus {
  PENDING_DISPATCH = 'pending_dispatch',
  DISPATCHED = 'dispatched',
  ACKNOWLEDGED = 'acknowledged',
  APPLIED = 'applied',
  RETRYING = 'retrying',
  FAILED = 'failed',
  EXPIRED = 'expired',
  UNCERTAIN = 'uncertain',
}

export enum AutomationEndpointEventStatus {
  PENDING = 'pending',
  PROCESSED = 'processed',
  DISCARDED = 'discarded',
  FAILED = 'failed',
}

export enum DeviceCommandReceiptType {
  RPC_ACK = 'rpc_ack',
  RPC_RESULT = 'rpc_result',
  STATE_REPORT = 'state_report',
  TIMEOUT_MARK = 'timeout_mark',
  RETRY_MARK = 'retry_mark',
}

export enum AutomationRuleExecutionStatus {
  PENDING = 'pending',
  SKIPPED = 'skipped',
  DISPATCHED = 'dispatched',
  ACKNOWLEDGED = 'acknowledged',
  APPLIED = 'applied',
  RETRYING = 'retrying',
  FAILED = 'failed',
  EXPIRED = 'expired',
  UNCERTAIN = 'uncertain',
}

const RETRY_DELAYS_MS = [5000, 15000, 45000] as const;

export function buildEndpointEventDedupeKey(input: {
  deviceId: string;
  deviceInstanceEndpointId: string;
  eventTimestamp: Date;
  normalizedValue: string | number | boolean;
  telemetryFingerprint: string;
}): string {
  return [
    input.deviceId,
    input.deviceInstanceEndpointId,
    input.eventTimestamp.getTime(),
    String(input.normalizedValue),
    input.telemetryFingerprint,
  ].join(':');
}

export function buildCommandIdempotencyKey(input: {
  ruleId: string;
  targetDeviceEndpointId: string;
  targetValue: string | number | boolean;
  triggeringEventId: string;
}): string {
  return [
    input.ruleId,
    input.targetDeviceEndpointId,
    String(input.targetValue),
    input.triggeringEventId,
  ].join(':');
}

export function computeNextAttemptAt(base: Date, attemptNumber: number): Date | null {
  const delay = RETRY_DELAYS_MS[attemptNumber - 1];
  return delay ? new Date(base.getTime() + delay) : null;
}
  • Step 4: Run the policy test again and verify it passes

Run from backend/: bun test -- src/automation/runtime/automation-runtime-policy.spec.ts

Expected: PASS with 6 passing tests.

  • Step 5: Commit
git add backend/src/automation/enums/automation-endpoint-event-status.enum.ts \
  backend/src/automation/enums/automation-command-intent-status.enum.ts \
  backend/src/automation/enums/device-command-receipt-type.enum.ts \
  backend/src/automation/enums/automation-rule-execution-status.enum.ts \
  backend/src/automation/runtime/automation-runtime.types.ts \
  backend/src/automation/runtime/automation-runtime-policy.ts \
  backend/src/automation/runtime/automation-runtime-policy.spec.ts
git commit -m "feat: add automation runtime policy primitives"

Task 2: Add durable automation runtime tables and execution envelope fields

Files:

  • Create: backend/src/automation/entities/automation-endpoint-event.entity.ts

  • Create: backend/src/automation/entities/automation-command-intent.entity.ts

  • Create: backend/src/automation/entities/device-command-receipt.entity.ts

  • Create: backend/src/migrations/1778371200000-CreateReliableAutomationRuntimeTables.ts

  • Test: backend/src/automation/entities/automation-runtime.entities.spec.ts

  • Modify: backend/src/automation/entities/automation-rule-execution.entity.ts

  • Modify: backend/src/automation/automation.module.ts

  • Modify: backend/src/app.module.ts

  • Step 1: Write the failing entity metadata test

import { DataSource } from 'typeorm';
import { AutomationCommandIntent } from '@/automation/entities/automation-command-intent.entity';
import { AutomationEndpointEvent } from '@/automation/entities/automation-endpoint-event.entity';
import { AutomationRuleExecution } from '@/automation/entities/automation-rule-execution.entity';
import { DeviceCommandReceipt } from '@/automation/entities/device-command-receipt.entity';

describe('automation runtime entities', () => {
  const dataSource = new DataSource({ type: 'postgres', entities: [] });

  it('defines the durable runtime tables and execution envelope fields', () => {
    const eventMetadata = dataSource.getMetadata(AutomationEndpointEvent);
    const intentMetadata = dataSource.getMetadata(AutomationCommandIntent);
    const receiptMetadata = dataSource.getMetadata(DeviceCommandReceipt);
    const executionMetadata = dataSource.getMetadata(AutomationRuleExecution);

    expect(eventMetadata.tableName).toBe('automation_endpoint_events');
    expect(intentMetadata.findColumnWithPropertyName('idempotencyKey')).toBeDefined();
    expect(receiptMetadata.findColumnWithPropertyName('receiptType')).toBeDefined();
    expect(executionMetadata.findColumnWithPropertyName('triggeringEventId')).toBeDefined();
    expect(executionMetadata.findColumnWithPropertyName('commandIntentId')).toBeDefined();
  });
});
  • Step 2: Run the entity test and verify it fails

Run from backend/: bun test -- src/automation/entities/automation-runtime.entities.spec.ts

Expected: FAIL because the three new entities and expanded execution fields do not exist yet.

  • Step 3: Implement the entities, migration, and registrations
@Entity('automation_endpoint_events')
export class AutomationEndpointEvent {
  @PrimaryGeneratedColumn('uuid')
  id: string;

  @Column({ name: 'organization_id', type: 'uuid', nullable: true })
  organizationId: string | null;

  @Column({ name: 'farm_id', type: 'uuid' })
  farmId: string;

  @Column({ name: 'device_id', type: 'uuid' })
  deviceId: string;

  @Column({ name: 'device_instance_endpoint_id', type: 'uuid' })
  deviceInstanceEndpointId: string;

  @Column({ name: 'source_telemetry_id', type: 'uuid', nullable: true })
  sourceTelemetryId: string | null;

  @Column({ name: 'event_timestamp', type: 'timestamptz' })
  eventTimestamp: Date;

  @Column({ name: 'normalized_value', type: 'jsonb' })
  normalizedValue: string | number | boolean;

  @Column({ name: 'raw_payload_snapshot', type: 'jsonb' })
  rawPayloadSnapshot: Record<string, unknown>;

  @Column({ name: 'telemetry_fingerprint', type: 'varchar', length: 255 })
  telemetryFingerprint: string;

  @Column({ name: 'dedupe_key', type: 'varchar', length: 255, unique: true })
  dedupeKey: string;

  @Column({
    name: 'processing_status',
    type: 'enum',
    enum: AutomationEndpointEventStatus,
    default: AutomationEndpointEventStatus.PENDING,
  })
  processingStatus: AutomationEndpointEventStatus;
}

@Entity('automation_command_intents')
export class AutomationCommandIntent {
  @PrimaryGeneratedColumn('uuid')
  id: string;

  @Column({ name: 'rule_execution_id', type: 'uuid' })
  ruleExecutionId: string;

  @Column({ name: 'triggering_event_id', type: 'uuid' })
  triggeringEventId: string;

  @Column({ name: 'rule_id', type: 'uuid' })
  ruleId: string;

  @Column({ name: 'target_device_id', type: 'uuid' })
  targetDeviceId: string;

  @Column({ name: 'target_device_endpoint_id', type: 'uuid' })
  targetDeviceEndpointId: string;

  @Column({ name: 'target_serial_number', type: 'varchar', length: 100 })
  targetSerialNumber: string;

  @Column({ name: 'command_method', type: 'varchar', length: 100 })
  commandMethod: string;

  @Column({ name: 'command_params', type: 'jsonb' })
  commandParams: Record<string, unknown>;

  @Column({ name: 'target_value', type: 'jsonb' })
  targetValue: string | number | boolean;

  @Column({ name: 'idempotency_key', type: 'varchar', length: 255, unique: true })
  idempotencyKey: string;
}

@Entity('device_command_receipts')
export class DeviceCommandReceipt {
  @PrimaryGeneratedColumn('uuid')
  id: string;

  @Column({ name: 'command_intent_id', type: 'uuid' })
  commandIntentId: string;

  @Column({ name: 'device_id', type: 'uuid' })
  deviceId: string;

  @Column({ name: 'serial_number', type: 'varchar', length: 100 })
  serialNumber: string;

  @Column({ name: 'receipt_type', type: 'enum', enum: DeviceCommandReceiptType })
  receiptType: DeviceCommandReceiptType;

  @Column({ type: 'jsonb' })
  payload: Record<string, unknown>;

  @Column({ name: 'received_at', type: 'timestamptz', default: () => 'now()' })
  receivedAt: Date;

  @Column({ name: 'correlation_key', type: 'varchar', length: 255 })
  correlationKey: string;
}

@Column({ name: 'triggering_event_id', type: 'uuid', nullable: true })
triggeringEventId: string | null;

@Column({ name: 'command_intent_id', type: 'uuid', nullable: true })
commandIntentId: string | null;

@Column({ name: 'final_outcome', type: 'varchar', length: 50, nullable: true })
finalOutcome: string | null;

@Column({ name: 'evidence_snapshot', type: 'jsonb', nullable: true })
evidenceSnapshot: Record<string, unknown> | null;

export class CreateReliableAutomationRuntimeTables1778371200000 implements MigrationInterface {
  name = 'CreateReliableAutomationRuntimeTables1778371200000';

  public async up(queryRunner: QueryRunner): Promise<void> {
    await queryRunner.query(`
      CREATE TABLE "automation_endpoint_events" (
        "id" UUID PRIMARY KEY DEFAULT gen_random_uuid(),
        "organization_id" UUID REFERENCES organizations(id) ON DELETE CASCADE,
        "farm_id" UUID NOT NULL REFERENCES areas(id) ON DELETE CASCADE,
        "device_id" UUID NOT NULL REFERENCES devices(id) ON DELETE CASCADE,
        "device_instance_endpoint_id" UUID NOT NULL REFERENCES device_instance_endpoints(id) ON DELETE CASCADE,
        "source_telemetry_id" UUID REFERENCES device_telemetry(id) ON DELETE SET NULL,
        "event_timestamp" TIMESTAMPTZ NOT NULL,
        "normalized_value" JSONB NOT NULL,
        "raw_payload_snapshot" JSONB NOT NULL,
        "telemetry_fingerprint" VARCHAR(255) NOT NULL,
        "dedupe_key" VARCHAR(255) NOT NULL,
        "processing_status" VARCHAR(50) NOT NULL DEFAULT 'pending',
        "created_at" TIMESTAMPTZ NOT NULL DEFAULT NOW(),
        "updated_at" TIMESTAMPTZ NOT NULL DEFAULT NOW(),
        CONSTRAINT "uq_automation_endpoint_events_dedupe_key" UNIQUE ("dedupe_key")
      );
    `);
  }
}
  • Step 4: Run the metadata test and lint/format the touched files

Run from backend/: bun test -- src/automation/entities/automation-runtime.entities.spec.ts

Expected: PASS with entity metadata resolving.

Run from backend/: bun run lint:fix

Expected: PASS with Biome fixes applied.

Run from backend/: bun run format

Expected: PASS with formatted entity and migration files.

  • Step 5: Commit
git add backend/src/automation/entities/automation-endpoint-event.entity.ts \
  backend/src/automation/entities/automation-command-intent.entity.ts \
  backend/src/automation/entities/device-command-receipt.entity.ts \
  backend/src/automation/entities/automation-rule-execution.entity.ts \
  backend/src/automation/enums/automation-rule-execution-status.enum.ts \
  backend/src/automation/automation.module.ts \
  backend/src/app.module.ts \
  backend/src/migrations/1778371200000-CreateReliableAutomationRuntimeTables.ts \
  backend/src/automation/entities/automation-runtime.entities.spec.ts
git commit -m "feat: add durable automation runtime schema"

Task 3: Persist endpoint events during telemetry ingestion and remove inline dispatch

Files:

  • Create: backend/src/automation/runtime/automation-endpoint-events.service.ts

  • Test: backend/src/automation/runtime/automation-endpoint-events.service.spec.ts

  • Modify: backend/src/telemetry/telemetry.service.ts

  • Modify: backend/src/telemetry/telemetry.service.spec.ts

  • Modify: backend/src/automation/endpoint-event-normalizer.service.ts

  • Step 1: Write the failing telemetry ingestion tests

it('persists normalized automation endpoint events and does not dispatch automation inline', async () => {
  endpointEventNormalizerService.normalizeTelemetryEvent.mockResolvedValue([
    {
      organizationId: 'org-1',
      farmId: 'farm-1',
      deviceId: 'device-1',
      deviceInstanceEndpointId: 'endpoint-1',
      eventTimestamp: new Date('2026-05-10T10:00:00.000Z'),
      normalizedValue: 3.8,
      rawPayloadSnapshot: { do: 3.8 },
      telemetryFingerprint: 'telemetry:abc',
    },
  ]);

  await service.save({
    deviceId: 'device-1',
    timestamp: new Date('2026-05-10T10:00:00.000Z'),
    data: { do: 3.8 },
  });

  expect(automationEndpointEventsService.persistNormalizedEvents).toHaveBeenCalledWith(
    expect.objectContaining({ deviceId: 'device-1' }),
  );
  expect(automationEngineService.handleEndpointEvent).not.toHaveBeenCalled();
});

it('deduplicates replayed telemetry events on dedupe key', async () => {
  repository.save.mockRejectedValueOnce(Object.assign(new Error('duplicate key'), { code: '23505' }));

  await expect(
    automationEndpointEventsService.persistNormalizedEvents([
      {
        organizationId: 'org-1',
        farmId: 'farm-1',
        deviceId: 'device-1',
        deviceInstanceEndpointId: 'endpoint-1',
        eventTimestamp: new Date('2026-05-10T10:00:00.000Z'),
        normalizedValue: 3.8,
        rawPayloadSnapshot: { do: 3.8 },
        telemetryFingerprint: 'telemetry:abc',
      },
    ]),
  ).resolves.toEqual([]);
});
  • Step 2: Run the telemetry and event persistence tests and verify they fail

Run from backend/: bun test -- src/telemetry/telemetry.service.spec.ts src/automation/runtime/automation-endpoint-events.service.spec.ts

Expected: FAIL because AutomationEndpointEventsService does not exist and TelemetryService still calls automationEngineService.handleEndpointEvent.

  • Step 3: Implement durable endpoint-event persistence
@Injectable()
export class AutomationEndpointEventsService {
  constructor(
    @InjectRepository(AutomationEndpointEvent)
    private endpointEventRepository: Repository<AutomationEndpointEvent>,
  ) {}

  async persistNormalizedEvents(
    events: PersistableEndpointEvent[],
  ): Promise<AutomationEndpointEvent[]> {
    const persisted: AutomationEndpointEvent[] = [];

    for (const event of events) {
      const entity = this.endpointEventRepository.create({
        ...event,
        dedupeKey: buildEndpointEventDedupeKey({
          deviceId: event.deviceId,
          deviceInstanceEndpointId: event.deviceInstanceEndpointId,
          eventTimestamp: event.eventTimestamp,
          normalizedValue: event.normalizedValue,
          telemetryFingerprint: event.telemetryFingerprint,
        }),
        processingStatus: AutomationEndpointEventStatus.PENDING,
      });

      try {
        persisted.push(await this.endpointEventRepository.save(entity));
      } catch (error) {
        if ((error as { code?: string }).code === '23505') {
          continue;
        }
        throw error;
      }
    }

    return persisted;
  }
}

const endpointEvents = await this.endpointEventNormalizerService.normalizeTelemetryEvent({
  deviceId: dto.deviceId,
  timestamp,
  rawPayload: dto.data,
  sourceTelemetryId: telemetry.id,
});

await this.automationEndpointEventsService.persistNormalizedEvents(endpointEvents);
  • Step 4: Run the focused tests and then lint/format

Run from backend/: bun test -- src/telemetry/telemetry.service.spec.ts src/automation/runtime/automation-endpoint-events.service.spec.ts

Expected: PASS with telemetry persisting endpoint events and no inline dispatch.

Run from backend/: bun run lint:fix

Expected: PASS.

Run from backend/: bun run format

Expected: PASS.

  • Step 5: Commit
git add backend/src/automation/runtime/automation-endpoint-events.service.ts \
  backend/src/automation/runtime/automation-endpoint-events.service.spec.ts \
  backend/src/automation/endpoint-event-normalizer.service.ts \
  backend/src/telemetry/telemetry.service.ts \
  backend/src/telemetry/telemetry.service.spec.ts
git commit -m "refactor: persist automation endpoint events from telemetry"

Task 4: Evaluate pending endpoint events into executions and command intents

Files:

  • Modify: backend/src/automation/automation-engine.service.ts

  • Modify: backend/src/automation/automation-engine.service.spec.ts

  • Create: backend/src/automation/runtime/automation-runtime-worker.service.ts

  • Test: backend/src/automation/runtime/automation-runtime-worker.service.spec.ts

  • Step 1: Write the failing evaluator and worker tests

it('creates a pending execution and a pending_dispatch command intent when a rule matches', async () => {
  ruleRepository.find.mockResolvedValue([
    {
      id: 'rule-1',
      status: 'active',
      cooldownSeconds: 0,
      conditions: [{ sourceDeviceEndpointId: 'source-endpoint-1', operator: '<', thresholdValue: 4.5 }],
      actions: [{
        targetDeviceEndpointId: 'target-endpoint-1',
        targetValue: true,
        targetEndpoint: {
          id: 'target-endpoint-1',
          isEnabled: true,
          isWritable: true,
          device: {
            id: 'device-2',
            serialNumber: 'GW-001',
            areaId: 'farm-1',
            inventoryStatus: 'assigned',
            status: 'online',
          },
        },
      }],
    },
  ]);

  await service.handleEndpointEvent(eventEntity);

  expect(commandIntentRepository.save).toHaveBeenCalledWith(
    expect.objectContaining({
      ruleId: 'rule-1',
      status: 'pending_dispatch',
      targetSerialNumber: 'GW-001',
    }),
  );
  expect(dispatchService.dispatchEndpointValue).not.toHaveBeenCalled();
});

it('worker marks unmatched events as discarded', async () => {
  endpointEventRepository.find.mockResolvedValue([eventEntity]);
  ruleRepository.find.mockResolvedValue([]);

  await worker.processPendingEndpointEvents();

  expect(endpointEventRepository.save).toHaveBeenCalledWith(
    expect.objectContaining({ id: eventEntity.id, processingStatus: 'discarded' }),
  );
});
  • Step 2: Run the evaluator tests and verify they fail

Run from backend/: bun test -- src/automation/automation-engine.service.spec.ts src/automation/runtime/automation-runtime-worker.service.spec.ts

Expected: FAIL because AutomationEngineService still dispatches through MQTT and the worker service does not exist.

  • Step 3: Implement event evaluation without direct MQTT publish
async handleEndpointEvent(event: AutomationEndpointEvent): Promise<'processed' | 'discarded'> {
  const rules = await this.ruleRepository.find({
    where: { farmId: event.farmId, status: AutomationRuleStatus.ACTIVE },
    relations: ['conditions', 'actions', 'actions.targetEndpoint', 'actions.targetEndpoint.device'],
  });

  let matched = false;

  for (const rule of rules) {
    const condition = rule.conditions[0];
    const action = rule.actions[0];
    const targetEndpoint = action?.targetEndpoint;
    const targetDevice = targetEndpoint?.device;

    if (!condition || !action) continue;
    if (condition.sourceDeviceEndpointId !== event.deviceInstanceEndpointId) continue;
    if (!this.evaluateCondition(event.normalizedValue, condition.operator, condition.thresholdValue)) {
      continue;
    }

    matched = true;

    const execution = await this.executionRepository.save({
      ruleId: rule.id,
      triggeringEventId: event.id,
      triggeredAt: event.eventTimestamp,
      sourceTelemetryPayload: event.rawPayloadSnapshot,
      evaluatedConditionSnapshot: {
        operator: condition.operator,
        thresholdValue: condition.thresholdValue,
        actualValue: event.normalizedValue,
      },
      status: AutomationRuleExecutionStatus.PENDING,
      failureReason: null,
    });

    await this.commandIntentRepository.save({
      ruleExecutionId: execution.id,
      triggeringEventId: event.id,
      ruleId: rule.id,
      targetDeviceId: targetDevice.id,
      targetDeviceEndpointId: action.targetDeviceEndpointId,
      targetSerialNumber: targetDevice.serialNumber,
      commandMethod: 'setIO',
      commandParams: { io1: action.targetValue },
      targetValue: action.targetValue,
      desiredStateFingerprint: JSON.stringify({ endpoint: action.targetDeviceEndpointId, value: action.targetValue }),
      idempotencyKey: buildCommandIdempotencyKey({
        ruleId: rule.id,
        targetDeviceEndpointId: action.targetDeviceEndpointId,
        targetValue: action.targetValue as boolean,
        triggeringEventId: event.id,
      }),
      status: AutomationCommandIntentStatus.PENDING_DISPATCH,
      attemptCount: 0,
      maxAttempts: 3,
      nextAttemptAt: new Date(),
    });
  }

  return matched ? 'processed' : 'discarded';
}

@Cron(CronExpression.EVERY_5_SECONDS)
async processPendingEndpointEvents(): Promise<void> {
  const events = await this.endpointEventRepository.find({
    where: { processingStatus: AutomationEndpointEventStatus.PENDING },
    order: { eventTimestamp: 'ASC' },
    take: 50,
  });

  for (const event of events) {
    const outcome = await this.automationEngineService.handleEndpointEvent(event);
    event.processingStatus =
      outcome === 'processed'
        ? AutomationEndpointEventStatus.PROCESSED
        : AutomationEndpointEventStatus.DISCARDED;
    await this.endpointEventRepository.save(event);
  }
}
  • Step 4: Run tests and then lint/format

Run from backend/: bun test -- src/automation/automation-engine.service.spec.ts src/automation/runtime/automation-runtime-worker.service.spec.ts

Expected: PASS with matching rules creating command intents and worker discarding unmatched events.

Run from backend/: bun run lint:fix

Expected: PASS.

Run from backend/: bun run format

Expected: PASS.

  • Step 5: Commit
git add backend/src/automation/automation-engine.service.ts \
  backend/src/automation/automation-engine.service.spec.ts \
  backend/src/automation/runtime/automation-runtime-worker.service.ts \
  backend/src/automation/runtime/automation-runtime-worker.service.spec.ts
git commit -m "feat: evaluate automation events into command intents"

Task 5: Dispatch command intents through MQTT with correlation payload and endpoint serialization

Files:

  • Create: backend/src/automation/runtime/automation-command-payload.factory.ts

  • Test: backend/src/automation/runtime/automation-command-payload.factory.spec.ts

  • Create: backend/src/automation/runtime/automation-command-dispatcher.service.ts

  • Test: backend/src/automation/runtime/automation-command-dispatcher.service.spec.ts

  • Modify: backend/src/mqtt/dto/rpc-request.dto.ts

  • Modify: backend/src/mqtt/mqtt.service.ts

  • Modify: backend/src/mqtt/mqtt.service.spec.ts

  • Modify: backend/src/devices/device-command-dispatch.service.ts

  • Modify: backend/src/devices/device-command-dispatch.service.spec.ts

  • Step 1: Write the failing dispatch and payload tests

it('builds an automation RPC payload with commandId, sentAt, and deadlineAt', () => {
  expect(
    factory.build({
      commandId: 'cmd-1',
      method: 'setIO',
      params: { io1: true },
      sentAt: new Date('2026-05-10T10:00:00.000Z'),
      deadlineAt: new Date('2026-05-10T10:00:05.000Z'),
    }),
  ).toEqual({
    commandId: 'cmd-1',
    method: 'setIO',
    params: { io1: true },
    sentAt: '2026-05-10T10:00:00.000Z',
    deadlineAt: '2026-05-10T10:00:05.000Z',
  });
});

it('dispatches a pending intent, writes correlation ids, and acquires endpoint serialization lease', async () => {
  redisService.acquireLease.mockResolvedValue(true);
  mqttService.publishAutomationCommand.mockResolvedValue({ requestId: 'req-1' });

  await service.dispatchPendingIntent(intent);

  expect(redisService.acquireLease).toHaveBeenCalledWith(
    'automation:dispatch-lock:device-2:target-endpoint-1',
    expect.any(String),
    10,
  );
  expect(mqttService.publishAutomationCommand).toHaveBeenCalledWith(
    'GW-001',
    expect.objectContaining({ commandId: expect.stringMatching(/^cmd_/) }),
  );
  expect(commandIntentRepository.save).toHaveBeenCalledWith(
    expect.objectContaining({ status: 'dispatched', mqttRequestId: 'req-1' }),
  );
});
  • Step 2: Run the dispatch tests and verify they fail

Run from backend/: bun test -- src/automation/runtime/automation-command-payload.factory.spec.ts src/automation/runtime/automation-command-dispatcher.service.spec.ts src/mqtt/mqtt.service.spec.ts

Expected: FAIL because the payload factory, dispatcher, and publishAutomationCommand helper do not exist yet.

  • Step 3: Implement correlated dispatch and serialization
export class RpcRequestDto {
  @IsString()
  method: string;

  @IsObject()
  params: Record<string, unknown>;

  @IsOptional()
  @IsString()
  commandId?: string;

  @IsOptional()
  @IsISO8601()
  sentAt?: string;

  @IsOptional()
  @IsISO8601()
  deadlineAt?: string;
}

async publishAutomationCommand(
  serialNumber: string,
  command: RpcRequestDto,
): Promise<{ requestId: string }> {
  const requestId = uuidv4();
  const topic = `v1/devices/${serialNumber}/rpc/request/${requestId}`;

  await new Promise<void>((resolve, reject) => {
    this.client.publish(topic, JSON.stringify(command), { qos: 1 }, (error) => {
      if (error) {
        reject(error);
        return;
      }

      resolve();
    });
  });

  return { requestId };
}

async dispatchPendingIntent(intent: AutomationCommandIntent): Promise<void> {
  const leaseKey = `automation:dispatch-lock:${intent.targetDeviceId}:${intent.targetDeviceEndpointId}`;
  const leaseOwner = `${intent.id}:${Date.now()}`;
  const acquired = await this.redisService.acquireLease(leaseKey, leaseOwner, 10);

  if (!acquired) {
    return;
  }

  const commandId = `cmd_${intent.id.replace(/-/g, '')}`;
  const sentAt = new Date();
  const deadlineAt = new Date(sentAt.getTime() + 5000);
  const payload = this.payloadFactory.build({
    commandId,
    method: intent.commandMethod,
    params: intent.commandParams,
    sentAt,
    deadlineAt,
  });

  const { requestId } = await this.mqttService.publishAutomationCommand(
    intent.targetSerialNumber,
    payload,
  );

  intent.status = AutomationCommandIntentStatus.DISPATCHED;
  intent.deviceCommandId = commandId;
  intent.mqttRequestId = requestId;
  intent.attemptCount += 1;
  intent.lastDispatchedAt = sentAt;
  intent.nextAttemptAt = deadlineAt;

  await this.commandIntentRepository.save(intent);
  await this.redisService.setex(`automation:command:${commandId}`, 300, intent.id);
  await this.redisService.releaseLease(leaseKey, leaseOwner);
}
  • Step 4: Run focused tests and then lint/format

Run from backend/: bun test -- src/automation/runtime/automation-command-payload.factory.spec.ts src/automation/runtime/automation-command-dispatcher.service.spec.ts src/mqtt/mqtt.service.spec.ts src/devices/device-command-dispatch.service.spec.ts

Expected: PASS with correlation fields present and serialized dispatch enforced.

Run from backend/: bun run lint:fix

Expected: PASS.

Run from backend/: bun run format

Expected: PASS.

  • Step 5: Commit
git add backend/src/automation/runtime/automation-command-payload.factory.ts \
  backend/src/automation/runtime/automation-command-payload.factory.spec.ts \
  backend/src/automation/runtime/automation-command-dispatcher.service.ts \
  backend/src/automation/runtime/automation-command-dispatcher.service.spec.ts \
  backend/src/mqtt/dto/rpc-request.dto.ts \
  backend/src/mqtt/mqtt.service.ts \
  backend/src/mqtt/mqtt.service.spec.ts \
  backend/src/devices/device-command-dispatch.service.ts \
  backend/src/devices/device-command-dispatch.service.spec.ts
git commit -m "feat: dispatch automation commands with correlation"

Task 6: Persist receipts and reconcile ack versus applied evidence

Files:

  • Create: backend/src/automation/runtime/automation-reconciliation.service.ts

  • Test: backend/src/automation/runtime/automation-reconciliation.service.spec.ts

  • Modify: backend/src/mqtt/mqtt.service.ts

  • Modify: backend/src/mqtt/mqtt.service.spec.ts

  • Step 1: Write the failing reconciliation tests

it('marks a command acknowledged on rpc ack and applied on matching state evidence', async () => {
  commandIntentRepository.findOne.mockResolvedValue(intent);

  await service.recordRpcReceipt({
    serialNumber: 'GW-001',
    commandId: 'cmd_1',
    requestId: 'req-1',
    payload: { commandId: 'cmd_1', status: 'accepted', resultCode: 'ok', message: null },
  });

  expect(commandIntentRepository.save).toHaveBeenCalledWith(
    expect.objectContaining({ status: 'acknowledged' }),
  );

  await service.recordStateReport({
    serialNumber: 'GW-001',
    commandId: 'cmd_1',
    payload: { commandId: 'cmd_1', reportedState: { io1: true }, appliedAt: '2026-05-10T10:00:02.100Z' },
  });

  expect(commandIntentRepository.save).toHaveBeenCalledWith(
    expect.objectContaining({ status: 'applied' }),
  );
});

it('marks a command uncertain when ack exists but no applied evidence arrives before deadline', async () => {
  commandIntentRepository.find.mockResolvedValue([
    {
      ...intent,
      status: 'acknowledged',
      lastDispatchedAt: new Date('2026-05-10T10:00:00.000Z'),
      nextAttemptAt: new Date('2026-05-10T10:00:05.000Z'),
    },
  ]);

  await service.reconcileExpiredAcknowledgements(new Date('2026-05-10T10:00:10.000Z'));

  expect(commandIntentRepository.save).toHaveBeenCalledWith(
    expect.objectContaining({ status: 'uncertain' }),
  );
});
  • Step 2: Run reconciliation tests and verify they fail

Run from backend/: bun test -- src/automation/runtime/automation-reconciliation.service.spec.ts src/mqtt/mqtt.service.spec.ts

Expected: FAIL because MqttService only resolves in-memory promises and no durable receipt service exists.

  • Step 3: Implement durable receipt ingestion and applied-state reconciliation
async recordRpcReceipt(input: {
  serialNumber: string;
  commandId: string;
  requestId: string;
  payload: Record<string, unknown>;
}): Promise<void> {
  const intentId = await this.redisService.get(`automation:command:${input.commandId}`);
  if (!intentId) {
    return;
  }

  const intent = await this.commandIntentRepository.findOne({ where: { id: intentId } });
  if (!intent) {
    return;
  }

  await this.receiptRepository.save({
    commandIntentId: intent.id,
    deviceId: intent.targetDeviceId,
    serialNumber: input.serialNumber,
    receiptType: DeviceCommandReceiptType.RPC_ACK,
    payload: input.payload,
    correlationKey: input.commandId,
  });

  intent.status = AutomationCommandIntentStatus.ACKNOWLEDGED;
  await this.commandIntentRepository.save(intent);
}

async recordStateReport(input: {
  serialNumber: string;
  commandId: string;
  payload: Record<string, unknown>;
}): Promise<void> {
  const intentId = await this.redisService.get(`automation:command:${input.commandId}`);
  if (!intentId) {
    return;
  }

  const intent = await this.commandIntentRepository.findOne({ where: { id: intentId } });
  if (!intent) {
    return;
  }

  const reportedState = input.payload.reportedState as Record<string, unknown>;
  const desiredValue = Object.values(intent.commandParams)[0];
  const applied = Object.values(reportedState).some((value) => value === desiredValue);

  await this.receiptRepository.save({
    commandIntentId: intent.id,
    deviceId: intent.targetDeviceId,
    serialNumber: input.serialNumber,
    receiptType: DeviceCommandReceiptType.STATE_REPORT,
    payload: input.payload,
    correlationKey: input.commandId,
  });

  intent.status = applied
    ? AutomationCommandIntentStatus.APPLIED
    : AutomationCommandIntentStatus.UNCERTAIN;

  await this.commandIntentRepository.save(intent);
  await this.executionRepository.update(intent.ruleExecutionId, {
    status: applied
      ? AutomationRuleExecutionStatus.APPLIED
      : AutomationRuleExecutionStatus.UNCERTAIN,
    finalOutcome: intent.status,
    evidenceSnapshot: input.payload,
  });
}

private async handleRpcResponse(topic: string, payload: Record<string, unknown>) {
  const requestId = topic.match(/rpc\/response\/(.+)$/)?.[1];
  const serialNumber = this.extractSerialNumber(topic);
  const commandId = typeof payload.commandId === 'string' ? payload.commandId : undefined;

  if (!requestId || !serialNumber || !commandId) {
    return;
  }

  await this.automationReconciliationService.recordRpcReceipt({
    serialNumber,
    commandId,
    requestId,
    payload,
  });
}
  • Step 4: Run tests and then lint/format

Run from backend/: bun test -- src/automation/runtime/automation-reconciliation.service.spec.ts src/mqtt/mqtt.service.spec.ts

Expected: PASS with ack and applied evidence persisted durably.

Run from backend/: bun run lint:fix

Expected: PASS.

Run from backend/: bun run format

Expected: PASS.

  • Step 5: Commit
git add backend/src/automation/runtime/automation-reconciliation.service.ts \
  backend/src/automation/runtime/automation-reconciliation.service.spec.ts \
  backend/src/mqtt/mqtt.service.ts \
  backend/src/mqtt/mqtt.service.spec.ts
git commit -m "feat: reconcile automation command receipts"

Task 7: Add retry, timeout recovery, Redis leases, and restart-safe worker loops

Files:

  • Modify: backend/src/redis/redis.service.ts

  • Modify: backend/src/automation/runtime/automation-runtime-worker.service.ts

  • Modify: backend/src/automation/runtime/automation-command-dispatcher.service.ts

  • Modify: backend/src/automation/runtime/automation-reconciliation.service.ts

  • Test: backend/src/automation/runtime/automation-runtime-worker.service.spec.ts

  • Step 1: Write the failing recovery tests

it('requeues a timed-out dispatched command for retry when attempts remain', async () => {
  commandIntentRepository.find.mockResolvedValue([
    {
      ...intent,
      status: 'dispatched',
      attemptCount: 1,
      maxAttempts: 3,
      nextAttemptAt: new Date('2026-05-10T10:00:05.000Z'),
    },
  ]);

  await worker.processTimedOutCommands(new Date('2026-05-10T10:00:06.000Z'));

  expect(commandIntentRepository.save).toHaveBeenCalledWith(
    expect.objectContaining({
      status: 'retrying',
      nextAttemptAt: new Date('2026-05-10T10:00:15.000Z'),
    }),
  );
});

it('expires a timed-out command once the retry budget is exhausted', async () => {
  commandIntentRepository.find.mockResolvedValue([
    {
      ...intent,
      status: 'dispatched',
      attemptCount: 3,
      maxAttempts: 3,
      nextAttemptAt: new Date('2026-05-10T10:00:45.000Z'),
    },
  ]);

  await worker.processTimedOutCommands(new Date('2026-05-10T10:00:46.000Z'));

  expect(commandIntentRepository.save).toHaveBeenCalledWith(
    expect.objectContaining({ status: 'expired' }),
  );
});
  • Step 2: Run recovery tests and verify they fail

Run from backend/: bun test -- src/automation/runtime/automation-runtime-worker.service.spec.ts

Expected: FAIL because timeout handling and Redis lease helpers are not implemented.

  • Step 3: Implement retry scheduling and worker coordination
async acquireLease(key: string, value: string, seconds: number): Promise<boolean> {
  const result = await this.client.set(key, value, { NX: true, EX: seconds });
  return result === 'OK';
}

async releaseLease(key: string, value: string): Promise<void> {
  const currentValue = await this.client.get(key);
  if (currentValue === value) {
    await this.client.del(key);
  }
}

@Cron(CronExpression.EVERY_5_SECONDS)
async processTimedOutCommands(now = new Date()): Promise<void> {
  const intents = await this.commandIntentRepository.find({
    where: { status: In([AutomationCommandIntentStatus.DISPATCHED, AutomationCommandIntentStatus.RETRYING]) },
    order: { nextAttemptAt: 'ASC' },
    take: 50,
  });

  for (const intent of intents) {
    if (!intent.nextAttemptAt || intent.nextAttemptAt > now) {
      continue;
    }

    const nextAttemptAt = computeNextAttemptAt(now, intent.attemptCount + 1);

    if (nextAttemptAt) {
      intent.status = AutomationCommandIntentStatus.RETRYING;
      intent.nextAttemptAt = nextAttemptAt;
      intent.lastError = 'Command timeout';
      await this.commandIntentRepository.save(intent);
      continue;
    }

    intent.status = AutomationCommandIntentStatus.EXPIRED;
    intent.lastError = 'Retry budget exhausted';
    await this.commandIntentRepository.save(intent);
  }
}
  • Step 4: Run tests and then lint/format

Run from backend/: bun test -- src/automation/runtime/automation-runtime-worker.service.spec.ts src/automation/runtime/automation-command-dispatcher.service.spec.ts src/automation/runtime/automation-reconciliation.service.spec.ts

Expected: PASS with retry, expiry, and restart pickup behavior covered.

Run from backend/: bun run lint:fix

Expected: PASS.

Run from backend/: bun run format

Expected: PASS.

  • Step 5: Commit
git add backend/src/redis/redis.service.ts \
  backend/src/automation/runtime/automation-runtime-worker.service.ts \
  backend/src/automation/runtime/automation-runtime-worker.service.spec.ts \
  backend/src/automation/runtime/automation-command-dispatcher.service.ts \
  backend/src/automation/runtime/automation-reconciliation.service.ts
git commit -m "feat: add automation retry and recovery workers"

Task 8: Add observability, diagnostics queries, contract tests, and end-to-end runtime coverage

Files:

  • Create: backend/src/automation/runtime/automation-runtime-diagnostics.service.ts

  • Test: backend/src/automation/runtime/automation-runtime-diagnostics.service.spec.ts

  • Test: backend/src/automation/automation-runtime.contract.spec.ts

  • Test: backend/src/automation/automation-runtime.integration.spec.ts

  • Modify: backend/src/automation/runtime/automation-runtime-worker.service.ts

  • Modify: backend/src/automation/runtime/automation-command-dispatcher.service.ts

  • Modify: backend/src/automation/runtime/automation-reconciliation.service.ts

  • Step 1: Write the failing diagnostics and integration tests

it('returns retrying and uncertain command lists for production debugging', async () => {
  commandIntentRepository.find.mockResolvedValue([
    { id: 'intent-1', status: 'retrying', targetSerialNumber: 'GW-001' },
    { id: 'intent-2', status: 'uncertain', targetSerialNumber: 'GW-002' },
  ]);

  await expect(service.listRetryingCommands()).resolves.toHaveLength(1);
  await expect(service.listUncertainCommands()).resolves.toHaveLength(1);
});

it('runs telemetry -> event -> intent -> dispatch -> ack -> applied as a single runtime flow', async () => {
  await telemetryService.save({
    deviceId: 'device-1',
    timestamp: new Date('2026-05-10T10:00:00.000Z'),
    data: { do: 3.8 },
  });

  await runtimeWorker.processPendingEndpointEvents();
  await runtimeWorker.processPendingDispatches();
  await reconciliationService.recordRpcReceipt({
    serialNumber: 'GW-001',
    commandId: 'cmd_1',
    requestId: 'req-1',
    payload: { commandId: 'cmd_1', status: 'accepted', resultCode: 'ok', message: null },
  });
  await reconciliationService.recordStateReport({
    serialNumber: 'GW-001',
    commandId: 'cmd_1',
    payload: { commandId: 'cmd_1', reportedState: { io1: true }, appliedAt: '2026-05-10T10:00:02.100Z' },
  });

  expect(await diagnosticsService.getExecutionTimeline('rule-execution-1')).toEqual(
    expect.arrayContaining([
      expect.objectContaining({ stage: 'event_persisted' }),
      expect.objectContaining({ stage: 'command_applied' }),
    ]),
  );
});
  • Step 2: Run the diagnostics and integration tests and verify they fail

Run from backend/: bun test -- src/automation/runtime/automation-runtime-diagnostics.service.spec.ts src/automation/automation-runtime.contract.spec.ts src/automation/automation-runtime.integration.spec.ts

Expected: FAIL because the diagnostics service and full runtime flow assertions are not implemented.

  • Step 3: Implement structured runtime observability and diagnostics
@Injectable()
export class AutomationRuntimeDiagnosticsService {
  constructor(
    @InjectRepository(AutomationCommandIntent)
    private commandIntentRepository: Repository<AutomationCommandIntent>,
    @InjectRepository(DeviceCommandReceipt)
    private receiptRepository: Repository<DeviceCommandReceipt>,
    @InjectRepository(AutomationRuleExecution)
    private executionRepository: Repository<AutomationRuleExecution>,
  ) {}

  async listRetryingCommands(): Promise<AutomationCommandIntent[]> {
    return this.commandIntentRepository.find({
      where: { status: AutomationCommandIntentStatus.RETRYING },
      order: { updatedAt: 'DESC' },
      take: 100,
    });
  }

  async listUncertainCommands(): Promise<AutomationCommandIntent[]> {
    return this.commandIntentRepository.find({
      where: { status: AutomationCommandIntentStatus.UNCERTAIN },
      order: { updatedAt: 'DESC' },
      take: 100,
    });
  }

  async getExecutionTimeline(ruleExecutionId: string): Promise<Array<{ stage: string; at: string }>> {
    const execution = await this.executionRepository.findOne({ where: { id: ruleExecutionId } });
    const receipts = await this.receiptRepository.find({
      where: { commandIntentId: execution?.commandIntentId ?? '' },
      order: { receivedAt: 'ASC' },
    });

    return [
      { stage: 'execution_created', at: execution?.triggeredAt.toISOString() ?? '' },
      ...receipts.map((receipt) => ({
        stage: receipt.receiptType,
        at: receipt.receivedAt.toISOString(),
      })),
      { stage: `execution_${execution?.status ?? 'unknown'}`, at: execution?.updatedAt?.toISOString() ?? '' },
    ];
  }
}

this.logger.log({
  eventId: event.id,
  ruleId: intent.ruleId,
  ruleExecutionId: intent.ruleExecutionId,
  commandIntentId: intent.id,
  deviceCommandId: intent.deviceCommandId,
  deviceId: intent.targetDeviceId,
  serialNumber: intent.targetSerialNumber,
  stage: 'dispatch_succeeded',
});
  • Step 4: Run the full automation verification suite and then lint/format

Run from backend/: bun test -- src/automation/runtime/automation-runtime-diagnostics.service.spec.ts src/automation/automation-runtime.contract.spec.ts src/automation/automation-runtime.integration.spec.ts src/automation/automation-engine.service.spec.ts src/telemetry/telemetry.service.spec.ts src/mqtt/mqtt.service.spec.ts

Expected: PASS with happy path, duplicate handling, timeout, retry, and diagnostics coverage.

Run from backend/: bun run lint:fix

Expected: PASS.

Run from backend/: bun run format

Expected: PASS.

  • Step 5: Commit
git add backend/src/automation/runtime/automation-runtime-diagnostics.service.ts \
  backend/src/automation/runtime/automation-runtime-diagnostics.service.spec.ts \
  backend/src/automation/automation-runtime.contract.spec.ts \
  backend/src/automation/automation-runtime.integration.spec.ts \
  backend/src/automation/runtime/automation-runtime-worker.service.ts \
  backend/src/automation/runtime/automation-command-dispatcher.service.ts \
  backend/src/automation/runtime/automation-reconciliation.service.ts
git commit -m "feat: add automation runtime observability coverage"

Self-Review

Spec coverage

  • Telemetry ingestion decoupling: Task 3.
  • Durable endpoint events and command ledger: Tasks 2 and 3.
  • Rule evaluation worker and idempotent command-intent creation: Task 4.
  • MQTT correlation payload, dispatcher, and serialization: Task 5.
  • Ack versus applied-state reconciliation: Task 6.
  • Retry, timeout, dedupe, and restart recovery: Task 7.
  • Structured logs, diagnostics, and reliability coverage: Task 8.
  • Deferred items remain deferred: manual control unification, scheduler UI, conflict engine, replay console, and broader distributed platform work are intentionally not planned.

Placeholder scan

  • No TODO, TBD, or “implement later” placeholders remain.
  • Every task lists concrete file paths, commands, and code snippets.

Type consistency

  • Command intent status values are consistently pending_dispatch, dispatched, acknowledged, applied, retrying, failed, expired, uncertain.
  • Execution envelope status values mirror the runtime lifecycle to keep diagnostics and final outcome mapping simple.
  • Correlation terminology is consistent across plan tasks: commandId, mqttRequestId, deviceCommandId, correlationKey.