fix: settle traceScenario as success when the consumer unsubscribes after a value

projects/ngx-ui-watchtower/DESIGN.md

@@ -614,8 +614,11 @@ all along — use the named method instead.
 For Observable-driven journeys, the pipeable `traceScenario` operator
 completes or fails the scenario based on how the stream ends. If it is torn
 down some other way first — a superseding `switchMap`, `takeUntilDestroyed()`,
-a manual unsubscribe — it cancels the scenario instead of leaving it to
-settle as a `timeout`.
+a manual unsubscribe — what it records depends on whether a value got
+through. Before any value, it cancels the scenario instead of leaving it to
+settle as a `timeout`. After a value, it completes the scenario, because that
+is how `firstValueFrom`, `take(1)` and `first()` end a stream once they have
+what they need.
 
 ```ts
 import { traceScenario } from '@absaoss-cps/ngx-ui-watchtower';
@@ -650,13 +653,36 @@ this.searchSubject$.pipe(
 ```
 
 How it decides: `traceScenario` settles on `complete` with optional derived
-outcome metadata, and fails with the caught error on `error`. A teardown that reaches neither — a superseding `switchMap`, `takeUntilDestroyed()`,
-a manual unsubscribe — cancels the scenario instead, guarded by `isSettled`
-so it never re-cancels one that already completed or failed (`tap`'s own
-`unsubscribe` hook fires after every teardown, settled or not). Without
-this, a cancelled-by-unsubscription scenario would sit active until its own
-timeout and record as `timeout` rather than the caller-driven abandonment
-it actually was.
+outcome metadata, and fails with the caught error on `error`. A teardown that
+reaches neither — a superseding `switchMap`, `takeUntilDestroyed()`, a manual
+unsubscribe — settles it by whether the stream has emitted:
+
+- **No value yet:** cancels the scenario. Without this, a
+  cancelled-by-unsubscription scenario would sit active until its own timeout
+  and record as `timeout` rather than the caller-driven abandonment it
+  actually was.
+- **After a value:** completes the scenario, with the outcome mapped from the
+  last value. `firstValueFrom`, `take(1)` and `first()` unsubscribe upstream
+  as soon as they have a value, before the source's `complete` arrives, so
+  without this a successful request would record as `abandoned`.
+
+Both are guarded by `isSettled`, so a teardown never re-settles a scenario
+that already completed or failed (`tap`'s own `unsubscribe` hook fires after
+every teardown, settled or not).
+
+For a multi-value stream — polling, progress, a websocket — stopping after a
+value usually means the user gave up, not that the journey succeeded. Opt
+back into cancelling with `valueThenTeardown: 'cancel'`:
+
+```ts
+this.api
+  .pollJobStatus(jobId)
+  .pipe(
+    traceScenario(scenario, { valueThenTeardown: 'cancel' }),
+    takeUntilDestroyed(this.destroyRef)
+  )
+  .subscribe();
+```
 
 ### Measuring what the user waited for
 

projects/ngx-ui-watchtower/src/lib/scenario/uwt-scenario-operators/uwt-scenario-operators.spec.ts

@@ -1,4 +1,21 @@
-import { defer, EMPTY, of, Subject, switchMap, throwError } from 'rxjs';
+import {
+  defer,
+  EMPTY,
+  first,
+  type Observable,
+  firstValueFrom,
+  lastValueFrom,
+  of,
+  retry,
+  Subject,
+  switchMap,
+  take,
+  throwError
+} from 'rxjs';
+import { UWT_DEFAULT_TELEMETRY_CONFIG } from '../../config/uwt-telemetry-common.config/uwt-telemetry-common.config';
+import { UwtScenarioRecord } from '../../models/uwt-scenario.models/uwt-scenario.models';
+import { UwtNoopTelemetrySink } from '../../sinks/uwt-telemetry/uwt-noop-telemetry.sink/uwt-noop-telemetry.sink';
+import { UWT_DEFAULT_REDACT_CONFIG } from '../../utils/uwt-telemetry-redact.util/uwt-telemetry-redact.util';
 import { UwtScenario } from '../uwt-scenario/uwt-scenario';
 import { traceScenario } from './uwt-scenario-operators';
 
@@ -292,4 +309,202 @@ describe('traceScenario operator', () => {
         });
     });
   });
