diff --git a/.changeset/28230-batch-recall-past-first-page.md b/.changeset/28230-batch-recall-past-first-page.md
new file mode 100644
index 00000000000..084a30ca11b
--- /dev/null
+++ b/.changeset/28230-batch-recall-past-first-page.md
@@ -0,0 +1,14 @@
+---
+'@objectstack/plugin-approvals': patch
+'@objectstack/service-automation': patch
+---
+
+fix(plugin-approvals): a revise-window recall on a batch item finds its batch run past the first 1000 parked runs
+
+Clause-②: no (a recall on a store with more than 1000 parked runs now wakes its batch, as it already does below that; no schema, accept set, error code or result field changes)
+Metadata-protocol: none
+
+The recall found the batch above the item it cancels in `listSuspendedRunsDurable()`. On a database-backed store that listing reads one page of 1000 `paused` rows, and a process adds only the runs it parked itself. So when another replica (or this process before a restart) parked the batch and more than 1000 runs were parked ahead of it, the recall did not find the batch. The batch stayed parked on the cancelled item, and the recall said nothing.
+
+- **Now.** Before it cancels the item, the recall asks the engine for the batch through the item's own link to it. The new `AutomationEngine.suspendedMapParentOf(itemRunId)` reads the item's suspension, then its parent's, by id. It answers the parent only while the parent is parked on `map:<itemRunId>` at the `map` node that launched the item. The read comes before `cancelRun` because the cancel removes the item's suspension, and the link with it.
+- **Unchanged.** Below 1000 parked runs the recall wakes the batch as before. A recall with no batch above it, and a run parked on the item at a node other than the built-in `map`, are left alone as before. A failed lookup is reported in `resumeError` after the cancel, as before.
diff --git a/packages/plugins/plugin-approvals/src/approval-service.ts b/packages/plugins/plugin-approvals/src/approval-service.ts
index dc8e31a3319..0cdf9f83890 100644
--- a/packages/plugins/plugin-approvals/src/approval-service.ts
+++ b/packages/plugins/plugin-approvals/src/approval-service.ts
@@ -288,6 +288,20 @@ export interface ApprovalResumeSurface {
    * pause it cannot see is a pause this verb will not act on.
    */
   listSuspendedRunsDurable?(): Promise<Array<{ runId: string; flowName: string; nodeId: string; correlation?: string }>>;
+  /**
+   * [#28230] The run parked at a `map` node on the item run `itemRunId`, or
+   * `undefined` when there is none. Read by
+   * {@link ApprovalService.recall}'s revise-window branch BEFORE it cancels
+   * the item, so it can wake the batch above it afterwards: the engine
+   * answers from the item's own `$parentRunId` linkage with keyed reads,
+   * where {@link listSuspendedRunsDurable} reads one capped page of a
+   * DB-backed store and can miss the batch.
+   *
+   * ⚠️ Declares the method `AutomationEngine` implements publicly
+   * (`suspendedMapParentOf`). Read-only; rejects when the store cannot be
+   * read. Optional: without it no batch is woken, as before #28183.
+   */
+  suspendedMapParentOf?(itemRunId: string): Promise<BatchParent | undefined>;
   /**
    * [#15358] Whether the consumed suspension behind a `failed` run SURVIVES —
    * i.e. whether the engine's `restoreConsumedSuspension` would re-arm it.
@@ -352,6 +366,12 @@ export interface ApprovalResumeSurface {
   takeSubflowParentStrand?(childRunId: string): { runId: string; repairable: boolean; error: string } | undefined;
 }
 
+/** A run parked at a `map` node on one item, as {@link ApprovalResumeSurface.suspendedMapParentOf} answers it. */
+type BatchParent = { runId: string; flowName: string; nodeId: string; correlation?: string };
+
+/** What the revise-window recall read about the batch above its item before cancelling it. */
+type BatchLookup = { parent?: BatchParent } | { lookupFailure: unknown };
+
 /** What {@link ApprovalResumeSurface.inspectConsumedSuspension} answers. */
 type ConsumedSuspensionVerdict =
   Awaited<ReturnType<NonNullable<ApprovalResumeSurface['inspectConsumedSuspension']>>>;
@@ -5277,6 +5297,9 @@ export class ApprovalService implements IApprovalService {
       if (runId) {
         resumeError = this.missingRunCapability(runId, requestId, 'the recall', 'cancelRun');
         if (!resumeError) {
+          // [#28230] Read the batch above the item while the item's
+          // suspension still carries the linkage: `cancelRun` removes it.
+          const batch = await this.batchAboveItem(runId);
           let cancelled: unknown;
           try {
             cancelled = await this.automation!.cancelRun!(runId, `approval request ${requestId} recalled during revision`);
@@ -5288,7 +5311,7 @@ export class ApprovalService implements IApprovalService {
           }
           // [#28183] `cancelRun` never touches a parent, so a batch whose
           // `map` item this was would stay parked with nothing to wake it.
-          if (cancelled === true) ({ resumeError, resumeFailure } = await this.resumeBatchOfRecalledItem(runId, requestId));
+          if (cancelled === true) ({ resumeError, resumeFailure } = await this.resumeBatchOfRecalledItem(runId, requestId, batch));
         }
       }
     } else if (runId) {
@@ -5370,6 +5393,29 @@ export class ApprovalService implements IApprovalService {
     };
   }
 
+  /**
+   * [#28230] The batch run parked on the revise-window item `itemRunId`, read
+   * before the recall cancels the item — {@link ApprovalResumeSurface.suspendedMapParentOf}
+   * reads the item's own `$parentRunId`, which `cancelRun` removes with the
+   * item's suspension. Never throws: a failed read is carried to
+   * {@link resumeBatchOfRecalledItem}, which tells it once the cancel is done,
+   * as it always told a failed lookup.
+   *
+   * It used to be the entry of `listSuspendedRunsDurable()` parked on
+   * `map:<itemRunId>`, read after the cancel. That listing reads one capped
+   * page of a DB-backed store, and a process adds only what it parked itself,
+   * so a batch another process parked behind more than 1000 runs was not
+   * found and stayed parked with nothing said.
+   */
+  private async batchAboveItem(itemRunId: string): Promise<BatchLookup> {
+    if (typeof this.automation?.suspendedMapParentOf !== 'function') return {};
+    try {
+      return { parent: await this.automation.suspendedMapParentOf(itemRunId) };
+    } catch (err) {
+      return { lookupFailure: err };
+    }
+  }
+
   /**
    * [#28183] Wake the batch above a run a revise-window recall just cancelled.
    *
@@ -5386,21 +5432,22 @@ export class ApprovalService implements IApprovalService {
    * Only a run parked at the built-in `map` node on that item is woken: the
    * engine records the marker for no other node, and re-running any other
    * node would advance it past the item with nothing recorded. Whatever is
-   * missing to prove that — no `listSuspendedRunsDurable`, no `getFlow`, no
+   * missing to prove that — no `suspendedMapParentOf`, no `getFlow`, no
    * such parent — leaves things exactly as they were before, which for a
-   * recall with no batch above it is the whole answer.
+   * recall with no batch above it is the whole answer. The parent is the one
+   * {@link batchAboveItem} read before the cancel.
    */
   private async resumeBatchOfRecalledItem(
     itemRunId: string,
     requestId: string,
+    batch: BatchLookup,
   ): Promise<{ resumeError?: string; resumeFailure?: ResumeFailureReport }> {
-    if (typeof this.automation?.listSuspendedRunsDurable !== 'function') return {};
-    if (typeof this.automation.getFlow !== 'function') return {};
     let batchRunId: string | undefined;
     try {
-      const parent = (await this.automation.listSuspendedRunsDurable())
-        .find(r => r.correlation === `map:${itemRunId}`);
+      if ('lookupFailure' in batch) throw batch.lookupFailure;
+      const parent = batch.parent;
       if (!parent) return {};
+      if (typeof this.automation?.getFlow !== 'function') return {};
       const flow: any = await this.automation.getFlow(parent.flowName);
       const node = Array.isArray(flow?.nodes) ? flow.nodes.find((n: any) => n?.id === parent.nodeId) : undefined;
       if (node?.type !== 'map') return {};
diff --git a/packages/plugins/plugin-approvals/src/batch-item-recall-resumes-batch.test.ts b/packages/plugins/plugin-approvals/src/batch-item-recall-resumes-batch.test.ts
index 630cc01d1fa..0cd079bc808 100644
--- a/packages/plugins/plugin-approvals/src/batch-item-recall-resumes-batch.test.ts
+++ b/packages/plugins/plugin-approvals/src/batch-item-recall-resumes-batch.test.ts
@@ -38,7 +38,7 @@
  */
 
 import { describe, it, expect, beforeEach } from 'vitest';
-import { AutomationEngine, InMemorySuspendedRunStore, installBuiltinNodes } from '@objectstack/service-automation';
+import { AutomationEngine, InMemorySuspendedRunStore, ObjectStoreSuspendedRunStore, installBuiltinNodes } from '@objectstack/service-automation';
 import { APPROVAL_REVISE_NODE_TYPE, defineActionDescriptor } from '@objectstack/spec/automation';
 import { assertEngineDeleteDispatch, assertEngineUpdateDispatch } from '@objectstack/objectql';
 import { ApprovalService } from './approval-service.js';
@@ -158,6 +158,47 @@ const RELEASE_SIGNOFF = {
   ],
 };
 
+/** Ticking clock: rounds of a revise loop must not share created_at. */
+function tickingClock() {
+  let t = Date.parse('2026-10-30T00:00:00Z');
+  return { now: () => new Date((t += 1000)) };
+}
+
+/**
+ * One process: an automation engine over `store` and an approval service over
+ * `fake`. Two of these over one durable store and one `fake` are two replicas
+ * (or one process before and after a restart).
+ */
+function wireReplica(
+  fake: ReturnType<typeof makeFakeEngine>,
+  store: InMemorySuspendedRunStore | ObjectStoreSuspendedRunStore,
+  clock: { now: () => Date },
+  onPeek: (seen: { results: unknown; count: unknown }) => void,
+) {
+  const service = new ApprovalService({ engine: fake as any, logger: silentLogger(), clock });
+  const automation = new AutomationEngine(silentLogger(), store);
+  installBuiltinNodes(automation, { logger: silentLogger(), getService() { throw new Error('none'); } } as any);
+  registerApprovalNode(automation, service, silentLogger());
+  automation.registerNodeExecutor({
+    type: 'decided',
+    async execute(node: any, variables: Map<string, unknown>) {
+      variables.set('decision', node.config.decision);
+      return { success: true };
+    },
+  } as never);
+  automation.registerNodeExecutor({
+    type: 'peek',
+    async execute(_node: any, variables: Map<string, unknown>) {
+      onPeek({ results: variables.get('signoffResults'), count: variables.get('signoffs.count') });
+      return { success: true };
+    },
+  } as never);
+  automation.registerFlow('task_signoff', TASK_SIGNOFF as never);
+  automation.registerFlow('release_signoff', RELEASE_SIGNOFF as never);
+  service.attachAutomation(automation);
+  return { automation, service };
+}
+
 describe('a revise-window recall on a batch item resumes the batch (#28183)', () => {
   let automation: AutomationEngine;
   let service: ApprovalService;
@@ -167,29 +208,7 @@ describe('a revise-window recall on a batch item resumes the batch (#28183)', ()
   beforeEach(() => {
     seen = undefined;
     fake = makeFakeEngine();
-    // Ticking clock: rounds of a revise loop must not share created_at.
-    let t = Date.parse('2026-10-30T00:00:00Z');
-    service = new ApprovalService({ engine: fake as any, logger: silentLogger(), clock: { now: () => new Date((t += 1000)) } });
-    automation = new AutomationEngine(silentLogger(), new InMemorySuspendedRunStore());
-    installBuiltinNodes(automation, { logger: silentLogger(), getService() { throw new Error('none'); } } as any);
-    registerApprovalNode(automation, service, silentLogger());
-    automation.registerNodeExecutor({
-      type: 'decided',
-      async execute(node: any, variables: Map<string, unknown>) {
-        variables.set('decision', node.config.decision);
-        return { success: true };
-      },
-    } as never);
-    automation.registerNodeExecutor({
-      type: 'peek',
-      async execute(_node: any, variables: Map<string, unknown>) {
-        seen = { results: variables.get('signoffResults'), count: variables.get('signoffs.count') };
-        return { success: true };
-      },
-    } as never);
-    automation.registerFlow('task_signoff', TASK_SIGNOFF as never);
-    automation.registerFlow('release_signoff', RELEASE_SIGNOFF as never);
-    service.attachAutomation(automation);
+    ({ automation, service } = wireReplica(fake, new InMemorySuspendedRunStore(), tickingClock(), (s) => { seen = s; }));
   });
 
   const launchBatch = async (ids: string[]) => {
@@ -328,3 +347,161 @@ describe('a revise-window recall on a batch item resumes the batch (#28183)', ()
     expect(parked()).toEqual([{ runId: fan.runId, nodeId: 'fan', correlation: `map:${req.flow_run_id}` }]);
   });
 });
+
+/**
+ * #28230: the recall finds the batch above the item it cancels past the first
+ * page of parked runs.
+ *
+ * A DB-backed store lists one 1000-row page of `paused` rows in driver order,
+ * and a process lists what it parked itself beside that page. So the batch is
+ * missing from the listing exactly when it was parked by ANOTHER process (a
+ * replica, or this one before a restart) and more than 1000 runs are parked
+ * ahead of it. Both replicas below share one store and one table set; the
+ * batch is launched and sent back on `a` and recalled on `b`.
+ */
+describe('a revise-window recall finds its batch past the first page of parked runs (#28230)', () => {
+  let fake: ReturnType<typeof makeFakeEngine>;
+  let a: ReturnType<typeof wireReplica>;
+  let b: ReturnType<typeof wireReplica>;
+  let seen: { results: unknown; count: unknown } | undefined;
+
+  /** `count` parked runs of some other flow, stored ahead of everything this test parks. */
+  const parkOthers = (count: number) => {
+    const rows = fake.tables.get('sys_automation_run') ?? (fake.tables.set('sys_automation_run', []), fake.tables.get('sys_automation_run')!);
+    for (let i = 0; i < count; i++) {
+      rows.push({ id: `other-${String(i).padStart(4, '0')}`, flow_name: 'other_flow', node_id: 'wait', status: 'paused' });
+    }
+  };
+
+  beforeEach(() => {
+    seen = undefined;
+    fake = makeFakeEngine();
+    const store = new ObjectStoreSuspendedRunStore(fake as any, silentLogger());
+    const clock = tickingClock();
+    a = wireReplica(fake, store, clock, (s) => { seen = s; });
+    b = wireReplica(fake, store, clock, (s) => { seen = s; });
+  });
+
+  const requestFor = async (recordId: string, status: string) =>
+    (await fake.find('sys_approval_request', { where: { record_id: recordId, status } }))[0];
+  const storedRun = async (runId: string) =>
+    (await fake.find('sys_automation_run', { where: { id: runId } }))[0];
+
+  /** Launch the batch on `a` and send its first item back into the revise window. */
+  const batchInReviseWindow = async () => {
+    const paused = await a.automation.execute('release_signoff', {
+      userId: 'submitter', params: { items: [{ id: 't1', title: 'Task t1' }, { id: 't2', title: 'Task t2' }] },
+    } as never);
+    expect(paused.status, 'the batch parks on its first item').toBe('paused');
+    const first = await requestFor('t1', 'pending');
+    await a.service.sendBack(first.id, { actorId: 'u1', comment: 'rework' }, asUser('u1'));
+    return { batchRunId: paused.runId!, first };
+  };
+
+  it('the batch past the first page resumes, and the item keeps its slot as the cancelled marker', async () => {
+    parkOthers(1000);
+    const { batchRunId, first } = await batchInReviseWindow();
+    expect((await b.automation.listSuspendedRunsDurable()).some(r => r.runId === batchRunId),
+      'premise: the batch is not on the listing replica b reads').toBe(false);
+
+    const out = await b.service.recall(first.id, { actorId: 'submitter' }, asUser('submitter'));
+
+    expect(out.request.status).toBe('recalled');
+    expect(out.resumed).toBe(false);
+    expect(out.resumeError).toBeUndefined();
+    expect((await b.automation.getRun(first.flow_run_id))?.status).toBe('cancelled');
+    const second = await requestFor('t2', 'pending');
+    expect(second, 'the batch opened the next item').toBeDefined();
+    expect(await storedRun(batchRunId)).toMatchObject({ node_id: 'signoffs', correlation: `map:${second.flow_run_id}` });
+
+    await b.service.decide(second.id, { decision: 'approve', actorId: 'u1' }, SYSTEM_CTX);
+
+    expect(seen).toEqual({ results: [{ $status: 'cancelled', $itemRunId: first.flow_run_id }, { decision: 'approved' }], count: 1 });
+    expect((await b.automation.getRun(batchRunId))?.status).toBe('completed');
+  });
+
+  it('control: a batch within the first page behaves as before', async () => {
+    parkOthers(10);
+    const { batchRunId, first } = await batchInReviseWindow();
+    expect((await b.automation.listSuspendedRunsDurable()).some(r => r.runId === batchRunId),
+      'premise: the batch is on the listing replica b reads').toBe(true);
+
+    const out = await b.service.recall(first.id, { actorId: 'submitter' }, asUser('submitter'));
+
+    expect(out).toMatchObject({ resumed: false });
+    expect(out.resumeError).toBeUndefined();
+    const second = await requestFor('t2', 'pending');
+    expect(second, 'the batch opened the next item').toBeDefined();
+    await b.service.decide(second.id, { decision: 'approve', actorId: 'u1' }, SYSTEM_CTX);
+    expect(seen).toEqual({ results: [{ $status: 'cancelled', $itemRunId: first.flow_run_id }, { decision: 'approved' }], count: 1 });
+  });
+
+  it('control: a recall with no batch above it past the first page behaves as before, and the batch beside it is untouched', async () => {
+    parkOthers(1000);
+    const { batchRunId, first } = await batchInReviseWindow();
+    const lone = await a.automation.execute('task_signoff', {
+      object: 'rel_task', record: { id: 'solo' }, userId: 'submitter',
+    } as never);
+    const req = await requestFor('solo', 'pending');
+    await a.service.sendBack(req.id, { actorId: 'u1' }, asUser('u1'));
+
+    const out = await b.service.recall(req.id, { actorId: 'submitter' }, asUser('submitter'));
+
+    expect(out).toMatchObject({ runId: lone.runId, resumed: false });
+    expect(out.resumeError).toBeUndefined();
+    expect('resumeFailure' in out).toBe(false);
+    expect((await b.automation.getRun(lone.runId!))?.status).toBe('cancelled');
+    expect(await storedRun(batchRunId)).toMatchObject({ node_id: 'signoffs', correlation: `map:${first.flow_run_id}` });
+    expect(await storedRun(first.flow_run_id)).toMatchObject({ node_id: 'wait_revision', status: 'paused' });
+    expect(await requestFor('t2', 'pending')).toBeUndefined();
+  });
+
+  it('control: a parent linked to the item but parked at a node other than the built-in map is left alone', async () => {
+    // A home-grown fan-out that launches its child with the same linkage the
+    // `map` node writes, so the engine answers it as the item's parent; it is
+    // the recall's own `map` check that leaves it parked.
+    let entries = 0;
+    for (const replica of [a, b]) {
+      replica.automation.registerNodeExecutor({
+        type: 'fan_out',
+        descriptor: defineActionDescriptor({
+          type: 'fan_out', version: '1.0.0', name: 'Fan out', supportsPause: true, resumeAuthority: 'any',
+        }),
+        async execute(node: any, variables: Map<string, unknown>, context: any) {
+          entries++;
+          const child = await replica.automation.execute('task_signoff', {
+            ...context, object: 'rel_task', record: { id: `f${entries}` },
+            $parentRunId: String(variables.get('$runId')), $parentMapNode: node.id,
+          } as never);
+          return { success: true, suspend: true, correlation: `map:${child.runId}` };
+        },
+      } as never);
+      replica.automation.registerFlow('fan_signoff', {
+        name: 'fan_signoff', label: 'Fan Sign-off', type: 'autolaunched',
+        nodes: [
+          { id: 'start', type: 'start', label: 'Start' },
+          { id: 'fan', type: 'fan_out', label: 'Fan out' },
+          { id: 'end', type: 'end', label: 'End' },
+        ],
+        edges: [
+          { id: 'f1', source: 'start', target: 'fan' },
+          { id: 'f2', source: 'fan', target: 'end' },
+        ],
+      } as never);
+    }
+    parkOthers(1000);
+    const fan = await a.automation.execute('fan_signoff', { userId: 'submitter' } as never);
+    const req = await requestFor('f1', 'pending');
+    await a.service.sendBack(req.id, { actorId: 'u1' }, asUser('u1'));
+    expect(await b.automation.suspendedMapParentOf(req.flow_run_id),
+      'premise: the engine answers the fan-out as the item\'s parent').toMatchObject({ runId: fan.runId, nodeId: 'fan' });
+
+    const out = await b.service.recall(req.id, { actorId: 'submitter' }, asUser('submitter'));
+
+    expect(out).toMatchObject({ resumed: false });
+    expect(out.resumeError).toBeUndefined();
+    expect(entries).toBe(1);
+    expect(await requestFor('f2', 'pending')).toBeUndefined();
+    expect(await storedRun(fan.runId!)).toMatchObject({ node_id: 'fan', correlation: `map:${req.flow_run_id}`, status: 'paused' });
+  });
+});
diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts
index 4c800e9b65b..a63f1be5343 100644
--- a/packages/services/service-automation/src/engine.ts
+++ b/packages/services/service-automation/src/engine.ts
@@ -6874,6 +6874,37 @@ export class AutomationEngine implements IAutomationService {
         return 'cause' in target ? target.cause : undefined;
     }
 
+    /**
+     * [#28230] The run parked at a `map` node on `itemRunId`, read through the
+     * item's own linkage: the `$parentRunId` / `$parentMapNode` the `map` node
+     * launched it with, then the parent's suspension by that id. Two keyed
+     * reads, never a listing — {@link listSuspendedRunsDurable} reads one
+     * capped page of a DB-backed store, so past 1000 parked runs a parent
+     * another process parked may not be on it.
+     *
+     * Answers the parent only while it is parked on `map:<itemRunId>` at the
+     * node the item names, and `undefined` otherwise: no suspension under
+     * `itemRunId`, an item no `map` launched, a parent that is gone or has
+     * moved on. Read it BEFORE {@link cancelRun}, which removes the item's
+     * suspension and with it the linkage.
+     *
+     * Read-only. THROWS when the store cannot be read, as
+     * {@link hasSuspendedRun} does: an outage means "unknown", never "no
+     * parent".
+     */
+    async suspendedMapParentOf(
+        itemRunId: string,
+    ): Promise<{ runId: string; flowName: string; nodeId: string; correlation?: string } | undefined> {
+        const item = await this.loadSuspendedRunStrict(itemRunId);
+        const ctx = item?.context as Record<string, unknown> | undefined;
+        const parentRunId = ctx?.$parentRunId;
+        const mapNode = ctx?.$parentMapNode;
+        if (typeof parentRunId !== 'string' || !parentRunId) return undefined;
+        const parent = await this.loadSuspendedRunStrict(parentRunId);
+        if (!parent || parent.nodeId !== mapNode || parent.correlation !== `map:${itemRunId}`) return undefined;
+        return { runId: parent.runId, flowName: parent.flowName, nodeId: parent.nodeId, correlation: parent.correlation };
+    }
+
     /**
      * Credit a completed child run's totals to the parent step waiting on it
      * (#4354).
diff --git a/packages/services/service-automation/src/suspended-map-parent-of.test.ts b/packages/services/service-automation/src/suspended-map-parent-of.test.ts
new file mode 100644
index 00000000000..8165de87d06
--- /dev/null
+++ b/packages/services/service-automation/src/suspended-map-parent-of.test.ts
@@ -0,0 +1,152 @@
+// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
+
+/**
+ * `suspendedMapParentOf` answers the run parked at a `map` node on an item run
+ * from the item's own linkage, with keyed reads (#28230).
+ *
+ * The approvals recall reads it before it cancels a batch item, so it can wake
+ * the batch afterwards. It used to find the batch in `listSuspendedRunsDurable`,
+ * which reads one capped page of a DB-backed store; the end-to-end pin past
+ * that page is `batch-item-recall-resumes-batch.test.ts` (plugin-approvals).
+ *
+ * The pins: a `map` item answers its parent, also on an engine that did not
+ * park it; a subflow child, a subflow nested in a `map` item (which inherits
+ * the item's `$parentMapNode`), a top-level run and an unknown id answer
+ * `undefined`; and so does an item whose parent is no longer parked on it.
+ */
+
+import { describe, it, expect, beforeEach } from 'vitest';
+import { defineActionDescriptor } from '@objectstack/spec/automation';
+import { AutomationEngine } from './engine.js';
+import { InMemorySuspendedRunStore } from './suspended-run-store.js';
+import { registerMapNode } from './builtin/map-node.js';
+import { registerSubflowNode } from './builtin/subflow-node.js';
+
+function silentLogger(): any {
+  return { info() {}, warn() {}, error() {}, debug() {}, child() { return silentLogger(); } };
+}
+function pluginCtx(): any {
+  return { logger: silentLogger(), getService() { throw new Error('none'); } };
+}
+
+/** start → `open_pause` (raw-resumable) → end. */
+const ITEM_FLOW = {
+  name: 'one_item', label: 'one_item', type: 'autolaunched',
+  nodes: [
+    { id: 'start', type: 'start', label: 'Start' },
+    { id: 'pause', type: 'open_pause', label: 'Pause' },
+    { id: 'end', type: 'end', label: 'End' },
+  ],
+  edges: [
+    { id: 'e1', source: 'start', target: 'pause' },
+    { id: 'e2', source: 'pause', target: 'end' },
+  ],
+};
+
+/** start → `call` (a subflow into `one_item`) → end. */
+const WRAP_FLOW = {
+  name: 'wrap_item', label: 'wrap_item', type: 'autolaunched',
+  nodes: [
+    { id: 'ws', type: 'start', label: 'Start' },
+    { id: 'call', type: 'subflow', label: 'Call', config: { flowName: 'one_item' } },
+    { id: 'we', type: 'end', label: 'End' },
+  ],
+  edges: [
+    { id: 'w1', source: 'ws', target: 'call' },
+    { id: 'w2', source: 'call', target: 'we' },
+  ],
+};
+
+/** start → `signoffs` (a map over `items`, each item running `itemFlow`) → end. */
+const batchFlow = (name: string, itemFlow: string) => ({
+  name, label: name, type: 'autolaunched',
+  variables: [{ name: 'items', type: 'list', isInput: true }],
+  nodes: [
+    { id: 'hs', type: 'start', label: 'Start' },
+    { id: 'signoffs', type: 'map', label: 'Each', config: { collection: { dialect: 'cel', source: 'items' }, flowName: itemFlow } },
+    { id: 'he', type: 'end', label: 'End' },
+  ],
+  edges: [
+    { id: 'h1', source: 'hs', target: 'signoffs' },
+    { id: 'h2', source: 'signoffs', target: 'he' },
+  ],
+});
+
+function makeEngine(store: InMemorySuspendedRunStore): AutomationEngine {
+  const engine = new AutomationEngine(silentLogger(), store);
+  engine.registerNodeExecutor({
+    type: 'open_pause',
+    descriptor: defineActionDescriptor({
+      type: 'open_pause', version: '1.0.0', name: 'Open Pause', supportsPause: true, resumeAuthority: 'any',
+    }),
+    async execute() { return { success: true, suspend: true }; },
+  });
+  registerMapNode(engine, pluginCtx());
+  registerSubflowNode(engine, pluginCtx());
+  engine.registerFlow('one_item', ITEM_FLOW as never);
+  engine.registerFlow('wrap_item', WRAP_FLOW as never);
+  engine.registerFlow('batch_flow', batchFlow('batch_flow', 'one_item') as never);
+  engine.registerFlow('wrapped_batch', batchFlow('wrapped_batch', 'wrap_item') as never);
+  return engine;
+}
+
+describe('suspendedMapParentOf answers the batch above a map item from its linkage (#28230)', () => {
+  let store: InMemorySuspendedRunStore;
+  let engine: AutomationEngine;
+
+  beforeEach(() => {
+    store = new InMemorySuspendedRunStore();
+    engine = makeEngine(store);
+  });
+
+  const parked = (flowName: string) => engine.listSuspendedRuns().filter(r => r.flowName === flowName);
+  const items = [{ id: 't1' }, { id: 't2' }];
+
+  it('a map item answers the run parked on it at the map node, on an engine that did not park it too', async () => {
+    const batch = await engine.execute('batch_flow', { params: { items } });
+    const [item] = parked('one_item');
+    const parent = { runId: batch.runId, flowName: 'batch_flow', nodeId: 'signoffs', correlation: `map:${item!.runId}` };
+
+    expect(await engine.suspendedMapParentOf(item!.runId)).toEqual(parent);
+    // Another process over the same store: nothing in its memory, keyed reads only.
+    expect(await makeEngine(store).suspendedMapParentOf(item!.runId)).toEqual(parent);
+  });
+
+  it('a subflow child answers undefined: its parent waits at a subflow node, not a map', async () => {
+    const wrap = await engine.execute('wrap_item', {});
+    const [child] = parked('one_item');
+    expect(parked('wrap_item')).toEqual([expect.objectContaining({ runId: wrap.runId, correlation: `subflow:${child!.runId}` })]);
+
+    expect(await engine.suspendedMapParentOf(child!.runId)).toBeUndefined();
+  });
+
+  it('a subflow nested in a map item answers undefined, though it inherits the item\'s $parentMapNode', async () => {
+    await engine.execute('wrapped_batch', { params: { items } });
+    const [inner] = parked('one_item');
+    const [item] = parked('wrap_item');
+    expect((await store.load(inner!.runId))?.context).toMatchObject({ $parentRunId: item!.runId, $parentMapNode: 'signoffs' });
+
+    expect(await engine.suspendedMapParentOf(inner!.runId)).toBeUndefined();
+    // The item itself still answers its batch.
+    expect(await engine.suspendedMapParentOf(item!.runId)).toMatchObject({ nodeId: 'signoffs', correlation: `map:${item!.runId}` });
+  });
+
+  it('an item whose parent is no longer parked on it answers undefined', async () => {
+    const batch = await engine.execute('batch_flow', { params: { items } });
+    const [item1] = parked('one_item');
+    // Copy item 1's suspension aside, move the batch on to item 2, then put the copy back.
+    const copy = await store.load(item1!.runId);
+    await engine.cancelRun(item1!.runId, 'operator');
+    expect((await engine.resume(batch.runId!, {})).status).toBe('paused');
+    await store.save(copy!);
+
+    expect(await engine.suspendedMapParentOf(item1!.runId)).toBeUndefined();
+  });
+
+  it('a top-level run and an unknown id answer undefined', async () => {
+    const batch = await engine.execute('batch_flow', { params: { items } });
+
+    expect(await engine.suspendedMapParentOf(batch.runId!)).toBeUndefined();
+    expect(await engine.suspendedMapParentOf('no-such-run')).toBeUndefined();
+  });
+});