feat(ft8): add multistream WW Digi contest support

Author: boybook

packages/builtin-plugins/src/_shared/parallel-qso/ParallelQSOCoordinator.test.ts

@@ -0,0 +1,71 @@
+import type {
+  ParsedFT8Message,
+  QSOFailureInfo,
+  StrategyDecisionMetaV2,
+  StrategyQSOCompletionEffect,
+  StrategyQSOCompletionSettlement,
+  StrategyStreamSnapshot,
+  StreamPhysicalReceipt,
+  QueuedStrategyObservationMeta,
+} from '@tx5dr/plugin-api';
+
+export interface ParallelQSOQueueEntry<TData> {
+  entryId: string;
+  targetKey: string;
+  callsign: string;
+  requestedTransmitCycle?: 0 | 1;
+  data: TData;
+}
+
+export interface ProtocolLaneIdentity {
+  streamId: string;
+  laneIndex: number;
+}
+
+export interface ProtocolLaneActivation {
+  accepted: boolean;
+}
+
+export interface ProtocolLaneRelease {
+  disposition: 'remove-entry' | 'retain-entry';
+  reason: string;
+}
+
+export interface ProtocolLaneDecision<TData = unknown> {
+  qsoCompletion?: StrategyQSOCompletionEffect;
+  qsoFailure?: QSOFailureInfo;
+  release?: ProtocolLaneRelease;
+  entryData?: TData;
+  queueChanged?: boolean;
+}
+
+export type ProtocolLaneSnapshot = Omit<
+  StrategyStreamSnapshot,
+  'streamId' | 'audioFrequencyHz'
+>;
+
+/** One stable RF stream that can host successive protocol sessions. */
+export interface ProtocolLane<TData> {
+  readonly streamId: string;
+  readonly audioFrequencyHz: number;
+
+  activate(entry: Readonly<ParallelQSOQueueEntry<TData>>): ProtocolLaneActivation;
+  deactivate(reason: string): void;
+  hasPendingWork(): boolean;
+  shouldObserve?(): boolean;
+
+  observe?(messages: ParsedFT8Message[], meta: QueuedStrategyObservationMeta): boolean;
+  decide(
+    messages: ParsedFT8Message[],
+    meta: StrategyDecisionMetaV2,
+  ): ProtocolLaneDecision<TData> | Promise<ProtocolLaneDecision<TData>>;
+
+  getTransmitText(): string | null;
+  getSnapshot(): ProtocolLaneSnapshot | null;
+
+  checkpoint(): unknown;
+  restore(checkpoint: unknown): void;
+  onPhysicalSuccess?(receipt: StreamPhysicalReceipt): void;
+  settleQSOCompletion?(settlement: StrategyQSOCompletionSettlement): boolean | void;
+  reset(reason?: string): void;
+}

packages/builtin-plugins/src/_shared/parallel-qso/ParallelQSOCoordinator.ts

@@ -0,0 +1,2 @@
+export * from './ProtocolLane.js';
+export * from './ParallelQSOCoordinator.js';

packages/builtin-plugins/src/_shared/parallel-qso/ProtocolLane.ts

