Delaying all items except specific one

It may not work this way because you're using concatMap. As you know, it waits for the inner observable to complete before starting to process(to subscribe) the pending ones. It internally uses a buffer, such that if an inner observable is still active(did not complete), the emitted value will be added to that buffer. When the inner observable becomes inactive, the oldest value from the buffer is selected and a new inner observable will be created, based on the provided callback function.

There is also delayWhen, which emits a complete notification after all of its pending observables complete:

// called when an inner observable sends a `next`/`complete` notification
const notify = () => {
  // Notify the consumer.
  subscriber.next(value);

  // Ensure our inner subscription is cleaned up
  // as soon as possible. Once the first `next` fires,
  // we have no more use for this subscription.
  durationSubscriber?.unsubscribe();

  if (!closed) {
    active--;
    closed = true;
    checkComplete();
  }
};

checkComplete() will check if there is a need to send a complete notification to the main stream:

const checkComplete = () => isComplete && !active && subscriber.complete();

We've seen that active decreases in notify(). isComplete becomes true when the main source completes:

// this is the `complete` callback
() => {
  isComplete = true;
  checkComplete();
}

So, this is why it does not work this way:

• the PROMPT action is used to create the concatMap's first inner observable
• the observable emits 3 consecutive actions [PROMPT, BEGIN, END]
• the first 2 will get timer(0), whereas the third one, END, will get (timer(5000)); notice that in this time, before the PROMPT action got emitted, the isComplete variable is set to true, because from() completes synchronously in this case
• so there is a timer(5000) that keeps the inner obs. active; then a RESPONSE is emitted from the actions$ stream, but since there is no place for it yet, it will be added to the buffer and an inner obs. will be created when timer(5000) finally expires

A way to solve this might be to replace concatMap with mergeMap.

If I understand it right, this is a very interesting problem to address with Observables streams. This is the way I would attack it.

First I would store in a constant actionDelayed$ the result of your original logic, i.e. a stream where we have introduced, after each PROMPT, BEGIN and END actions divided by a delay.

const actionDelayed$ = action$.pipe(
  concatMap(action =>
    from(convertAction(action)).pipe(
      delayWhen(action => (action == END) ? timer(5000) : timer(0)),
    ),
  ),
);

Then I would create 2 separate streams, response$ and promptDelayed$, containing only the RESPONSE actions before the delay was introduced and the PROMPT actions after the delayed was introduced, like this

const response$ = action$.pipe(
  filter(a => a == RESPONSE)
)
const promptDelayed$ = actionDelayed$.pipe(
  filter(a => a == PROMPT)
)

With these 2 streams, I can create another stream of RESPONSE actions emitted just after the PROMPT delayed actions are emitted, like this

const responseN1AfterPromptN$ = zip(response$, promptDelayed$).pipe(
  map(([r, p]) => r)
)

At this point I have just to remove all RESPONSE actions from actionDelayed$ like this

const actionNoResponseDelayed$ = actionDelayed$.pipe(
  filter(a => a != RESPONSE)
)

and merge actionNoResponseDelayed$ with responseN1AfterPromptN$ to get the final stream.

The entirety of the code, to be tried with rxviz is this

const { interval, from, zip, timer, merge } = Rx;
const { concatMap, delayWhen, share, filter, map } = RxOperators;

const PROMPT = 'P';
const RESPONSE = 'R';
const EFFECT = 'E';

const BEGIN = '^';
const END = '&';

const convertAction = action => (action === PROMPT) ? [PROMPT, BEGIN, END] : [action];

// Just actions coming at regular intervals
const action$ = zip(
  from([PROMPT, RESPONSE, EFFECT, PROMPT, RESPONSE, EFFECT, EFFECT, EFFECT]),
    interval(1000),
  (a, b) => a,
).pipe(share());

const actionDelayed$ = action$.pipe(
  concatMap(action =>
    from(convertAction(action)).pipe(
      delayWhen(action => (action == END) ? timer(5000) : timer(0)),
    ),
  ),
  share()
);

const response$ = action$.pipe(
  filter(a => a == RESPONSE)
)
const promptDelayed$ = actionDelayed$.pipe(
  filter(a => a == PROMPT)
)
const responseN1AfterPromptN$ = zip(response$, promptDelayed$).pipe(
  map(([r, p]) => r)
)
const actionNoResponseDelayed$ = actionDelayed$.pipe(
  filter(a => a != RESPONSE)
)

merge(actionNoResponseDelayed$, responseN1AfterPromptN$)

The use of the share operator while creating action$ and actionDelayed$ streams allows to avoid repeated subscriptions to these streams when creating the subsequent streams used in the solution.