+
+  describe('with a real scenario', () => {
+    let settled: UwtScenarioRecord[];
+
+    const countRows = (rows: string[]) => ({
+      metadata: { count: rows.length }
+    });
+
+    function createScenario(name = 'load-customers'): UwtScenario {
+      return new UwtScenario(
+        { name },
+        {
+          identity: {
+            application: 'test-app',
+            environment: 'test',
+            version: '1.0.0'
+          },
+          scenarioTelemetryConfig: {
+            ...UWT_DEFAULT_TELEMETRY_CONFIG.scenario
+          },
+          redact: UWT_DEFAULT_REDACT_CONFIG,
+          sink: new UwtNoopTelemetrySink(),
+          onSettled: (_id, record) => settled.push(record)
+        }
+      );
+    }
+
+    beforeEach(() => {
+      settled = [];
+    });
+
+    describe('single-value sources', () => {
+      it('should settle as success with subscribe()', () => {
+        const real = createScenario();
+
+        of(['a', 'b']).pipe(traceScenario(real, countRows)).subscribe();
+
+        expect(real.status).toBe('success');
+        expect(settled[0].metadata).toEqual({ count: 2 });
+      });
+
+      it('should settle as success with lastValueFrom', async () => {
+        const real = createScenario();
+
+        await lastValueFrom(
+          of(['a', 'b']).pipe(traceScenario(real, countRows))
+        );
+
+        expect(real.status).toBe('success');
+        expect(settled[0].metadata).toEqual({ count: 2 });
+      });
+
+      it('should settle as success with firstValueFrom', async () => {
+        const real = createScenario();
+
+        const rows = await firstValueFrom(
+          of(['a', 'b']).pipe(traceScenario(real, countRows))
+        );
+
+        expect(rows).toEqual(['a', 'b']);
+        expect(real.status).toBe('success');
+        expect(settled[0].metadata).toEqual({ count: 2 });
+      });
+
+      it('should settle as success with take(1) after the operator', () => {
+        const real = createScenario();
+
+        of(['a', 'b'])
+          .pipe(traceScenario(real, countRows), take(1))
+          .subscribe();
+
+        expect(real.status).toBe('success');
+        expect(settled[0].metadata).toEqual({ count: 2 });
+      });
+
+      it('should settle as success with first() after the operator', () => {
+        const real = createScenario();
+
+        of(['a', 'b'])
+          .pipe(traceScenario(real, countRows), first())
+          .subscribe();
+
+        expect(real.status).toBe('success');
+        expect(settled[0].metadata).toEqual({ count: 2 });
+      });
+
+      it('should settle as success with take(1) before the operator', () => {
+        const real = createScenario();
+
+        of(['a', 'b'])
+          .pipe(take(1), traceScenario(real, countRows))
+          .subscribe();
+
+        expect(real.status).toBe('success');
+        expect(settled[0].metadata).toEqual({ count: 2 });
+      });
+
+      it('should settle as success with take(1) before the operator under firstValueFrom', async () => {
+        const real = createScenario();
+
+        await firstValueFrom(
+          of(['a', 'b']).pipe(take(1), traceScenario(real, countRows))
+        );
+
+        expect(real.status).toBe('success');
+        expect(settled[0].metadata).toEqual({ count: 2 });
+      });
+    });
+
+    describe('switchMap superseding an inner stream', () => {
+      it('should abandon a stream superseded before it emitted', () => {
+        const superseded = createScenario('superseded');
+        const latest = createScenario('latest');
+        const pendingFirstRequest = new Subject<string[]>();
+        const trigger = new Subject<number>();
+
+        trigger
+          .pipe(
+            switchMap((n) =>
+              n === 1
+                ? pendingFirstRequest.pipe(traceScenario(superseded))
+                : of(['b']).pipe(traceScenario(latest))
+            )
+          )
+          .subscribe();
+
+        trigger.next(1);
+        trigger.next(2);
+
+        expect(superseded.status).toBe('abandoned');
+        expect(latest.status).toBe('success');
+      });
+
+      it('should settle as success a stream superseded after it emitted', () => {
+        const superseded = createScenario('superseded');
+        const latest = createScenario('latest');
+        const firstRequest = new Subject<string[]>();
+        const trigger = new Subject<number>();
+
+        trigger
+          .pipe(
+            switchMap((n) =>
+              n === 1
+                ? firstRequest.pipe(traceScenario(superseded, countRows))
+                : of(['b']).pipe(traceScenario(latest))
+            )
+          )
+          .subscribe();
+
+        trigger.next(1);
+        firstRequest.next(['a']); // emitted, but never completed
+        trigger.next(2);
+
+        expect(superseded.status).toBe('success');
+        expect(superseded.toRecord().metadata).toEqual({ count: 1 });
+        expect(latest.status).toBe('success');
+      });
+    });
+
+    it("should abandon a stream torn down after a value with valueThenTeardown: 'cancel'", () => {
+      const real = createScenario('poll-status');
+
+      of(['a'], ['b'])
+        .pipe(traceScenario(real, { valueThenTeardown: 'cancel' }), take(1))
+        .subscribe();
+
+      expect(real.status).toBe('abandoned');
+    });
+
+    describe('retry placement', () => {
+      function failOnce(): Observable<string[]> {
+        let attempts = 0;
+        return defer(() =>
+          attempts++ === 0
+            ? throwError(() => new Error('transient'))
+            : of(['a'])
+        );
+      }
+
+      it('should settle as success when retry() comes before the operator', () => {
+        const real = createScenario();
+
+        failOnce().pipe(retry(1), traceScenario(real)).subscribe();
+
+        expect(real.status).toBe('success');
+      });
+
+      it('should settle as failure when retry() comes after the operator, even if the retry succeeds', () => {
+        const real = createScenario();
+
+        failOnce().pipe(traceScenario(real), retry(1)).subscribe();
+
+        // Documented limitation: the first error settles the scenario and
+        // the operator cannot see the downstream retry.
+        expect(real.status).toBe('failure');
+      });
+    });
+  });
 });

