fix(chat-flow): serialize per-chat processing, end flow on dead-leaf nodes (v1.0.3)

- processMessage now serializes per (session, chat) through a self-evicting promise
  chain, closing a read->write race where two near-simultaneous messages could lose or
  duplicate navigation (double greeting, resurrected leaf). The bounded invalid-path
  re-process recurses on the locked body to avoid self-deadlock.
- When a config edit parks an in-flight user on a node that no longer has options, the
  flow ends cleanly instead of looping "Invalid option" until the 15-minute expiry.

Author: rmyndharis

README.md

@@ -35,7 +35,7 @@ This repository provides:
 | Plugin | Description | Version | Status |
 | ------ | ----------- | ------- | ------ |
 | [`after-hours`](./after-hours) | Auto-replies with a configurable away/closing message to messages received outside business hours. | 0.1.1 | stable |
-| [`chat-flow`](./chat-flow) | Interactive, stateful auto-reply: a trigger word starts a greeting + numbered menu, replies traverse a configurable menu tree, and per-chat state expires after 15 minutes. | 1.0.2 | stable |
+| [`chat-flow`](./chat-flow) | Interactive, stateful auto-reply: a trigger word starts a greeting + numbered menu, replies traverse a configurable menu tree, and per-chat state expires after 15 minutes. | 1.0.3 | stable |
 | [`faq-bot`](./faq-bot) | Auto-replies to inbound WhatsApp messages from configurable FAQ keyword/regex rules. | 0.1.1 | stable |
 | [`group-translate`](./group-translate) | Auto-translates group messages between participants' languages via a LibreTranslate backend. Configure in-chat with /tr commands. Admin-gated; disabled until enabled. | 1.0.2 | stable |
 | [`gsheets-logger`](./gsheets-logger) | Logs WhatsApp message events to a Google Sheet via a service account. | 0.2.1 | stable |

chat-flow/CHANGELOG.md

@@ -8,6 +8,17 @@ The version here always matches `manifest.json`'s `version`.
 
 ## [Unreleased]
 
+## [1.0.3] — 2026-06-23
+
+### Fixed
+
+- Messages for the same chat are now processed one at a time (per-session/chat lock), closing a race
+  where two near-simultaneous messages could read the same flow state and produce lost or duplicated
+  navigation (e.g. a double greeting or a resurrected leaf). The bounded invalid-path re-process runs
+  inside the lock to avoid self-deadlock, and the lock map self-evicts when a chat's queue drains.
+- If a config edit leaves an in-flight user parked on a node that no longer has options, the flow now
+  ends cleanly instead of replying "Invalid option" on every message until the 15-minute expiry.
+
 ## [1.0.2] — 2026-06-23
 
 ### Added

chat-flow/README.md

@@ -13,7 +13,7 @@
 | Field | Value |
 | ----- | ----- |
 | **Identifier** | `chat-flow` |
-| **Version** | 1.0.2 |
+| **Version** | 1.0.3 |
 | **Released** | 2026-06-23 |
 | **Status** | stable |
 | **Author** | Yudhi Armyndharis |

chat-flow/flow-engine.test.ts

@@ -132,6 +132,28 @@ test('invalid stored path with a non-trigger input resets and stops (bounded)',
   assert.equal(storage.has(key), false);
 });
 