@@ -60,16 +60,20 @@ function createLogger(): PluginLogger {
 function createRuntime(options: {
   transmitting?: boolean;
   operator?: StandardQSOPluginOperator;
+  maxStreams?: number;
 } = {}) {
   let transmitting = options.transmitting ?? false;
+  let maxStreams = options.maxStreams ?? 1;
   const runtimeOptions: AssistedQSOQueueRuntimeOptions = {
     operator: options.operator ?? createOperator(),
     isTransmitting: () => transmitting,
     logger: createLogger(),
+    getMaxStreams: () => maxStreams,
   };
   return {
     runtime: new AssistedQSOQueueRuntime(runtimeOptions),
     setTransmitting(value: boolean) { transmitting = value; },
+    setMaxStreams(value: number) { maxStreams = value; },
     logger: runtimeOptions.logger,
   };
 }
@@ -181,6 +185,112 @@ describe('AssistedQSOQueueRuntime queue capability', () => {
     expect(runtime.getQueueSnapshot().version).toBe(activeVersion);
   });
 
+  it('runs two or three stable protocol lanes when configured while defaulting to one', async () => {
+    const defaultRuntime = createRuntime({ transmitting: true }).runtime;
+    defaultRuntime.enqueueTarget({ callsign: 'JA1AAA' });
+    defaultRuntime.enqueueTarget({ callsign: 'JA2BBB' });
+    const defaultDecision = await defaultRuntime.decide([], decision());
+    expect(defaultDecision.transmissions).toHaveLength(1);
+    expect(defaultRuntime.getQueueSnapshot()).toMatchObject({
+      maxActiveStreams: 1,
+      activeEntryIds: [defaultRuntime.getQueueSnapshot().rows[0]?.entryId],
+    });
+
+    const parallel = createRuntime({ transmitting: true, maxStreams: 3 }).runtime;
+    parallel.enqueueTarget({ callsign: 'JA1AAA' });
+    parallel.enqueueTarget({ callsign: 'JA2BBB' });
+    parallel.enqueueTarget({ callsign: 'JA3CCC' });
+    const parallelDecision = await parallel.decide([], decision());
+
+    expect(parallelDecision.transmissions).toMatchObject([
+      { streamId: 'stream-1', text: expect.stringContaining('JA1AAA') },
+      { streamId: 'stream-2', text: expect.stringContaining('JA2BBB') },
+      { streamId: 'stream-3', text: expect.stringContaining('JA3CCC') },
+    ]);
+    expect(parallelDecision.transmissions!.map((item) => item.audioFrequencyHz)).toEqual([
+      1_500,
+      1_560,
+      1_620,
+    ]);
+    expect(parallelDecision.snapshot.streams).toHaveLength(3);
+    expect(parallel.getQueueSnapshot().activeEntryIds).toHaveLength(3);
+  });
+
+  it('applies non-preemptive dynamic shrink and uses a newly raised limit on the next decision', async () => {
+    const { runtime, setMaxStreams } = createRuntime({ transmitting: true, maxStreams: 2 });
+    runtime.enqueueTarget({ callsign: 'JA1AAA' });
+    runtime.enqueueTarget({ callsign: 'JA2BBB' });
+    runtime.enqueueTarget({ callsign: 'JA3CCC' });
+    await runtime.decide([], decision());
+    expect(runtime.getQueueSnapshot().activeEntryIds).toHaveLength(2);
+
+    setMaxStreams(3);
+    await runtime.decide([], decision(2));
+    expect(runtime.getQueueSnapshot()).toMatchObject({ maxActiveStreams: 3 });
+    expect(runtime.getQueueSnapshot().activeEntryIds).toHaveLength(3);
+
+    setMaxStreams(1);
+    await runtime.decide([], decision(3));
+    expect(runtime.getQueueSnapshot()).toMatchObject({ maxActiveStreams: 1 });
+    expect(runtime.getQueueSnapshot().activeEntryIds).toHaveLength(3);
+  });
+
+  it('routes decoded replies to one standard protocol lane without advancing its peers', async () => {
+    const { runtime } = createRuntime({ transmitting: true, maxStreams: 3 });
+    runtime.enqueueTarget({ callsign: 'JA1AAA', lastMessage: selected('CQ JA1AAA PM95') });
+    runtime.enqueueTarget({ callsign: 'JA2BBB', lastMessage: selected('CQ JA2BBB PM95') });
+    runtime.enqueueTarget({ callsign: 'JA3CCC', lastMessage: selected('CQ JA3CCC PM95') });
+    const initial = await runtime.decide([], decision());
+    runtime.onTransmissionsCompleted(initial.transmissions!.map((transmission) => ({
+      ...transmission,
+      frameId: 'frame-1',
+      revision: 1,
+      physicalConfirmed: true as const,
+    })));
+
+    const replyAt = BASE_TIME + MODES.FT8.slotMs;
+    const reply = parsed('BG5DRB JA1AAA -05', replyAt, { snr: -7 });
+    runtime.observeDecodedMessages([reply], observation(replyAt));
+    const advanced = await runtime.decide([reply], decision(2));
+
+    expect(advanced.transmissions).toMatchObject([
+      { streamId: 'stream-1', text: expect.stringContaining('R-07') },
+      { streamId: 'stream-2', text: expect.stringContaining('JA2BBB') },
+      { streamId: 'stream-3', text: expect.stringContaining('JA3CCC') },
+    ]);
+    expect(advanced.snapshot.streams).toMatchObject([
+      { streamId: 'stream-1', currentState: 'TX3', targetCallsign: 'JA1AAA' },
+      { streamId: 'stream-2', currentState: 'TX1', targetCallsign: 'JA2BBB' },
+      { streamId: 'stream-3', currentState: 'TX1', targetCallsign: 'JA3CCC' },
+    ]);
+  });
+
+  it('hot-updates protocol and physical lane frequencies through one bounded resolver', async () => {
+    const source = createOperator();
+    const config = { ...source.config, frequency: 299 };
+    const operator: StandardQSOPluginOperator = {
+      get config() { return config; },
+      hasWorkedCallsign: (callsign, options) => source.hasWorkedCallsign(callsign, options),
+      isTargetBeingWorkedByOthers: (callsign) => source.isTargetBeingWorkedByOthers(callsign),
+    };
+    const { runtime } = createRuntime({ transmitting: true, maxStreams: 3, operator });
+    runtime.enqueueTarget({ callsign: 'JA1AAA' });
+    runtime.enqueueTarget({ callsign: 'JA2BBB' });
+    runtime.enqueueTarget({ callsign: 'JA3CCC' });
+    await runtime.decide([], decision());
+    expect(runtime.getTransmissions().map((item) => item.audioFrequencyHz)).toEqual([1_500, 1_560, 1_620]);
+
+    config.frequency = 4_700;
+    expect(runtime.getTransmissions().map((item) => item.audioFrequencyHz)).toEqual([4_700, 4_760, 4_820]);
+    expect(runtime.getSnapshot().context?.actualFrequency).toBeUndefined();
+
+    config.frequency = 4_701;
+    expect(runtime.getTransmissions().map((item) => item.audioFrequencyHz)).toEqual([1_500, 1_560, 1_620]);
+
+    config.mode = MODES.FT4;
+    expect(runtime.getTransmissions().map((item) => item.audioFrequencyHz)).toEqual([1_500, 1_600, 1_700]);
+  });
+
   it('uses standard TX6 CQ only when no queue target is executable', async () => {
     const { runtime, setTransmitting } = createRuntime();
     expect(runtime.getTransmitText()).toBeNull();
@@ -454,6 +564,35 @@ describe('AssistedQSOQueueRuntime queue capability', () => {
       .toEqual(['JA1AAA', 'JA2BBB', 'JA3CCC']);
   });
 
+  it('gives a preempted lane to the direct opportunity that triggered preemption', async () => {
+    const { runtime } = createRuntime({ transmitting: true, maxStreams: 3 });
+    runtime.enqueueTarget({ callsign: 'JA1AAA' });
+    runtime.enqueueTarget({ callsign: 'JA2BBB' });
+    runtime.enqueueTarget({ callsign: 'JA3CCC' });
+    await runtime.decide([], decision());
+    runtime.enqueueTarget({ callsign: 'JA4DDD' });
+    const directAt = BASE_TIME;
+    runtime.observeDecodedMessages([
+      parsed('BG5DRB JA5EEE PM95', directAt),
+    ], observation(directAt));
+    const promoted = runtime.getQueueSnapshot();
+    const direct = promoted.rows.find((row) => row.callsign === 'JA5EEE')!;
+    expect(runtime.reorderTarget(direct.entryId, null, promoted.version).snapshot.rows.map((row) => row.callsign))
+      .toEqual(['JA1AAA', 'JA2BBB', 'JA3CCC', 'JA4DDD', 'JA5EEE']);
+
+    const preempted = await runtime.decide([], decision(2));
+
+    expect(preempted.transmissions).toMatchObject([
+      { streamId: 'stream-1', text: expect.stringContaining('JA5EEE') },
+      { streamId: 'stream-2', text: expect.stringContaining('JA2BBB') },
+      { streamId: 'stream-3', text: expect.stringContaining('JA3CCC') },
+    ]);
+    expect(runtime.getQueueSnapshot().rows.slice(0, 3).map((row) => row.callsign))
+      .toEqual(['JA5EEE', 'JA2BBB', 'JA3CCC']);
+    expect(runtime.getQueueSnapshot().rows.slice(3).map((row) => row.callsign))
+      .toEqual(['JA4DDD', 'JA1AAA']);
+  });
+
   it('releases an unengaged active target when it starts working another station', async () => {
     const { runtime } = createRuntime({ transmitting: true });
     runtime.enqueueTarget({ callsign: 'JA1AAA', lastMessage: selected('CQ JA1AAA PM95') });
@@ -687,6 +826,11 @@ describe('AssistedQSOQueueRuntime queue capability', () => {
       currentState: 'TX5',
       context: { targetCallsign: 'JA1AAA' },
     });
+    expect(retry.snapshot.streams).toMatchObject([{
+      streamId: 'stream-1',
+      currentState: 'TX1',
+      targetCallsign: 'JA2BBB',
+    }]);
     expect(retry.qsoCompletion).toBeUndefined();
     expect(runtime.getQueueSnapshot()).toMatchObject({
       activeEntryId: nextEntryId,
@@ -701,6 +845,38 @@ describe('AssistedQSOQueueRuntime queue capability', () => {
     expect(runtime.getQueueSnapshot().activeEntryId).toBe(nextEntryId);
   });
 
+  it('clears an unscheduled final-73 lease after the completed row is gone', async () => {
+    const { runtime } = createRuntime({ transmitting: true });
+    runtime.enqueueTarget({
+      callsign: 'JA1AAA',
+      lastMessage: selected('BG5DRB JA1AAA -08'),
+    });
+    const first = await runtime.decide([], decision());
+    runtime.onTransmissionQueued(first.transmission!);
+    const rrr = parsed('BG5DRB JA1AAA RR73', BASE_TIME + MODES.FT8.slotMs);
+    runtime.observeDecodedMessages([rrr], observation(rrr.timestamp));
+    const completion = await runtime.decide([rrr], decision(2));
+    runtime.settleQSOCompletion({
+      lifecycleEpoch: completion.qsoCompletion!.lifecycleEpoch,
+      recordId: completion.qsoCompletion!.record.id,
+      status: 'committed',
+    });
+    runtime.onTransmissionQueued(completion.transmission!);
+    const releaseAt = BASE_TIME + MODES.FT8.slotMs * 2;
+    runtime.observeDecodedMessages([], observation(releaseAt));
+    await runtime.decide([], decision(3));
+    expect(runtime.getQueueSnapshot().rows).toHaveLength(0);
+
+    const beforeClear = runtime.getQueueSnapshot();
+    const cleared = runtime.clearTargets(beforeClear.version);
+    expect(cleared.snapshot.version).toBe(beforeClear.version + 1);
+    const repeated = parsed('BG5DRB JA1AAA RR73', releaseAt);
+    runtime.observeDecodedMessages([repeated], observation(releaseAt));
+    const afterClear = await runtime.decide([repeated], decision(4));
+    expect(afterClear.transmission).toBe('CQ BG5DRB OL32');
+    expect(afterClear.transmission).not.toBe('JA1AAA BG5DRB 73');
+  });
+
   it('does not let a completed target retry preempt an engaged next QSO', async () => {
     const { runtime } = createRuntime({ transmitting: true });
     const { releaseSlotStartMs } = await prepareFinal73Lease(runtime);
@@ -720,7 +896,7 @@ describe('AssistedQSOQueueRuntime queue capability', () => {
     });
   });
 
-  it('holds a delegate-released completion until asynchronous log settlement arrives', async () => {
+  it('releases a delegate completion when durable persistence merged to a different record id', async () => {
     const { runtime } = createRuntime({ transmitting: true });
     runtime.enqueueTarget({
       callsign: 'JA1AAA',
@@ -742,6 +918,7 @@ describe('AssistedQSOQueueRuntime queue capability', () => {
     runtime.settleQSOCompletion({
       lifecycleEpoch: completion.qsoCompletion!.lifecycleEpoch,
       recordId: completion.qsoCompletion!.record.id,
+      persistedRecordId: 'existing-merged-qso',
       status: 'committed',
     });
     expect(runtime.getQueueSnapshot()).toMatchObject({