The engine channel

The gRPC bidi stream, the ack cursor, and how a run survives its engine being replaced mid-scan.

One long-lived bidirectional gRPC stream per run, authenticated by the runToken from the manifest. Every up-message carries a monotonic seq, so a stream that drops can re-dial and replay from the last acked cursor with delivery idempotent by (runId, seq).

Source: verifi-harness-sdk/src/grpc-channel.ts; proto in @pragyacyber/engine-contract/proto.

The interface

You only implement this if you are building a test double. In production you use GrpcEngineChannel.

export interface EngineChannel {
  handshake(connectorVersion: string, harnessName: string, harnessVersion: string): Promise<void>;
  sendEvent(e: { type: string; severity: string; payload: Record<string, unknown>; ts: string }): void;
  sendFinding(f: Record<string, unknown>): void;
  sendDone(outcome: { status: string; findings: number; coverage: string[]; exitReason: string }): Promise<void>;
  onStop(cb: (reason: string) => void): void;
  close(): Promise<void>;
}

The wire protocol

service EngineChannel {
  rpc Open(stream UpMessage) returns (stream DownMessage);
}

message UpMessage {
  string run_id = 1;
  string correlation_id = 2;
  uint64 seq = 3;               // monotonic per run; delivery idempotent by (run_id, seq)
  oneof body {
    Handshake       handshake        = 4;
    Event           event            = 5;
    FindingDraft    finding          = 6;
    Heartbeat       heartbeat        = 7;
    ApprovalRequest approval_request = 8;
    Done            done             = 9;
  }
}

message DownMessage {
  oneof body {
    Stop             stop              = 1;   // 'deadline' | 'operator' | 'failure'
    ApprovalDecision approval_decision = 2;
    BudgetUpdate     budget_update     = 3;   // ADVISORY — cannot halt a run
    Ack              ack               = 4;
  }
}

Note what is not there: Stop has no cost reason, and BudgetUpdate is explicitly advisory. Time is the only hard bound.

The proto ships inside the contract package, so client and server cannot diverge:

export function protoPath(): string;         // resolves '@pragyacyber/engine-contract/proto'
export function loadEngineProto(): grpc.GrpcObject;

Constructing the channel

export interface GrpcChannelConfig {
  grpcAddr: string;
  runId: string;
  correlationId: string;
  runToken: string;
  /** Below the ALB idle timeout (default 60s) so the stream never idles out. */
  keepaliveMs?: number;
  /**
   * The STABLE engine HTTP base (the internal ALB). If the gRPC stream drops the
   * channel re-discovers a live engine by GETting `${httpAddr}/grpc-address`, re-dials
   * it, and replays from the last acked cursor. Absent → the channel re-dials the SAME
   * address (only useful for a transient blip, not a task replacement).
   */
  httpAddr?: string;
  fetchImpl?: typeof fetch;
}

const channel = new GrpcEngineChannel({
  grpcAddr,
  runId: manifest.runId,
  correlationId: manifest.correlationId,
  runToken,
  keepaliveMs: 30_000,
  ...(httpAddr ? { httpAddr } : {}),
});

The runToken is carried as authorization: Bearer … gRPC metadata on every dial, because a replacement engine authorizes the reconnecting harness from that token's hash. It is single-run, expires at hardDeadline, and is never logged.

Reconnect and replay

send(msg)
  ├─ seq += 1
  ├─ push to #unacked          ← ALWAYS, before the write
  └─ try call.write(msg)       ← may throw mid-reconnect; that is fine

on ack(seq)
  ├─ lastAcked = max(lastAcked, seq)
  └─ drop everything at or below it from #unacked

on error | end  (and not deliberately closed)
  └─ reconnect loop, 500ms → ×2 → 15s cap, plus up to 250ms jitter
       ├─ GET {httpAddr}/grpc-address   → the live engine's address
       ├─ dial: fresh client + call
       ├─ re-send the handshake at seq 0  (a CONTROL message, written directly)
       └─ replayUnacked()

Three details worth knowing:

Buffer first, always. If the call is mid-reconnect the write throws or no-ops, but the message is retained and replayed on the next dial. Losing it is the one thing the ack cursor exists to prevent.
The handshake is seq 0. It is re-sent on every dial and written directly rather than buffered, so its ack never advances the cursor past unacked events.
The loop is bounded only by close(). The run's hardDeadline is the real ceiling, and until then a scan that is still producing findings should keep trying to deliver them.

Observable state, useful in tests:

channel.lastAckedSeq   // number
channel.unackedCount   // number
channel.replayUnacked() // returns the number of messages resent

This is verified behaviour, not aspiration: an engine was killed mid-run on the dev environment and the run's finding count continued 3 → 9 → 35 past the kill.

Keepalive

A heartbeat every keepaliveMs (default 25 000 ms), plus gRPC-level keepalive options:

'grpc.keepalive_time_ms': keepaliveMs ?? 25_000,
'grpc.keepalive_timeout_ms': 10_000,
'grpc.keepalive_permit_without_calls': 1,

Keep it below your load balancer's idle timeout — 60s on a default ALB.

Shutdown ordering

sendDone() sends the outcome and then waits for its ack, up to 5 seconds:

Without the wait, close() in runHarness's finally block tears the connection down while writes are still buffered, and the last messages of every run are silently lost. A run's own outcome is the worst possible thing to drop, so the shutdown is ordered rather than optimistic.

close() marks the channel closed first, so the error/end handlers on the closing stream do not mistake a deliberate teardown for a drop and start a reconnect loop that never ends. Then it half-closes, waits up to 2s for end/status/error, and tears down.

The interface

The wire protocol

Constructing the channel

Reconnect and replay

Keepalive

Shutdown ordering

Deploying the harness and the engine