+test('concurrent messages for the same chat are serialized (greeting sent once)', async () => {
+  const { ctx, replies } = makeCtx();
+  // Two messages racing on a fresh chat: without serialization both read "no state" and both greet.
+  await Promise.all([
+    FlowEngine.processMessage(ctx, xyz, 'xyz', 'user1', 'hello', 'm1'),
+    FlowEngine.processMessage(ctx, xyz, 'xyz', 'user1', 'hello', 'm2'),
+  ]);
+  const greetings = replies.filter(r => r.text === xyz.greeting).length;
+  assert.equal(greetings, 1, 'the greeting must be sent once; a load/save race would send it twice');
+});
+
+test('a stored path landing on a now-leaf node ends the flow instead of looping invalid-option', async () => {
+  const { ctx, storage, replies } = makeCtx();
+  // Config changed under the user: they are parked at option "1", which is a leaf with no sub-options.
+  // The old behaviour re-saved state and replied "Invalid option" forever; it must now end cleanly.
+  storage.set('state__xyz__user1', { path: ['1'], lastActive: Date.now() });
+  const r = await FlowEngine.processMessage(ctx, xyz, 'xyz', 'user1', 'anything', 'm1');
+  assert.equal(r, false);
+  assert.equal(storage.has('state__xyz__user1'), false);
+  assert.equal(replies.length, 0);
+});
+
 test('a prototype-property name as input is an invalid option, never a false match', async () => {
   // `abc.options` is a plain object, so a bare `options["constructor"]` would resolve to the inherited
   // Object constructor (truthy) and reply with `nextNode.text === undefined`. Object.hasOwn prevents that.

chat-flow/flow-engine.ts

@@ -19,10 +19,13 @@ export interface UserState {
 export class FlowEngine {
   private static readonly TIMEOUT_MS = 15 * 60 * 1000; // 15 minutes state expiration
   private static readonly MAX_REPROCESS = 1; // bound the invalid-path reset (no unbounded recursion)
+  /** Per (session,chat) promise chain serializing the state read→write. Self-evicts when drained. */
+  private static readonly locks = new Map<string, Promise<unknown>>();
 
   /**
    * Process an incoming message and send auto-replies according to `flow` (the resolved per-session
-   * config). Returns true if a reply was sent, false otherwise.
+   * config). Returns true if a reply was sent, false otherwise. Serializes per (session, chat) so
+   * concurrent messages for the same chat can't interleave the state read→write.
    */
   public static async processMessage(
     context: PluginContext,
@@ -31,6 +34,29 @@ export class FlowEngine {
     chatId: string,
     messageBody: string,
     messageId: string,
+  ): Promise<boolean> {
+    // The bounded re-process inside the body calls processLocked directly (bypassing this lock) so a
+    // chat never waits on its own still-pending chain entry (self-deadlock). Store a settled tail so a
+    // rejection can't wedge the chain, and evict the key once the chain drains.
+    const lockKey = `${sessionId}__${chatId}`;
+    const prev = this.locks.get(lockKey) ?? Promise.resolve();
+    const run = prev.then(() => this.processLocked(context, flow, sessionId, chatId, messageBody, messageId, 0));
+    const tail = run.catch(() => {});
+    this.locks.set(lockKey, tail);
+    try {
+      return await run;
+    } finally {
+      if (this.locks.get(lockKey) === tail) this.locks.delete(lockKey);
+    }
+  }
+
+  private static async processLocked(
+    context: PluginContext,
+    flow: SessionFlow,
+    sessionId: string,
+    chatId: string,
+    messageBody: string,
+    messageId: string,
     depth = 0,
   ): Promise<boolean> {
     context.logger.debug('[FlowEngine] Processing message', { sessionId, chatId, body: messageBody });
@@ -86,7 +112,9 @@ export class FlowEngine {
           context.logger.debug('[FlowEngine] Max reprocess depth reached; not recursing.', { depth });
           return false;
         }
-        return this.processMessage(context, flow, sessionId, chatId, messageBody, messageId, depth + 1);
+        // Recurse on the locked body, NOT processMessage — re-entering the lock would deadlock on this
+        // chat's own still-pending chain entry.
+        return this.processLocked(context, flow, sessionId, chatId, messageBody, messageId, depth + 1);
       }
     }
 
@@ -114,6 +142,12 @@ export class FlowEngine {
         await context.storage.delete(stateKey);
       }
       return true;
+    } else if (!currentNode.options || Object.keys(currentNode.options).length === 0) {
+      // The resolved node is a leaf with no way forward (config changed under the user). End the flow
+      // instead of looping "Invalid option" forever; the next trigger starts cleanly.
+      context.logger.debug('[FlowEngine] Resolved node is a dead leaf (config changed). Ending flow.');
+      await context.storage.delete(stateKey);
+      return false;
     } else {
       context.logger.debug('[FlowEngine] Input did not match any options. Replying with fallback.');
       const invalidMsg = `Invalid option. Please choose one of the available options:\n\n${currentNode.text}`;

chat-flow/manifest.json

@@ -1,7 +1,7 @@
 {
   "id": "chat-flow",
   "name": "Chat Flow",
-  "version": "1.0.2",
+  "version": "1.0.3",
   "type": "extension",
   "main": "dist/index.js",
   "description": "Interactive, stateful auto-reply: a trigger word starts a greeting + numbered menu, replies traverse a configurable menu tree, and per-chat state expires after 15 minutes.",

plugins.json

@@ -197,7 +197,7 @@
   {
     "id": "chat-flow",
     "name": "Chat Flow",
-    "version": "1.0.2",
+    "version": "1.0.3",
     "type": "extension",
     "status": "stable",
     "description": "Interactive, stateful auto-reply: a trigger word starts a greeting + numbered menu, replies traverse a configurable menu tree, and per-chat state expires after 15 minutes.",
@@ -218,7 +218,7 @@
     "repoPath": "chat-flow",
     "repoUrl": "https://github.com/rmyndharis/OpenWA-plugins",
     "homepage": "https://github.com/rmyndharis/OpenWA-plugins/tree/main/chat-flow",
-    "download": "https://github.com/rmyndharis/OpenWA-plugins/releases/download/chat-flow-v1.0.2/chat-flow.zip",
+    "download": "https://github.com/rmyndharis/OpenWA-plugins/releases/download/chat-flow-v1.0.3/chat-flow.zip",
     "i18n": {
       "es": {
         "name": "Flujo de Chat",