projects/ngx-ui-watchtower/src/lib/scenario/uwt-scenario-operators/uwt-scenario-operators.ts

@@ -22,6 +22,14 @@
    * erroring — e.g. a `switchMap` moving on to a newer value.
    */
   cancelOutcome?: UwtScenarioOutcome | (() => UwtScenarioOutcome | void);
+
+  /**
+   * What a downstream unsubscribe means once a value has been delivered.
+   * Default `'complete'`, so `firstValueFrom`, `take(1)` and `first()` settle
+   * as success. Use `'cancel'` for multi-value streams (polling, progress),
+   * where stopping after a value is abandonment.
+   */
+  valueThenTeardown?: 'complete' | 'cancel';
 }
 
 /**
@@ -54,45 +62,55 @@
     typeof options === 'function' ? options : options?.outcome;
   const cancelOutcome =
     typeof options === 'function' ? undefined : options?.cancelOutcome;
+  const valueThenTeardown =
+    (typeof options === 'function' ? undefined : options?.valueThenTeardown) ??
+    'complete';
 
   return (source: Observable<T>): Observable<T> =>
     defer(() => {
       let lastValue: T | undefined;
       let hasValue = false;
 
+      const complete = () => {
+        const outcome =
+          hasValue && outcomeMapper
+            ? uwtSafe(
+                'scenario.traceOutcome',
+                () => outcomeMapper(lastValue as T),
+                undefined
+              )
+            : undefined;
+        scenario.complete(outcome || undefined);
+      };
+
+      const unsubscribe = () => {
+        if (scenario.isSettled) {
+          return;
+        }
+        // The consumer got a value and stopped listening, as
+        // firstValueFrom, take(1) and first() do: treat it as completion.
+        if (hasValue && valueThenTeardown === 'complete') {
+          complete();
+          return;
+        }
+        const resolved =
+          typeof cancelOutcome === 'function'
+            ? uwtSafe('scenario.traceCancelOutcome', cancelOutcome, undefined)
+            : cancelOutcome;
+        scenario.cancel(resolved || undefined);
+      };
+
       return source.pipe(
         tap({
           next: (value) => {
             lastValue = value;
             hasValue = true;
           },
-          complete: () => {
-            const outcome =
-              hasValue && outcomeMapper
-                ? uwtSafe(
-                    'scenario.traceOutcome',
-                    () => outcomeMapper(lastValue as T),
-                    undefined
-                  )
-                : undefined;
-            scenario.complete(outcome || undefined);
-          },
+          complete,
           error: (error: unknown) => {
             scenario.fail({ error });
           },
-          unsubscribe: () => {
-            if (!scenario.isSettled) {
-              const resolved =
-                typeof cancelOutcome === 'function'
-                  ? uwtSafe(
-                      'scenario.traceCancelOutcome',
-                      cancelOutcome,
-                      undefined
-                    )
-                  : cancelOutcome;
-              scenario.cancel(resolved || undefined);
-            }
-          }
+          unsubscribe
         })
       );
     });